From: Harman Singh <harmasin@cisco.com>
Date: Tue, 21 Oct 2014 21:50:04 +0000 (-0700)
Subject: Bug 2086: Adding normalized node  stream reader and writer.
X-Git-Tag: release/lithium~964^2
X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=commitdiff_plain;h=d0825a4c573ca648bf509916864645d2ad65dfda

Bug 2086: Adding normalized node  stream reader and writer.

Normalized node and its children are written in recursive manner. Besides the leafNode, leafSetEntryNode and anyxmlNode, all other nodes must call endNode method
while writing object to data stream. This helps to recreate the object.

Recursion  runs in opposite direction while reading the object.

Updated code with review comments

Change-Id: Ibb822e11fc76f52b5be78596ed979c3b97d51de8
Signed-off-by: Harman Singh <harmasin@cisco.com>
---

diff --git a/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NodeTypes.java b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NodeTypes.java
new file mode 100644
index 0000000000..3ff6efbb05
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NodeTypes.java
@@ -0,0 +1,29 @@
+/*
+ *
+ *  Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ *  and is available at http://www.eclipse.org/legal/epl-v10.html
+ *
+ */
+
+package org.opendaylight.controller.cluster.datastore.node.utils.stream;
+
+public class NodeTypes {
+
+    public static final byte LEAF_NODE = 1;
+    public static final byte LEAF_SET = 2;
+    public static final byte LEAF_SET_ENTRY_NODE = 3;
+    public static final byte CONTAINER_NODE = 4;
+    public static final byte UNKEYED_LIST = 5;
+    public static final byte UNKEYED_LIST_ITEM = 6;
+    public static final byte MAP_NODE = 7;
+    public static final byte MAP_ENTRY_NODE = 8;
+    public static final byte ORDERED_MAP_NODE = 9;
+    public static final byte CHOICE_NODE = 10;
+    public static final byte AUGMENTATION_NODE = 11;
+    public static final byte ANY_XML_NODE = 12;
+    public static final byte END_NODE = 13;
+
+}
diff --git a/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeInputStreamReader.java b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeInputStreamReader.java
new file mode 100644
index 0000000000..bdc52bca68
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeInputStreamReader.java
@@ -0,0 +1,391 @@
+/*
+ *
+ *  Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ *  and is available at http://www.eclipse.org/legal/epl-v10.html
+ *
+ */
+
+package org.opendaylight.controller.cluster.datastore.node.utils.stream;
+
+import com.google.common.base.Preconditions;
+import org.opendaylight.controller.cluster.datastore.node.utils.QNameFactory;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.data.api.Node;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.AugmentationNode;
+import org.opendaylight.yangtools.yang.data.api.schema.ChoiceNode;
+import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
+import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
+import org.opendaylight.yangtools.yang.data.api.schema.LeafSetEntryNode;
+import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
+import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.api.schema.OrderedMapNode;
+import org.opendaylight.yangtools.yang.data.api.schema.UnkeyedListEntryNode;
+import org.opendaylight.yangtools.yang.data.api.schema.UnkeyedListNode;
+import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.CollectionNodeBuilder;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.DataContainerNodeAttrBuilder;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.DataContainerNodeBuilder;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.ListNodeBuilder;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.NormalizedNodeAttrBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * NormalizedNodeInputStreamReader reads the byte stream and constructs the normalized node including its children nodes.
+ * This process goes in recursive manner, where each NodeTypes object signifies the start of the object, except END_NODE.
+ * If a node can have children, then that node's end is calculated based on appearance of END_NODE.
+ *
+ */
+
+public class NormalizedNodeInputStreamReader implements NormalizedNodeStreamReader {
+
+    private DataInputStream reader;
+
+    private static final Logger LOG = LoggerFactory.getLogger(NormalizedNodeInputStreamReader.class);
+
+    private Map<Integer, String> codedStringMap = new HashMap<>();
+    private static final String REVISION_ARG = "?revision=";
+
+    public NormalizedNodeInputStreamReader(InputStream stream) throws IOException {
+        Preconditions.checkNotNull(stream);
+        reader = new DataInputStream(stream);
+    }
+
+
+    public NormalizedNode<?, ?> readNormalizedNode() throws IOException {
+        NormalizedNode<?, ?> node = null;
+
+        // each node should start with a byte
+        byte nodeType = reader.readByte();
+
+        if(nodeType == NodeTypes.END_NODE) {
+            LOG.debug("End node reached. return");
+            return null;
+        }
+        else if(nodeType == NodeTypes.AUGMENTATION_NODE) {
+            LOG.debug("Reading augmentation node. will create augmentation identifier");
+
+            YangInstanceIdentifier.AugmentationIdentifier identifier =
+                new YangInstanceIdentifier.AugmentationIdentifier(readQNameSet());
+            DataContainerNodeBuilder<YangInstanceIdentifier.AugmentationIdentifier, AugmentationNode> augmentationBuilder =
+                Builders.augmentationBuilder().withNodeIdentifier(identifier);
+            augmentationBuilder = addDataContainerChildren(augmentationBuilder);
+            node = augmentationBuilder.build();
+
+        } else {
+            QName qName = readQName();
+
+            if(nodeType == NodeTypes.LEAF_SET_ENTRY_NODE) {
+                LOG.debug("Reading leaf set entry node. Will create NodeWithValue instance identifier");
+
+                // Read the object value
+                Object value = readObject();
+
+                YangInstanceIdentifier.NodeWithValue nodeWithValue = new YangInstanceIdentifier.NodeWithValue(qName, value);
+                node =  Builders.leafSetEntryBuilder().withNodeIdentifier(nodeWithValue).withValue(value).build();
+
+            } else if(nodeType == NodeTypes.MAP_ENTRY_NODE) {
+                LOG.debug("Reading map entry node. Will create node identifier with predicates.");
+
+                YangInstanceIdentifier.NodeIdentifierWithPredicates nodeIdentifier =
+                    new YangInstanceIdentifier.NodeIdentifierWithPredicates(qName, readKeyValueMap());
+                DataContainerNodeAttrBuilder<YangInstanceIdentifier.NodeIdentifierWithPredicates, MapEntryNode> mapEntryBuilder
+                    = Builders.mapEntryBuilder().withNodeIdentifier(nodeIdentifier);
+
+                mapEntryBuilder = (DataContainerNodeAttrBuilder<YangInstanceIdentifier.NodeIdentifierWithPredicates,
+                    MapEntryNode>)addDataContainerChildren(mapEntryBuilder);
+                node = mapEntryBuilder.build();
+
+            } else {
+                LOG.debug("Creating standard node identifier. ");
+                YangInstanceIdentifier.NodeIdentifier identifier = new YangInstanceIdentifier.NodeIdentifier(qName);
+                node = readNodeIdentifierDependentNode(nodeType, identifier);
+
+            }
+        }
+        return node;
+    }
+
+    private NormalizedNode<?, ?> readNodeIdentifierDependentNode(byte nodeType, YangInstanceIdentifier.NodeIdentifier identifier)
+        throws IOException {
+
+        switch(nodeType) {
+            case NodeTypes.LEAF_NODE :
+                LOG.debug("Read leaf node");
+                // Read the object value
+                NormalizedNodeAttrBuilder leafBuilder = Builders.leafBuilder();
+                return leafBuilder.withNodeIdentifier(identifier).withValue(readObject()).build();
+
+            case NodeTypes.ANY_XML_NODE :
+                LOG.debug("Read xml node");
+                Node value = (Node) readObject();
+                return Builders.anyXmlBuilder().withValue(value).build();
+
+            case NodeTypes.MAP_NODE :
+                LOG.debug("Read map node");
+                CollectionNodeBuilder<MapEntryNode, MapNode> mapBuilder = Builders.mapBuilder().withNodeIdentifier(identifier);
+                mapBuilder = addMapNodeChildren(mapBuilder);
+                return mapBuilder.build();
+
+            case NodeTypes.CHOICE_NODE :
+                LOG.debug("Read choice node");
+                DataContainerNodeBuilder<YangInstanceIdentifier.NodeIdentifier, ChoiceNode> choiceBuilder =
+                    Builders.choiceBuilder().withNodeIdentifier(identifier);
+                choiceBuilder = addDataContainerChildren(choiceBuilder);
+                return choiceBuilder.build();
+
+            case NodeTypes.ORDERED_MAP_NODE :
+                LOG.debug("Reading ordered map node");
+                CollectionNodeBuilder<MapEntryNode, OrderedMapNode> orderedMapBuilder =
+                    Builders.orderedMapBuilder().withNodeIdentifier(identifier);
+                orderedMapBuilder = addMapNodeChildren(orderedMapBuilder);
+                return orderedMapBuilder.build();
+
+            case NodeTypes.UNKEYED_LIST :
+                LOG.debug("Read unkeyed list node");
+                CollectionNodeBuilder<UnkeyedListEntryNode, UnkeyedListNode> unkeyedListBuilder =
+                    Builders.unkeyedListBuilder().withNodeIdentifier(identifier);
+                unkeyedListBuilder = addUnkeyedListChildren(unkeyedListBuilder);
+                return unkeyedListBuilder.build();
+
+            case NodeTypes.UNKEYED_LIST_ITEM :
+                LOG.debug("Read unkeyed list item node");
+                DataContainerNodeAttrBuilder<YangInstanceIdentifier.NodeIdentifier, UnkeyedListEntryNode> unkeyedListEntryBuilder
+                    = Builders.unkeyedListEntryBuilder().withNodeIdentifier(identifier);
+
+                unkeyedListEntryBuilder = (DataContainerNodeAttrBuilder<YangInstanceIdentifier.NodeIdentifier, UnkeyedListEntryNode>)
+                    addDataContainerChildren(unkeyedListEntryBuilder);
+                return unkeyedListEntryBuilder.build();
+
+            case NodeTypes.CONTAINER_NODE :
+                LOG.debug("Read container node");
+                DataContainerNodeAttrBuilder<YangInstanceIdentifier.NodeIdentifier, ContainerNode> containerBuilder =
+                    Builders.containerBuilder().withNodeIdentifier(identifier);
+
+                containerBuilder = (DataContainerNodeAttrBuilder<YangInstanceIdentifier.NodeIdentifier, ContainerNode>)
+                    addDataContainerChildren(containerBuilder);
+                return containerBuilder.build();
+
+            case NodeTypes.LEAF_SET :
+                LOG.debug("Read leaf set node");
+                ListNodeBuilder<Object, LeafSetEntryNode<Object>> leafSetBuilder =
+                    Builders.leafSetBuilder().withNodeIdentifier(identifier);
+                leafSetBuilder = addLeafSetChildren(leafSetBuilder);
+                return leafSetBuilder.build();
+
+            default :
+                return null;
+        }
+    }
+
+    private QName readQName() throws IOException {
+        // Read in the same sequence of writing
+        String localName = readCodedString();
+        String namespace = readCodedString();
+        String revision = readCodedString();
+        String qName;
+        // Not using stringbuilder as compiler optimizes string concatenation of +
+        if(revision != null){
+            qName = "(" + namespace+ REVISION_ARG + revision + ")" +localName;
+        } else {
+            qName = "(" + namespace + ")" +localName;
+        }
+
+        return QNameFactory.create(qName);
+    }
+
+
+    private String readCodedString() throws IOException {
+        boolean readFromMap = reader.readBoolean();
+        if(readFromMap) {
+            return codedStringMap.get(reader.readInt());
+        } else {
+            String value = reader.readUTF();
+            if(value != null) {
+                codedStringMap.put(Integer.valueOf(codedStringMap.size()), value);
+            }
+            return value;
+        }
+    }
+
+    private Set<QName> readQNameSet() throws IOException{
+        // Read the children count
+        int count = reader.readInt();
+        Set<QName> children = new HashSet<>(count);
+        for(int i = 0; i<count; i++) {
+            children.add(readQName());
+        }
+        return children;
+    }
+
+    private Map<QName, Object> readKeyValueMap() throws IOException {
+        int count = reader.readInt();
+        Map<QName, Object> keyValueMap = new HashMap<>(count);
+
+        for(int i = 0; i<count; i++) {
+            keyValueMap.put(readQName(), readObject());
+        }
+
+        return keyValueMap;
+    }
+
+    private Object readObject() throws IOException {
+        byte objectType = reader.readByte();
+        switch(objectType) {
+            case ValueTypes.BITS_TYPE:
+                return readObjSet();
+
+            case ValueTypes.BOOL_TYPE :
+                return reader.readBoolean();
+
+            case ValueTypes.BYTE_TYPE :
+                return reader.readByte();
+
+            case ValueTypes.INT_TYPE :
+                return reader.readInt();
+
+            case ValueTypes.LONG_TYPE :
+                return reader.readLong();
+
+            case ValueTypes.QNAME_TYPE :
+                return readQName();
+
+            case ValueTypes.SHORT_TYPE :
+                return reader.readShort();
+
+            case ValueTypes.STRING_TYPE :
+                return reader.readUTF();
+
+            case ValueTypes.BIG_DECIMAL_TYPE :
+                return new BigDecimal(reader.readUTF());
+
+            case ValueTypes.BIG_INTEGER_TYPE :
+                return new BigInteger(reader.readUTF());
+
+            case ValueTypes.YANG_IDENTIFIER_TYPE :
+                int size = reader.readInt();
+
+                List<YangInstanceIdentifier.PathArgument> pathArguments = new ArrayList<>(size);
+
+                for(int i=0; i<size; i++) {
+                    pathArguments.add(readPathArgument());
+                }
+                return YangInstanceIdentifier.create(pathArguments);
+
+            default :
+                return null;
+        }
+    }
+
+    private Set<String> readObjSet() throws IOException {
+        int count = reader.readInt();
+        Set<String> children = new HashSet<>(count);
+        for(int i = 0; i<count; i++) {
+            children.add(readCodedString());
+        }
+        return children;
+    }
+
+    private YangInstanceIdentifier.PathArgument readPathArgument() throws IOException {
+        // read Type
+        int type = reader.readByte();
+
+        switch(type) {
+
+            case PathArgumentTypes.AUGMENTATION_IDENTIFIER :
+                return new YangInstanceIdentifier.AugmentationIdentifier(readQNameSet());
+
+            case PathArgumentTypes.NODE_IDENTIFIER :
+            return new YangInstanceIdentifier.NodeIdentifier(readQName());
+
+            case PathArgumentTypes.NODE_IDENTIFIER_WITH_PREDICATES :
+            return new YangInstanceIdentifier.NodeIdentifierWithPredicates(readQName(), readKeyValueMap());
+
+            case PathArgumentTypes.NODE_IDENTIFIER_WITH_VALUE :
+            return new YangInstanceIdentifier.NodeWithValue(readQName(), readObject());
+
+            default :
+                return null;
+        }
+    }
+
+    private ListNodeBuilder<Object, LeafSetEntryNode<Object>> addLeafSetChildren(ListNodeBuilder<Object,
+        LeafSetEntryNode<Object>> builder)
+        throws IOException {
+
+        LOG.debug("Reading children of leaf set");
+        LeafSetEntryNode<Object> child = (LeafSetEntryNode<Object>)readNormalizedNode();
+
+        while(child != null) {
+            builder.withChild(child);
+            child = (LeafSetEntryNode<Object>)readNormalizedNode();
+        }
+        return builder;
+    }
+
+    private CollectionNodeBuilder<UnkeyedListEntryNode, UnkeyedListNode> addUnkeyedListChildren(
+        CollectionNodeBuilder<UnkeyedListEntryNode, UnkeyedListNode> builder)
+        throws IOException{
+
+        LOG.debug("Reading children of unkeyed list");
+        UnkeyedListEntryNode child = (UnkeyedListEntryNode)readNormalizedNode();
+
+        while(child != null) {
+            builder.withChild(child);
+            child = (UnkeyedListEntryNode)readNormalizedNode();
+        }
+        return builder;
+    }
+
+    private DataContainerNodeBuilder addDataContainerChildren(DataContainerNodeBuilder builder)
+        throws IOException {
+        LOG.debug("Reading data container (leaf nodes) nodes");
+
+        DataContainerChild<? extends YangInstanceIdentifier.PathArgument, ?> child =
+            (DataContainerChild<? extends YangInstanceIdentifier.PathArgument, ?>) readNormalizedNode();
+
+        while(child != null) {
+            builder.withChild(child);
+            child =
+                (DataContainerChild<? extends YangInstanceIdentifier.PathArgument, ?>) readNormalizedNode();
+        }
+        return builder;
+    }
+
+
+    private CollectionNodeBuilder addMapNodeChildren(CollectionNodeBuilder builder)
+        throws IOException {
+        LOG.debug("Reading map node children");
+        MapEntryNode child = (MapEntryNode)readNormalizedNode();
+
+        while(child != null){
+            builder.withChild(child);
+            child = (MapEntryNode)readNormalizedNode();
+        }
+
+        return builder;
+    }
+
+
+    @Override
+    public void close() throws IOException {
+        reader.close();
+    }
+
+}
diff --git a/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeOutputStreamWriter.java b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeOutputStreamWriter.java
new file mode 100644
index 0000000000..05a47a0401
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeOutputStreamWriter.java
@@ -0,0 +1,343 @@
+/*
+ *
+ *  Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ *  and is available at http://www.eclipse.org/legal/epl-v10.html
+ *
+ */
+
+package org.opendaylight.controller.cluster.datastore.node.utils.stream;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * NormalizedNodeOutputStreamWriter will be used by distributed datastore to send normalized node in
+ * a stream.
+ * A stream writer wrapper around this class will write node objects to stream in recursive manner.
+ * for example - If you have a ContainerNode which has a two LeafNode as children, then
+ * you will first call {@link #startContainerNode(YangInstanceIdentifier.NodeIdentifier, int)}, then will call
+ * {@link #leafNode(YangInstanceIdentifier.NodeIdentifier, Object)} twice and then, {@link #endNode()} to end
+ * container node.
+ *
+ * Based on the each node, the node type is also written to the stream, that helps in reconstructing the object,
+ * while reading.
+ *
+ *
+ */
+
+public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWriter{
+
+    private DataOutputStream writer;
+
+    private static final Logger LOG = LoggerFactory.getLogger(NormalizedNodeOutputStreamWriter.class);
+
+    private Map<String, Integer> stringCodeMap = new HashMap<>();
+
+    public NormalizedNodeOutputStreamWriter(OutputStream stream) throws IOException {
+        Preconditions.checkNotNull(stream);
+        writer = new DataOutputStream(stream);
+    }
+
+    @Override
+    public void leafNode(YangInstanceIdentifier.NodeIdentifier name, Object value) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(name, "Node identifier should not be null");
+        LOG.debug("Writing a new leaf node");
+        startNode(name.getNodeType(), NodeTypes.LEAF_NODE);
+
+        writeObject(value);
+    }
+
+    @Override
+    public void startLeafSet(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(name, "Node identifier should not be null");
+        LOG.debug("Starting a new leaf set");
+
+        startNode(name.getNodeType(), NodeTypes.LEAF_SET);
+    }
+
+    @Override
+    public void leafSetEntryNode(YangInstanceIdentifier.NodeWithValue name, Object value) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(name, "Node identifier should not be null");
+
+        LOG.debug("Writing a new leaf set entry node");
+        startNode(name.getNodeType(), NodeTypes.LEAF_SET_ENTRY_NODE);
+
+        writeObject(value);
+    }
+
+    @Override
+    public void startContainerNode(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(name, "Node identifier should not be null");
+
+        LOG.debug("Starting a new container node");
+
+        startNode(name.getNodeType(), NodeTypes.CONTAINER_NODE);
+    }
+
+    @Override
+    public void startUnkeyedList(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(name, "Node identifier should not be null");
+        LOG.debug("Starting a new unkeyed list");
+
+        startNode(name.getNodeType(), NodeTypes.UNKEYED_LIST);
+    }
+
+    @Override
+    public void startUnkeyedListItem(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint) throws IOException, IllegalStateException {
+        Preconditions.checkNotNull(name, "Node identifier should not be null");
+        LOG.debug("Starting a new unkeyed list item");
+
+        startNode(name.getNodeType(), NodeTypes.UNKEYED_LIST_ITEM);
+    }
+
+    @Override
+    public void startMapNode(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(name, "Node identifier should not be null");
+        LOG.debug("Starting a new map node");
+
+        startNode(name.getNodeType(), NodeTypes.MAP_NODE);
+    }
+
+    @Override
+    public void startMapEntryNode(YangInstanceIdentifier.NodeIdentifierWithPredicates identifier, int childSizeHint) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(identifier, "Node identifier should not be null");
+        LOG.debug("Starting a new map entry node");
+        startNode(identifier.getNodeType(), NodeTypes.MAP_ENTRY_NODE);
+
+        writeKeyValueMap(identifier.getKeyValues());
+
+    }
+
+    @Override
+    public void startOrderedMapNode(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(name, "Node identifier should not be null");
+        LOG.debug("Starting a new ordered map node");
+
+        startNode(name.getNodeType(), NodeTypes.ORDERED_MAP_NODE);
+    }
+
+    @Override
+    public void startChoiceNode(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(name, "Node identifier should not be null");
+        LOG.debug("Starting a new choice node");
+
+        startNode(name.getNodeType(), NodeTypes.CHOICE_NODE);
+    }
+
+    @Override
+    public void startAugmentationNode(YangInstanceIdentifier.AugmentationIdentifier identifier) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(identifier, "Node identifier should not be null");
+        LOG.debug("Starting a new augmentation node");
+
+        writer.writeByte(NodeTypes.AUGMENTATION_NODE);
+        writeQNameSet(identifier.getPossibleChildNames());
+    }
+
+    @Override
+    public void anyxmlNode(YangInstanceIdentifier.NodeIdentifier name, Object value) throws IOException, IllegalArgumentException {
+        Preconditions.checkNotNull(name, "Node identifier should not be null");
+        LOG.debug("Writing a new xml node");
+
+        startNode(name.getNodeType(), NodeTypes.ANY_XML_NODE);
+
+        writeObject(value);
+    }
+
+    @Override
+    public void endNode() throws IOException, IllegalStateException {
+        LOG.debug("Ending the node");
+
+        writer.writeByte(NodeTypes.END_NODE);
+    }
+
+    @Override
+    public void close() throws IOException {
+        writer.close();
+    }
+
+    @Override
+    public void flush() throws IOException {
+        writer.flush();
+    }
+
+    private void startNode(final QName qName, byte nodeType) throws IOException {
+
+        Preconditions.checkNotNull(qName, "QName of node identifier should not be null.");
+        // First write the type of node
+        writer.writeByte(nodeType);
+        // Write Start Tag
+        writeQName(qName);
+    }
+
+    private void writeQName(QName qName) throws IOException {
+
+        writeCodedString(qName.getLocalName());
+        writeCodedString(qName.getNamespace().toString());
+        writeCodedString(qName.getFormattedRevision());
+    }
+
+    private void writeCodedString(String key) throws IOException {
+        Integer value = stringCodeMap.get(key);
+
+        if(value != null) {
+            writer.writeBoolean(true);
+            writer.writeInt(value);
+        } else {
+            if(key != null) {
+                stringCodeMap.put(key, Integer.valueOf(stringCodeMap.size()));
+            }
+            writer.writeBoolean(false);
+            writer.writeUTF(key);
+        }
+    }
+
+    private void writeObjSet(Set set) throws IOException {
+        if(!set.isEmpty()){
+            writer.writeInt(set.size());
+            for(Object o : set){
+                if(o instanceof String){
+                    writeCodedString(o.toString());
+                } else {
+                    throw new IllegalArgumentException("Expected value type to be String but was : " +
+                        o.toString());
+                }
+            }
+        } else {
+            writer.writeInt(0);
+        }
+    }
+
+    private void writeYangInstanceIdentifier(YangInstanceIdentifier identifier) throws IOException {
+        Iterable<YangInstanceIdentifier.PathArgument> pathArguments = identifier.getPathArguments();
+        int size = Iterables.size(pathArguments);
+        writer.writeInt(size);
+
+        for(YangInstanceIdentifier.PathArgument pathArgument : pathArguments) {
+            writePathArgument(pathArgument);
+        }
+    }
+
+    private void writePathArgument(YangInstanceIdentifier.PathArgument pathArgument) throws IOException {
+
+        byte type = PathArgumentTypes.getSerializablePathArgumentType(pathArgument);
+
+        writer.writeByte(type);
+
+        switch(type) {
+            case PathArgumentTypes.NODE_IDENTIFIER :
+
+                YangInstanceIdentifier.NodeIdentifier nodeIdentifier =
+                    (YangInstanceIdentifier.NodeIdentifier) pathArgument;
+
+                writeQName(nodeIdentifier.getNodeType());
+                break;
+
+            case PathArgumentTypes.NODE_IDENTIFIER_WITH_PREDICATES:
+
+                YangInstanceIdentifier.NodeIdentifierWithPredicates nodeIdentifierWithPredicates =
+                    (YangInstanceIdentifier.NodeIdentifierWithPredicates) pathArgument;
+                writeQName(nodeIdentifierWithPredicates.getNodeType());
+
+                writeKeyValueMap(nodeIdentifierWithPredicates.getKeyValues());
+                break;
+
+            case PathArgumentTypes.NODE_IDENTIFIER_WITH_VALUE :
+
+                YangInstanceIdentifier.NodeWithValue nodeWithValue =
+                    (YangInstanceIdentifier.NodeWithValue) pathArgument;
+
+                writeQName(nodeWithValue.getNodeType());
+                writeObject(nodeWithValue.getValue());
+                break;
+
+            case PathArgumentTypes.AUGMENTATION_IDENTIFIER :
+
+                YangInstanceIdentifier.AugmentationIdentifier augmentationIdentifier =
+                    (YangInstanceIdentifier.AugmentationIdentifier) pathArgument;
+
+                // No Qname in augmentation identifier
+                writeQNameSet(augmentationIdentifier.getPossibleChildNames());
+                break;
+            default :
+                throw new IllegalStateException("Unknown node identifier type is found : " + pathArgument.getClass().toString() );
+        }
+    }
+
+    private void writeKeyValueMap(Map<QName, Object> keyValueMap) throws IOException {
+        if(keyValueMap != null && !keyValueMap.isEmpty()) {
+            writer.writeInt(keyValueMap.size());
+            Set<QName> qNameSet = keyValueMap.keySet();
+
+            for(QName qName : qNameSet) {
+                writeQName(qName);
+                writeObject(keyValueMap.get(qName));
+            }
+        } else {
+            writer.writeInt(0);
+        }
+    }
+
+    private void writeQNameSet(Set<QName> children) throws IOException {
+        // Write each child's qname separately, if list is empty send count as 0
+        if(children != null && !children.isEmpty()) {
+            writer.writeInt(children.size());
+            for(QName qName : children) {
+                writeQName(qName);
+            }
+        } else {
+            LOG.debug("augmentation node does not have any child");
+            writer.writeInt(0);
+        }
+    }
+
+    private void writeObject(Object value) throws IOException {
+
+        byte type = ValueTypes.getSerializableType(value);
+        // Write object type first
+        writer.writeByte(type);
+
+        switch(type) {
+            case ValueTypes.BOOL_TYPE:
+                writer.writeBoolean((Boolean) value);
+                break;
+            case ValueTypes.QNAME_TYPE:
+                writeQName((QName) value);
+                break;
+            case ValueTypes.INT_TYPE:
+                writer.writeInt((Integer) value);
+                break;
+            case ValueTypes.BYTE_TYPE:
+                writer.writeByte((Byte) value);
+                break;
+            case ValueTypes.LONG_TYPE:
+                writer.writeLong((Long) value);
+                break;
+            case ValueTypes.SHORT_TYPE:
+                writer.writeShort((Short) value);
+                break;
+            case ValueTypes.BITS_TYPE:
+                writeObjSet((Set) value);
+                break;
+            case ValueTypes.YANG_IDENTIFIER_TYPE:
+                writeYangInstanceIdentifier((YangInstanceIdentifier) value);
+                break;
+            default:
+                writer.writeUTF(value.toString());
+                break;
+        }
+    }
+}
diff --git a/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeStreamReader.java b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeStreamReader.java
new file mode 100644
index 0000000000..c619afd7ee
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeStreamReader.java
@@ -0,0 +1,23 @@
+/*
+ *
+ *  Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ *  and is available at http://www.eclipse.org/legal/epl-v10.html
+ *
+ */
+
+
+package org.opendaylight.controller.cluster.datastore.node.utils.stream;
+
+
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+
+import java.io.IOException;
+
+
+public interface NormalizedNodeStreamReader extends AutoCloseable {
+
+    NormalizedNode<?, ?> readNormalizedNode() throws IOException;
+}
diff --git a/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeStreamWriter.java b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeStreamWriter.java
new file mode 100644
index 0000000000..af95b61423
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeStreamWriter.java
@@ -0,0 +1,219 @@
+
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.datastore.node.utils.stream;
+
+import java.io.Closeable;
+import java.io.Flushable;
+import java.io.IOException;
+
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+
+/**
+ * Event Stream Writer based on Normalized Node tree representation
+ *
+ * <h3>Writing Event Stream</h3>
+ *
+ * <ul>
+ * <li><code>container</code> - Container node representation, start event is
+ * emitted using {@link #startContainerNode(YangInstanceIdentifier.NodeIdentifier, int)}
+ * and node end event is
+ * emitted using {@link #endNode()}. Container node is implementing
+ * {@link org.opendaylight.yangtools.yang.binding.DataObject} interface.
+ *
+ * <li><code>list</code> - YANG list statement has two representation in event
+ * stream - unkeyed list and map. Unkeyed list is YANG list which did not
+ * specify key.</li>
+ *
+ * <ul>
+ * <li><code>Map</code> - Map start event is emitted using
+ * {@link #startMapNode(YangInstanceIdentifier.NodeIdentifier, int)}
+ * and is ended using {@link #endNode()}. Each map entry start is emitted using
+ * {@link #startMapEntryNode(YangInstanceIdentifier.NodeIdentifierWithPredicates, int)}
+ * with Map of keys
+ * and finished using {@link #endNode()}.</li>
+ *
+ * <li><code>UnkeyedList</code> - Unkeyed list represent list without keys,
+ * unkeyed list start is emitted using
+ * {@link #startUnkeyedList(YangInstanceIdentifier.NodeIdentifier, int)} list
+ * end is emitted using {@link #endNode()}. Each list item is emitted using
+ * {@link #startUnkeyedListItem(YangInstanceIdentifier.NodeIdentifier, int)}
+ * and ended using {@link #endNode()}.</li>
+ * </ul>
+ *
+ * <li><code>leaf</code> - Leaf node event is emitted using
+ * {@link #leafNode(YangInstanceIdentifier.NodeIdentifier, Object)}.
+ * {@link #endNode()} MUST NOT BE emitted for
+ * leaf node.</li>
+ *
+ * <li><code>leaf-list</code> - Leaf list start is emitted using
+ * {@link #startLeafSet(YangInstanceIdentifier.NodeIdentifier, int)}.
+ * Leaf list end is emitted using
+ * {@link #endNode()}. Leaf list entries are emitted using
+ * {@link #leafSetEntryNode(YangInstanceIdentifier.NodeWithValue name, Object).
+ *
+ * <li><code>anyxml - Anyxml node event is emitted using
+ * {@link #leafNode(YangInstanceIdentifier.NodeIdentifier, Object)}. {@link #endNode()} MUST NOT BE emitted
+ * for anyxml node.</code></li>
+ *
+ *
+ * <li><code>choice</code> Choice node event is emmited by
+ * {@link #startChoiceNode(YangInstanceIdentifier.NodeIdentifier, int)} event and
+ * finished by invoking {@link #endNode()}
+ * <li>
+ * <code>augment</code> - Represents augmentation, augmentation node is started
+ * by invoking {@link #startAugmentationNode(YangInstanceIdentifier.AugmentationIdentifier)} and
+ * finished by invoking {@link #endNode()}.</li>
+ *
+ * </ul>
+ *
+ * <h3>Implementation notes</h3>
+ *
+ * <p>
+ * Implementations of this interface must not hold user suppled objects
+ * and resources needlessly.
+ *
+ */
+
+public interface NormalizedNodeStreamWriter extends Closeable, Flushable {
+
+    public final int UNKNOWN_SIZE = -1;
+
+    /**
+     * Write the leaf node identifier and value to the stream.
+     * @param name
+     * @param value
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void leafNode(YangInstanceIdentifier.NodeIdentifier name, Object value)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * Start writing leaf Set node. You must call {@link #endNode()} once you are done writing all of its children.
+     * @param name
+     * @param childSizeHint is the estimated children count. Usage is optional in implementation.
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void startLeafSet(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * Write the leaf Set Entry Node object to the stream with identifier and value.
+     * @param name
+     * @param value
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void leafSetEntryNode(YangInstanceIdentifier.NodeWithValue name, Object value)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * Start writing container node. You must call {@link #endNode()} once you are done writing all of its children.
+     * @param name
+     * @param childSizeHint is the estimated children count. Usage is optional in implementation.
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void startContainerNode(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * Start writing unkeyed list node. You must call {@link #endNode()} once you are done writing all of its children.
+     * @param name
+     * @param childSizeHint is the estimated children count. Usage is optional in implementation.
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void startUnkeyedList(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * Start writing unkeyed list item. You must call {@link #endNode()} once you are done writing all of its children.
+     * @param name
+     * @param childSizeHint is the estimated children count. Usage is optional in implementation.
+     * @throws IOException
+     * @throws IllegalStateException
+     */
+    void startUnkeyedListItem(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint)
+        throws IOException, IllegalStateException;
+
+    /**
+     * Start writing map node. You must call {@link #endNode()} once you are done writing all of its children.
+     * @param name
+     * @param childSizeHint is the estimated children count. Usage is optional in implementation.
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void startMapNode(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * Start writing map entry node. You must call {@link #endNode()} once you are done writing all of its children.
+     * @param identifier
+     * @param childSizeHint is the estimated children count. Usage is optional in implementation.
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void startMapEntryNode(YangInstanceIdentifier.NodeIdentifierWithPredicates identifier, int childSizeHint)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * Start writing ordered map node. You must call {@link #endNode()} once you are done writing all of its children.
+     * @param name
+     * @param childSizeHint is the estimated children count. Usage is optional in implementation.
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void startOrderedMapNode(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * Start writing choice node. You must call {@link #endNode()} once you are done writing all of its children.
+     * @param name
+     * @param childSizeHint is the estimated children count. Usage is optional in implementation.
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void startChoiceNode(YangInstanceIdentifier.NodeIdentifier name, int childSizeHint)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * Start writing augmentation node. You must call {@link #endNode()} once you are done writing all of its children.
+     * @param identifier
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void startAugmentationNode(YangInstanceIdentifier.AugmentationIdentifier identifier)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * Write any xml node identifier and value to the stream
+     * @param name
+     * @param value
+     * @throws IOException
+     * @throws IllegalArgumentException
+     */
+    void anyxmlNode(YangInstanceIdentifier.NodeIdentifier name, Object value)
+        throws IOException, IllegalArgumentException;
+
+    /**
+     * This method should be used to add end symbol/identifier of node in the stream.
+     * @throws IOException
+     * @throws IllegalStateException
+     */
+    void endNode() throws IOException, IllegalStateException;
+
+    @Override
+    void close() throws IOException;
+
+    @Override
+    void flush() throws IOException;
+}
diff --git a/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/PathArgumentTypes.java b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/PathArgumentTypes.java
new file mode 100644
index 0000000000..b01beb8c77
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/PathArgumentTypes.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.datastore.node.utils.stream;
+
+import com.google.common.collect.ImmutableMap;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+
+import java.util.Map;
+
+public class PathArgumentTypes {
+    public static final byte AUGMENTATION_IDENTIFIER = 1;
+    public static final byte NODE_IDENTIFIER = 2;
+    public static final byte NODE_IDENTIFIER_WITH_VALUE = 3;
+    public static final byte NODE_IDENTIFIER_WITH_PREDICATES = 4;
+
+    private static Map<Class<?>, Byte> CLASS_TO_ENUM_MAP =
+            ImmutableMap.<Class<?>, Byte>builder().
+                put(YangInstanceIdentifier.AugmentationIdentifier.class, AUGMENTATION_IDENTIFIER).
+                put(YangInstanceIdentifier.NodeIdentifier.class, NODE_IDENTIFIER).
+                put(YangInstanceIdentifier.NodeIdentifierWithPredicates.class, NODE_IDENTIFIER_WITH_PREDICATES).
+                put(YangInstanceIdentifier.NodeWithValue.class, NODE_IDENTIFIER_WITH_VALUE).build();
+
+    public static byte getSerializablePathArgumentType(YangInstanceIdentifier.PathArgument pathArgument){
+
+        Byte type = CLASS_TO_ENUM_MAP.get(pathArgument.getClass());
+        if(type == null) {
+            throw new IllegalArgumentException("Unknown type of PathArgument = " + pathArgument);
+        }
+
+        return type;
+    }
+
+}
diff --git a/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/ValueTypes.java b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/ValueTypes.java
new file mode 100644
index 0000000000..6035e3c644
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/ValueTypes.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.datastore.node.utils.stream;
+
+import com.google.common.base.Preconditions;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+public class ValueTypes {
+    public static final byte SHORT_TYPE = 1;
+    public static final byte BYTE_TYPE = 2;
+    public static final byte INT_TYPE = 3;
+    public static final byte LONG_TYPE = 4;
+    public static final byte BOOL_TYPE = 5;
+    public static final byte QNAME_TYPE = 6;
+    public static final byte BITS_TYPE = 7;
+    public static final byte YANG_IDENTIFIER_TYPE = 8;
+    public static final byte STRING_TYPE = 9;
+    public static final byte BIG_INTEGER_TYPE = 10;
+    public static final byte BIG_DECIMAL_TYPE = 11;
+
+    private static Map<Class, Byte> types = new HashMap<>();
+
+    static {
+        types.put(String.class, Byte.valueOf(STRING_TYPE));
+        types.put(Byte.class, Byte.valueOf(BYTE_TYPE));
+        types.put(Integer.class, Byte.valueOf(INT_TYPE));
+        types.put(Long.class, Byte.valueOf(LONG_TYPE));
+        types.put(Boolean.class, Byte.valueOf(BOOL_TYPE));
+        types.put(QName.class, Byte.valueOf(QNAME_TYPE));
+        types.put(Set.class, Byte.valueOf(BITS_TYPE));
+        types.put(YangInstanceIdentifier.class, Byte.valueOf(YANG_IDENTIFIER_TYPE));
+        types.put(Short.class, Byte.valueOf(SHORT_TYPE));
+        types.put(BigInteger.class, Byte.valueOf(BIG_INTEGER_TYPE));
+        types.put(BigDecimal.class, Byte.valueOf(BIG_DECIMAL_TYPE));
+    }
+
+    public static final byte getSerializableType(Object node){
+        Preconditions.checkNotNull(node, "node should not be null");
+
+        Byte type = types.get(node.getClass());
+        if(type != null) {
+            return type;
+        } else if(node instanceof Set){
+            return BITS_TYPE;
+        }
+
+        throw new IllegalArgumentException("Unknown value type " + node.getClass().getSimpleName());
+    }
+}
diff --git a/opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeStreamReaderWriterTest.java b/opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeStreamReaderWriterTest.java
new file mode 100644
index 0000000000..052f609e92
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeStreamReaderWriterTest.java
@@ -0,0 +1,65 @@
+/*
+ *
+ *  Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ *  and is available at http://www.eclipse.org/legal/epl-v10.html
+ *
+ */
+
+package org.opendaylight.controller.cluster.datastore.node.utils.stream;
+
+
+import org.apache.commons.lang.SerializationUtils;
+import org.junit.Assert;
+import org.junit.Test;
+import org.opendaylight.controller.cluster.datastore.util.TestModel;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+import static org.junit.Assert.fail;
+
+public class NormalizedNodeStreamReaderWriterTest {
+
+    final NormalizedNode<?, ?> input = TestModel.createTestContainer();
+
+    @Test
+    public void testNormalizedNodeStreamReaderWriter() {
+
+        byte[] byteData = null;
+
+        try(ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+            NormalizedNodeStreamWriter writer = new NormalizedNodeOutputStreamWriter(byteArrayOutputStream)) {
+
+            NormalizedNodeWriter normalizedNodeWriter = NormalizedNodeWriter.forStreamWriter(writer);
+            normalizedNodeWriter.write(input);
+            byteData = byteArrayOutputStream.toByteArray();
+
+        } catch (IOException e) {
+            fail("Writing to OutputStream failed :" + e.toString());
+        }
+
+        try(NormalizedNodeInputStreamReader reader = new NormalizedNodeInputStreamReader(new ByteArrayInputStream(byteData))) {
+
+            NormalizedNode<?,?> node = reader.readNormalizedNode();
+            Assert.assertEquals(input, node);
+
+        } catch (IOException e) {
+            fail("Reading from InputStream failed :" + e.toString());
+        }
+    }
+
+    @Test
+    public void testWithSerializable() {
+        SampleNormalizedNodeSerializable serializable = new SampleNormalizedNodeSerializable(input);
+        SampleNormalizedNodeSerializable clone = (SampleNormalizedNodeSerializable)SerializationUtils.clone(serializable);
+
+        Assert.assertEquals(input, clone.getInput());
+
+    }
+
+}
diff --git a/opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeWriter.java b/opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeWriter.java
new file mode 100644
index 0000000000..845038e3fd
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/NormalizedNodeWriter.java
@@ -0,0 +1,190 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore.node.utils.stream;
+
+import com.google.common.base.Preconditions;
+import org.opendaylight.yangtools.yang.data.api.schema.AnyXmlNode;
+import org.opendaylight.yangtools.yang.data.api.schema.AugmentationNode;
+import org.opendaylight.yangtools.yang.data.api.schema.ChoiceNode;
+import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
+import org.opendaylight.yangtools.yang.data.api.schema.LeafNode;
+import org.opendaylight.yangtools.yang.data.api.schema.LeafSetEntryNode;
+import org.opendaylight.yangtools.yang.data.api.schema.LeafSetNode;
+import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
+import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.api.schema.OrderedMapNode;
+import org.opendaylight.yangtools.yang.data.api.schema.UnkeyedListEntryNode;
+import org.opendaylight.yangtools.yang.data.api.schema.UnkeyedListNode;
+
+import java.io.Closeable;
+import java.io.Flushable;
+import java.io.IOException;
+import java.util.Collection;
+
+import static org.opendaylight.controller.cluster.datastore.node.utils.stream.NormalizedNodeStreamWriter.UNKNOWN_SIZE;
+
+
+/**
+ * This class is used only for testing purpose for now, we may use similar logic while integrating
+ * with cluster
+ */
+
+public class NormalizedNodeWriter implements Closeable, Flushable {
+    private final NormalizedNodeStreamWriter writer;
+
+    private NormalizedNodeWriter(final NormalizedNodeStreamWriter writer) {
+        this.writer = Preconditions.checkNotNull(writer);
+    }
+
+    protected final NormalizedNodeStreamWriter getWriter() {
+        return writer;
+    }
+
+    /**
+     * Create a new writer backed by a {@link org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeStreamWriter}.
+     *
+     * @param writer Back-end writer
+     * @return A new instance.
+     */
+    public static NormalizedNodeWriter forStreamWriter(final NormalizedNodeStreamWriter writer) {
+        return new NormalizedNodeWriter(writer);
+    }
+
+
+    /**
+     * Iterate over the provided {@link org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode} and emit write
+     * events to the encapsulated {@link org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeStreamWriter}.
+     *
+     * @param node Node
+     * @return
+     * @throws java.io.IOException when thrown from the backing writer.
+     */
+    public final NormalizedNodeWriter write(final NormalizedNode<?, ?> node) throws IOException {
+        if (wasProcessedAsComplexNode(node)) {
+            return this;
+        }
+
+        if (wasProcessAsSimpleNode(node)) {
+            return this;
+        }
+
+        throw new IllegalStateException("It wasn't possible to serialize node " + node);
+    }
+
+    @Override
+    public void flush() throws IOException {
+        writer.flush();
+    }
+
+    @Override
+    public void close() throws IOException {
+        writer.flush();
+        writer.close();
+    }
+
+    /**
+     * Emit a best guess of a hint for a particular set of children. It evaluates the
+     * iterable to see if the size can be easily gotten to. If it is, we hint at the
+     * real number of child nodes. Otherwise we emit UNKNOWN_SIZE.
+     *
+     * @param children Child nodes
+     * @return Best estimate of the collection size required to hold all the children.
+     */
+    static final int childSizeHint(final Iterable<?> children) {
+        return (children instanceof Collection) ? ((Collection<?>) children).size() : UNKNOWN_SIZE;
+    }
+
+    private boolean wasProcessAsSimpleNode(final NormalizedNode<?, ?> node) throws IOException {
+        if (node instanceof LeafSetEntryNode) {
+            final LeafSetEntryNode<?> nodeAsLeafList = (LeafSetEntryNode<?>)node;
+            writer.leafSetEntryNode(nodeAsLeafList.getIdentifier(), nodeAsLeafList.getValue());
+            return true;
+        } else if (node instanceof LeafNode) {
+            final LeafNode<?> nodeAsLeaf = (LeafNode<?>)node;
+            writer.leafNode(nodeAsLeaf.getIdentifier(), nodeAsLeaf.getValue());
+            return true;
+        } else if (node instanceof AnyXmlNode) {
+            final AnyXmlNode anyXmlNode = (AnyXmlNode)node;
+            writer.anyxmlNode(anyXmlNode.getIdentifier(), anyXmlNode.getValue());
+            return true;
+        }
+
+        return false;
+    }
+
+    /**
+     * Emit events for all children and then emit an endNode() event.
+     *
+     * @param children Child iterable
+     * @return True
+     * @throws java.io.IOException when the writer reports it
+     */
+    protected final boolean writeChildren(final Iterable<? extends NormalizedNode<?, ?>> children) throws IOException {
+        for (NormalizedNode<?, ?> child : children) {
+            write(child);
+        }
+
+        writer.endNode();
+        return true;
+    }
+
+    protected boolean writeMapEntryNode(final MapEntryNode node) throws IOException {
+        writer.startMapEntryNode(node.getIdentifier(), childSizeHint(node.getValue()));
+        return writeChildren(node.getValue());
+    }
+
+    private boolean wasProcessedAsComplexNode(final NormalizedNode<?, ?> node) throws IOException {
+        if (node instanceof ContainerNode) {
+            final ContainerNode n = (ContainerNode) node;
+            writer.startContainerNode(n.getIdentifier(), childSizeHint(n.getValue()));
+            return writeChildren(n.getValue());
+        }
+        if (node instanceof MapEntryNode) {
+            return writeMapEntryNode((MapEntryNode) node);
+        }
+        if (node instanceof UnkeyedListEntryNode) {
+            final UnkeyedListEntryNode n = (UnkeyedListEntryNode) node;
+            writer.startUnkeyedListItem(n.getIdentifier(), childSizeHint(n.getValue()));
+            return writeChildren(n.getValue());
+        }
+        if (node instanceof ChoiceNode) {
+            final ChoiceNode n = (ChoiceNode) node;
+            writer.startChoiceNode(n.getIdentifier(), childSizeHint(n.getValue()));
+            return writeChildren(n.getValue());
+        }
+        if (node instanceof AugmentationNode) {
+            final AugmentationNode n = (AugmentationNode) node;
+            writer.startAugmentationNode(n.getIdentifier());
+            return writeChildren(n.getValue());
+        }
+        if (node instanceof UnkeyedListNode) {
+            final UnkeyedListNode n = (UnkeyedListNode) node;
+            writer.startUnkeyedList(n.getIdentifier(), childSizeHint(n.getValue()));
+            return writeChildren(n.getValue());
+        }
+        if (node instanceof OrderedMapNode) {
+            final OrderedMapNode n = (OrderedMapNode) node;
+            writer.startOrderedMapNode(n.getIdentifier(), childSizeHint(n.getValue()));
+            return writeChildren(n.getValue());
+        }
+        if (node instanceof MapNode) {
+            final MapNode n = (MapNode) node;
+            writer.startMapNode(n.getIdentifier(), childSizeHint(n.getValue()));
+            return writeChildren(n.getValue());
+        }
+        if (node instanceof LeafSetNode) {
+            //covers also OrderedLeafSetNode for which doesn't exist start* method
+            final LeafSetNode<?> n = (LeafSetNode<?>) node;
+            writer.startLeafSet(n.getIdentifier(), childSizeHint(n.getValue()));
+            return writeChildren(n.getValue());
+        }
+
+        return false;
+    }
+}
diff --git a/opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/SampleNormalizedNodeSerializable.java b/opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/SampleNormalizedNodeSerializable.java
new file mode 100644
index 0000000000..33d48a5278
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/SampleNormalizedNodeSerializable.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.datastore.node.utils.stream;
+
+
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.net.URISyntaxException;
+
+public class SampleNormalizedNodeSerializable implements Serializable {
+
+    private NormalizedNode<?, ?> input;
+
+    public SampleNormalizedNodeSerializable(NormalizedNode<?, ?> input) {
+        this.input = input;
+    }
+
+    public NormalizedNode<?, ?> getInput() {
+        return input;
+    }
+
+    private void readObject(final ObjectInputStream stream) throws IOException, ClassNotFoundException, URISyntaxException {
+        NormalizedNodeStreamReader reader = new NormalizedNodeInputStreamReader(stream);
+        this.input = reader.readNormalizedNode();
+    }
+
+    private void writeObject(final ObjectOutputStream stream) throws IOException {
+        NormalizedNodeStreamWriter writer = new NormalizedNodeOutputStreamWriter(stream);
+        NormalizedNodeWriter normalizedNodeWriter = NormalizedNodeWriter.forStreamWriter(writer);
+
+        normalizedNodeWriter.write(this.input);
+    }
+
+}