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=bf9f8d803ac00f0e8ce2cbe983f81bfb9e9a44d0;hb=ef625b47256d806acce74b2490c3dd75b21d417e;hp=3f2125746300f5350ae42be63d10e6b761840a2e;hpb=4ed6793fcad89fcb4a3c0f7ec230753cb7ed31a9;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..bf9f8d803a 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,21 @@ 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.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 +30,48 @@ import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeWrit * @author Thomas Pantelis */ public final class SerializationUtils { + public static ThreadLocal REUSABLE_WRITER_TL = new ThreadLocal<>(); + public static ThreadLocal REUSABLE_READER_TL = new ThreadLocal<>(); public static interface Applier { void apply(T instance, YangInstanceIdentifier path, NormalizedNode node); } + private static NormalizedNodeOutputStreamWriter streamWriter(DataOutput out) throws IOException { + NormalizedNodeOutputStreamWriter streamWriter = REUSABLE_WRITER_TL.get(); + if(streamWriter == null) { + streamWriter = new NormalizedNodeOutputStreamWriter(out); + } + + return streamWriter; + } + + private static NormalizedNodeInputStreamReader streamReader(DataInput in) throws IOException { + NormalizedNodeInputStreamReader streamWriter = REUSABLE_READER_TL.get(); + if(streamWriter == null) { + streamWriter = new NormalizedNodeInputStreamReader(in); + } + + return streamWriter; + } + 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); + NormalizedNodeOutputStreamWriter 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); + NormalizedNodeInputStreamReader streamReader = streamReader(in); NormalizedNode node = streamReader.readNormalizedNode(); YangInstanceIdentifier path = streamReader.readYangInstanceIdentifier(); applier.apply(instance, path, node); @@ -57,42 +84,71 @@ public final class SerializationUtils { try { out.writeBoolean(node != null); if(node != null) { - NormalizedNodeOutputStreamWriter streamWriter = new NormalizedNodeOutputStreamWriter(out); - NormalizedNodeWriter.forStreamWriter(streamWriter).write(node); + NormalizedNodeOutputStreamWriter 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) { + NormalizedNodeInputStreamReader streamReader = streamReader(in); + return streamReader.readNormalizedNode(); + } + + return null; + } + + public static NormalizedNode deserializeNormalizedNode(byte [] bytes) { + NormalizedNode node = null; + try { + node = 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); + node = 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; + return node; + } + + 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); + NormalizedNodeOutputStreamWriter 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); + NormalizedNodeInputStreamReader streamReader = streamReader(in); return streamReader.readYangInstanceIdentifier(); } catch (IOException e) { throw new IllegalArgumentException("Error deserializing path", e);