X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShard.java;h=09ad00598fedb0730cb520740d4f4588a058446e;hp=5b4f7ef8989711dffdf4cdd8fbe7d483165f23a7;hb=18a4539ad844c05fcd30373efa43f873aca4c142;hpb=9c0508d8b591e356f145d5d1a277c10965a647bb

diff --git a/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
index 5b4f7ef898..09ad00598f 100644
--- a/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
@@ -20,8 +20,10 @@ import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
@@ -29,6 +31,7 @@ import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContex
 import org.opendaylight.controller.cluster.datastore.modification.Modification;
 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
@@ -61,6 +64,8 @@ public class Shard extends UntypedProcessor {
         Logging.getLogger(getContext().system(), this);
 
     private Shard(String name) {
+        log.info("Creating shard : {}", name );
+
         store = new InMemoryDOMDataStore(name, storeExecutor);
     }
 
@@ -77,6 +82,8 @@ public class Shard extends UntypedProcessor {
 
     @Override
     public void onReceive(Object message) throws Exception {
+        log.debug("Received message {}", message);
+
         if (message instanceof CreateTransactionChain) {
             createTransactionChain();
         } else if (message instanceof RegisterChangeListener) {
@@ -87,9 +94,21 @@ public class Shard extends UntypedProcessor {
             handleForwardedCommit((ForwardedCommitTransaction) message);
         } else if (message instanceof Persistent) {
             commit((Persistent) message);
+        } else if (message instanceof CreateTransaction) {
+            createTransaction();
         }
     }
 
+    private void createTransaction() {
+        DOMStoreReadWriteTransaction transaction =
+            store.newReadWriteTransaction();
+        ActorRef transactionActor = getContext().actorOf(
+            ShardTransaction.props(transaction, getSelf()));
+        getSender()
+            .tell(new CreateTransactionReply(transactionActor.path()),
+                getSelf());
+    }
+
     private void commit(Persistent message) {
         Modification modification = (Modification) message.payload();
         DOMStoreThreePhaseCommitCohort cohort =
@@ -130,18 +149,19 @@ public class Shard extends UntypedProcessor {
     private void registerChangeListener(
         RegisterChangeListener registerChangeListener) {
 
-        ActorSelection listenerRegistrationActor = getContext()
+        ActorSelection dataChangeListenerPath = getContext()
             .system().actorSelection(registerChangeListener.getDataChangeListenerPath());
 
         AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>
-            listener = new ListenerProxy(listenerRegistrationActor);
+            listener = new DataChangeListenerProxy(dataChangeListenerPath);
 
         org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>>
             registration =
             store.registerChangeListener(registerChangeListener.getPath(),
                 listener, registerChangeListener.getScope());
         ActorRef listenerRegistration =
-            getContext().actorOf(ListenerRegistration.props(registration));
+            getContext().actorOf(
+                DataChangeListenerRegistration.props(registration));
         getSender()
             .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
                 getSelf());