X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-clustering-commons%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Fnode%2Futils%2Fstream%2FNormalizedNodeOutputStreamWriter.java;h=b155b5af11b67ffa7856ac5a23fabd90ff2661ec;hb=5c3e3812eca4e7a06c69c376037578fae49828a2;hp=cbd7bf885373b331fb71a9d6899efc755b85f767;hpb=98b9623656bdd4cb4f86348f6bc9fe71ec443fbf;p=controller.git 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 index cbd7bf8853..b155b5af11 100644 --- 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 @@ -1,29 +1,30 @@ /* + * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others. All rights reserved. * - * 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 - * + * 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.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeStreamWriter; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - +import java.io.DataOutput; import java.io.DataOutputStream; import java.io.IOException; import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.Set; +import org.opendaylight.yangtools.yang.common.QName; +import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; +import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; +import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeStreamWriter; +import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeWriter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * NormalizedNodeOutputStreamWriter will be used by distributed datastore to send normalized node in @@ -40,17 +41,53 @@ import java.util.Set; * */ -public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWriter{ +public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWriter { private static final Logger LOG = LoggerFactory.getLogger(NormalizedNodeOutputStreamWriter.class); - private final DataOutputStream writer; + static final byte SIGNATURE_MARKER = (byte) 0xab; + static final short CURRENT_VERSION = (short) 1; + + static final byte IS_CODE_VALUE = 1; + static final byte IS_STRING_VALUE = 2; + static final byte IS_NULL_VALUE = 3; + + private final DataOutput output; private final Map stringCodeMap = new HashMap<>(); + private NormalizedNodeWriter normalizedNodeWriter; + + private boolean wroteSignatureMarker; + public NormalizedNodeOutputStreamWriter(OutputStream stream) throws IOException { Preconditions.checkNotNull(stream); - writer = new DataOutputStream(stream); + output = new DataOutputStream(stream); + } + + public NormalizedNodeOutputStreamWriter(DataOutput output) { + this.output = Preconditions.checkNotNull(output); + } + + private NormalizedNodeWriter normalizedNodeWriter() { + if(normalizedNodeWriter == null) { + normalizedNodeWriter = NormalizedNodeWriter.forStreamWriter(this); + } + + return normalizedNodeWriter; + } + + public void writeNormalizedNode(NormalizedNode node) throws IOException { + writeSignatureMarkerAndVersionIfNeeded(); + normalizedNodeWriter().write(node); + } + + private void writeSignatureMarkerAndVersionIfNeeded() throws IOException { + if(!wroteSignatureMarker) { + output.writeByte(SIGNATURE_MARKER); + output.writeShort(CURRENT_VERSION); + wroteSignatureMarker = true; + } } @Override @@ -74,7 +111,7 @@ public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWri public void leafSetEntryNode(Object value) throws IOException, IllegalArgumentException { LOG.debug("Writing a new leaf set entry node"); - writer.writeByte(NodeTypes.LEAF_SET_ENTRY_NODE); + output.writeByte(NodeTypes.LEAF_SET_ENTRY_NODE); writeObject(value); } @@ -142,7 +179,7 @@ public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWri Preconditions.checkNotNull(identifier, "Node identifier should not be null"); LOG.debug("Starting a new augmentation node"); - writer.writeByte(NodeTypes.AUGMENTATION_NODE); + output.writeByte(NodeTypes.AUGMENTATION_NODE); writeQNameSet(identifier.getPossibleChildNames()); } @@ -160,24 +197,29 @@ public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWri public void endNode() throws IOException, IllegalStateException { LOG.debug("Ending the node"); - writer.writeByte(NodeTypes.END_NODE); + output.writeByte(NodeTypes.END_NODE); } @Override public void close() throws IOException { - writer.close(); + flush(); } @Override public void flush() throws IOException { - writer.flush(); + if (output instanceof OutputStream) { + ((OutputStream)output).flush(); + } } private void startNode(final QName qName, byte nodeType) throws IOException { Preconditions.checkNotNull(qName, "QName of node identifier should not be null."); + + writeSignatureMarkerAndVersionIfNeeded(); + // First write the type of node - writer.writeByte(nodeType); + output.writeByte(nodeType); // Write Start Tag writeQName(qName); } @@ -191,22 +233,23 @@ public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWri private void writeCodedString(String key) throws IOException { Integer value = stringCodeMap.get(key); - if(value != null) { - writer.writeBoolean(true); - writer.writeInt(value); + output.writeByte(IS_CODE_VALUE); + output.writeInt(value); } else { if(key != null) { + output.writeByte(IS_STRING_VALUE); stringCodeMap.put(key, Integer.valueOf(stringCodeMap.size())); + output.writeUTF(key); + } else { + output.writeByte(IS_NULL_VALUE); } - writer.writeBoolean(false); - writer.writeUTF(key); } } - private void writeObjSet(Set set) throws IOException { + private void writeObjSet(Set set) throws IOException { if(!set.isEmpty()){ - writer.writeInt(set.size()); + output.writeInt(set.size()); for(Object o : set){ if(o instanceof String){ writeCodedString(o.toString()); @@ -216,25 +259,29 @@ public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWri } } } else { - writer.writeInt(0); + output.writeInt(0); } } - private void writeYangInstanceIdentifier(YangInstanceIdentifier identifier) throws IOException { - Iterable pathArguments = identifier.getPathArguments(); - int size = Iterables.size(pathArguments); - writer.writeInt(size); + public void writeYangInstanceIdentifier(YangInstanceIdentifier identifier) throws IOException { + writeSignatureMarkerAndVersionIfNeeded(); + writeYangInstanceIdentifierInternal(identifier); + } + + private void writeYangInstanceIdentifierInternal(YangInstanceIdentifier identifier) throws IOException { + Collection pathArguments = identifier.getPathArguments(); + output.writeInt(pathArguments.size()); for(YangInstanceIdentifier.PathArgument pathArgument : pathArguments) { writePathArgument(pathArgument); } } - private void writePathArgument(YangInstanceIdentifier.PathArgument pathArgument) throws IOException { + public void writePathArgument(YangInstanceIdentifier.PathArgument pathArgument) throws IOException { byte type = PathArgumentTypes.getSerializablePathArgumentType(pathArgument); - writer.writeByte(type); + output.writeByte(type); switch(type) { case PathArgumentTypes.NODE_IDENTIFIER : @@ -278,7 +325,7 @@ public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWri private void writeKeyValueMap(Map keyValueMap) throws IOException { if(keyValueMap != null && !keyValueMap.isEmpty()) { - writer.writeInt(keyValueMap.size()); + output.writeInt(keyValueMap.size()); Set qNameSet = keyValueMap.keySet(); for(QName qName : qNameSet) { @@ -286,20 +333,20 @@ public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWri writeObject(keyValueMap.get(qName)); } } else { - writer.writeInt(0); + output.writeInt(0); } } private void writeQNameSet(Set 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()); + output.writeInt(children.size()); for(QName qName : children) { writeQName(qName); } } else { LOG.debug("augmentation node does not have any child"); - writer.writeInt(0); + output.writeInt(0); } } @@ -307,35 +354,47 @@ public class NormalizedNodeOutputStreamWriter implements NormalizedNodeStreamWri byte type = ValueTypes.getSerializableType(value); // Write object type first - writer.writeByte(type); + output.writeByte(type); switch(type) { case ValueTypes.BOOL_TYPE: - writer.writeBoolean((Boolean) value); + output.writeBoolean((Boolean) value); break; case ValueTypes.QNAME_TYPE: writeQName((QName) value); break; case ValueTypes.INT_TYPE: - writer.writeInt((Integer) value); + output.writeInt((Integer) value); break; case ValueTypes.BYTE_TYPE: - writer.writeByte((Byte) value); + output.writeByte((Byte) value); break; case ValueTypes.LONG_TYPE: - writer.writeLong((Long) value); + output.writeLong((Long) value); break; case ValueTypes.SHORT_TYPE: - writer.writeShort((Short) value); + output.writeShort((Short) value); break; case ValueTypes.BITS_TYPE: - writeObjSet((Set) value); + writeObjSet((Set) value); + break; + case ValueTypes.BINARY_TYPE: + byte[] bytes = (byte[]) value; + output.writeInt(bytes.length); + output.write(bytes); break; case ValueTypes.YANG_IDENTIFIER_TYPE: - writeYangInstanceIdentifier((YangInstanceIdentifier) value); + writeYangInstanceIdentifierInternal((YangInstanceIdentifier) value); + break; + case ValueTypes.NULL_TYPE : + break; + case ValueTypes.STRING_BYTES_TYPE: + final byte[] valueBytes = value.toString().getBytes(StandardCharsets.UTF_8); + output.writeInt(valueBytes.length); + output.write(valueBytes); break; default: - writer.writeUTF(value.toString()); + output.writeUTF(value.toString()); break; } }