X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Futils%2FSerializationUtils.java;h=97e3383490d5bb058d3999c454464c435d16ee0c;hb=dea3effede98cfb561c44d66b24c2d71a44b10a3;hp=3f2125746300f5350ae42be63d10e6b761840a2e;hpb=e0e940ce65e9ae8845b94da03fa559c5980a2876;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/SerializationUtils.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/SerializationUtils.java index 3f21257463..97e3383490 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/SerializationUtils.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/SerializationUtils.java @@ -8,14 +8,23 @@ package org.opendaylight.controller.cluster.datastore.utils; import com.google.common.base.Preconditions; +import com.google.protobuf.InvalidProtocolBufferException; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; import java.io.DataInput; +import java.io.DataInputStream; import java.io.DataOutput; +import java.io.DataOutputStream; import java.io.IOException; +import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec; +import org.opendaylight.controller.cluster.datastore.node.utils.stream.InvalidNormalizedNodeStreamException; +import org.opendaylight.controller.cluster.datastore.node.utils.stream.NormalizedNodeDataInput; +import org.opendaylight.controller.cluster.datastore.node.utils.stream.NormalizedNodeDataOutput; +import org.opendaylight.controller.cluster.datastore.node.utils.stream.NormalizedNodeInputOutput; import org.opendaylight.controller.cluster.datastore.node.utils.stream.NormalizedNodeInputStreamReader; -import org.opendaylight.controller.cluster.datastore.node.utils.stream.NormalizedNodeOutputStreamWriter; +import org.opendaylight.controller.protobuff.messages.common.NormalizedNodeMessages; 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.NormalizedNodeWriter; /** * Provides various utility methods for serialization and de-serialization. @@ -23,28 +32,48 @@ import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeWrit * @author Thomas Pantelis */ public final class SerializationUtils { + public static final ThreadLocal REUSABLE_WRITER_TL = new ThreadLocal<>(); + public static final ThreadLocal REUSABLE_READER_TL = new ThreadLocal<>(); - public static interface Applier { + public interface Applier { void apply(T instance, YangInstanceIdentifier path, NormalizedNode node); } + private static NormalizedNodeDataOutput streamWriter(DataOutput out) throws IOException { + NormalizedNodeDataOutput streamWriter = REUSABLE_WRITER_TL.get(); + if(streamWriter == null) { + streamWriter = NormalizedNodeInputOutput.newDataOutput(out); + } + + return streamWriter; + } + + private static NormalizedNodeDataInput streamReader(DataInput in) throws IOException { + NormalizedNodeDataInput streamReader = REUSABLE_READER_TL.get(); + if(streamReader == null) { + streamReader = new NormalizedNodeInputStreamReader(in); + } + + return streamReader; + } + public static void serializePathAndNode(YangInstanceIdentifier path, NormalizedNode node, DataOutput out) { Preconditions.checkNotNull(path); Preconditions.checkNotNull(node); try { - NormalizedNodeOutputStreamWriter streamWriter = new NormalizedNodeOutputStreamWriter(out); - NormalizedNodeWriter.forStreamWriter(streamWriter).write(node); + NormalizedNodeDataOutput streamWriter = streamWriter(out); + streamWriter.writeNormalizedNode(node); streamWriter.writeYangInstanceIdentifier(path); } catch (IOException e) { - throw new IllegalArgumentException(String.format("Error serializing path {} and Node {}", + throw new IllegalArgumentException(String.format("Error serializing path %s and Node %s", path, node), e); } } public static void deserializePathAndNode(DataInput in, T instance, Applier applier) { try { - NormalizedNodeInputStreamReader streamReader = new NormalizedNodeInputStreamReader(in); + NormalizedNodeDataInput streamReader = streamReader(in); NormalizedNode node = streamReader.readNormalizedNode(); YangInstanceIdentifier path = streamReader.readYangInstanceIdentifier(); applier.apply(instance, path, node); @@ -57,42 +86,68 @@ public final class SerializationUtils { try { out.writeBoolean(node != null); if(node != null) { - NormalizedNodeOutputStreamWriter streamWriter = new NormalizedNodeOutputStreamWriter(out); - NormalizedNodeWriter.forStreamWriter(streamWriter).write(node); + NormalizedNodeDataOutput streamWriter = streamWriter(out); + streamWriter.writeNormalizedNode(node); } } catch (IOException e) { - throw new IllegalArgumentException(String.format("Error serializing NormalizedNode {}", + throw new IllegalArgumentException(String.format("Error serializing NormalizedNode %s", node), e); } } public static NormalizedNode deserializeNormalizedNode(DataInput in) { + try { + return tryDeserializeNormalizedNode(in); + } catch (IOException e) { + throw new IllegalArgumentException("Error deserializing NormalizedNode", e); + } + } + + private static NormalizedNode tryDeserializeNormalizedNode(DataInput in) throws IOException { + boolean present = in.readBoolean(); + if(present) { + NormalizedNodeDataInput streamReader = streamReader(in); + return streamReader.readNormalizedNode(); + } + + return null; + } + + public static NormalizedNode deserializeNormalizedNode(byte [] bytes) { + try { + return tryDeserializeNormalizedNode(new DataInputStream(new ByteArrayInputStream(bytes))); + } catch(InvalidNormalizedNodeStreamException e) { + // Probably from legacy protobuf serialization - try that. try { - boolean present = in.readBoolean(); - if(present) { - NormalizedNodeInputStreamReader streamReader = new NormalizedNodeInputStreamReader(in); - return streamReader.readNormalizedNode(); - } - } catch (IOException e) { + NormalizedNodeMessages.Node serializedNode = NormalizedNodeMessages.Node.parseFrom(bytes); + return new NormalizedNodeToNodeCodec(null).decode(serializedNode); + } catch (InvalidProtocolBufferException e2) { throw new IllegalArgumentException("Error deserializing NormalizedNode", e); } + } catch (IOException e) { + throw new IllegalArgumentException("Error deserializing NormalizedNode", e); + } + } - return null; + public static byte [] serializeNormalizedNode(NormalizedNode node) { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + serializeNormalizedNode(node, new DataOutputStream(bos)); + return bos.toByteArray(); } public static void serializePath(YangInstanceIdentifier path, DataOutput out) { Preconditions.checkNotNull(path); try { - NormalizedNodeOutputStreamWriter streamWriter = new NormalizedNodeOutputStreamWriter(out); + NormalizedNodeDataOutput streamWriter = streamWriter(out); streamWriter.writeYangInstanceIdentifier(path); } catch (IOException e) { - throw new IllegalArgumentException(String.format("Error serializing path {}", path), e); + throw new IllegalArgumentException(String.format("Error serializing path %s", path), e); } } public static YangInstanceIdentifier deserializePath(DataInput in) { try { - NormalizedNodeInputStreamReader streamReader = new NormalizedNodeInputStreamReader(in); + NormalizedNodeDataInput streamReader = streamReader(in); return streamReader.readYangInstanceIdentifier(); } catch (IOException e) { throw new IllegalArgumentException("Error deserializing path", e);