Merge "Bug 2265: Use new NormalizedNode streaming in messages"

diff --combined opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java
index f8bcb528b3e987ce36784725befdd91b4fd020e4,1661bb4b5dc95477f9e10c413171bc8d5ba03fb8..a3ef0339b7571172dfb0d5b2338f52911a86cf9c
--- 1/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java
--- 2/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java
@@@ -95,12 -95,9 +95,10 @@@ import scala.concurrent.duration.Finite
   */
  public class Shard extends RaftActor {
  
-     private static final Object COMMIT_TRANSACTION_REPLY = new CommitTransactionReply().toSerializable();
- 
      private static final Object TX_COMMIT_TIMEOUT_CHECK_MESSAGE = "txCommitTimeoutCheck";
  
 -    public static final String DEFAULT_NAME = "default";
 +    @VisibleForTesting
 +    static final String DEFAULT_NAME = "default";
  
      // The state of this Shard
      private final InMemoryDOMDataStore store;
@@@ -355,7 -352,7 +353,7 @@@
              cohortEntry = commitCoordinator.getAndRemoveCohortEntry(transactionID);
              if(cohortEntry != null) {
                  commitWithNewTransaction(cohortEntry.getModification());
-                 sender.tell(COMMIT_TRANSACTION_REPLY, getSelf());
+                 sender.tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf());
              } else {
                  // This really shouldn't happen - it likely means that persistence or replication
                  // took so long to complete such that the cohort entry was expired from the cache.
@@@ -377,7 -374,7 +375,7 @@@
              // currently uses a same thread executor anyway.
              cohortEntry.getCohort().commit().get();
  
-             sender.tell(COMMIT_TRANSACTION_REPLY, getSelf());
+             sender.tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf());
  
              shardMBean.incrementCommittedTransactionCount();
              shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis());
@@@ -413,7 -410,7 +411,7 @@@
          // transactionId so to maintain backwards compatibility, we create a separate cohort actor
          // to provide the compatible behavior.
          ActorRef replyActorPath = self();
-         if(ready.getTxnClientVersion() < CreateTransaction.HELIUM_1_VERSION) {
+         if(ready.getTxnClientVersion() < DataStoreVersions.HELIUM_1_VERSION) {
              LOG.debug("Creating BackwardsCompatibleThreePhaseCommitCohort");
              replyActorPath = getContext().actorOf(BackwardsCompatibleThreePhaseCommitCohort.props(
                      ready.getTransactionID()));
@@@ -448,7 -445,7 +446,7 @@@
                      shardMBean.incrementAbortTransactionsCount();
  
                      if(sender != null) {
-                         sender.tell(new AbortTransactionReply().toSerializable(), self);
+                         sender.tell(AbortTransactionReply.INSTANCE.toSerializable(), self);
                      }
                  }
  
@@@ -481,7 -478,7 +479,7 @@@
          // This must be for install snapshot. Don't want to open this up and trigger
          // deSerialization
  
-         self().tell(new CaptureSnapshotReply(ReadDataReply.getNormalizedNodeByteString(message)),
+         self().tell(new CaptureSnapshotReply(ReadDataReply.fromSerializableAsByteString(message)),
                  self());
  
          createSnapshotTransaction = null;
@@@ -501,7 -498,8 +499,8 @@@
      }
  
      private ActorRef createTypedTransactionActor(int transactionType,
-             ShardTransactionIdentifier transactionId, String transactionChainId, int clientVersion ) {
+             ShardTransactionIdentifier transactionId, String transactionChainId,
+             short clientVersion ) {
  
          DOMStoreTransactionFactory factory = store;
  
@@@ -569,7 -567,7 +568,7 @@@
      }
  
      private ActorRef createTransaction(int transactionType, String remoteTransactionId,
-             String transactionChainId, int clientVersion) {
+             String transactionChainId, short clientVersion) {
  
          ShardTransactionIdentifier transactionId =
              ShardTransactionIdentifier.builder()
@@@ -660,7 -658,7 +659,7 @@@
          dataChangeListeners.add(dataChangeListenerPath);
  
          AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener =
-                 new DataChangeListenerProxy(schemaContext, dataChangeListenerPath);
+                 new DataChangeListenerProxy(dataChangeListenerPath);
  
          LOG.debug("Registering for path {}", registerChangeListener.getPath());
  
@@@ -819,7 -817,7 +818,7 @@@
              createSnapshotTransaction = createTransaction(
                  TransactionProxy.TransactionType.READ_ONLY.ordinal(),
                  "createSnapshot" + ++createSnapshotTransactionCounter, "",
-                 CreateTransaction.CURRENT_VERSION);
+                 DataStoreVersions.CURRENT_VERSION);
  
              createSnapshotTransaction.tell(
                  new ReadData(YangInstanceIdentifier.builder().build()).toSerializable(), self());