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=221e874fd507b6ce1e52016ae09f33d9a0c5d514;hp=8365328669587b5e083c8e53ca819eb63857cd40;hb=1af23ad8e572896323ed5d1c723d977a16eb7e6d;hpb=9d50e2ca76cf1a5eb0636d4a5a704b49ec4a8a25

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 8365328669..221e874fd5 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
@@ -9,63 +9,186 @@
 package org.opendaylight.controller.cluster.datastore;
 
 import akka.actor.ActorRef;
+import akka.actor.ActorSelection;
+import akka.actor.Props;
 import akka.event.Logging;
 import akka.event.LoggingAdapter;
+import akka.japi.Creator;
+import akka.persistence.Persistent;
 import akka.persistence.UntypedProcessor;
+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.NonPersistent;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
+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;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 
 /**
- * A Shard represents a portion of the logical data tree
- * <p/>
+ * A Shard represents a portion of the logical data tree <br/>
+ * <p>
  * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
- *
+ * </p>
  */
 public class Shard extends UntypedProcessor {
 
-  ListeningExecutorService storeExecutor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2));
+    public static final String DEFAULT_NAME = "default";
+
+    private final ListeningExecutorService storeExecutor =
+        MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2));
+
+    private final InMemoryDOMDataStore store;
+
+    private final Map<Modification, DOMStoreThreePhaseCommitCohort>
+        modificationToCohort = new HashMap<>();
+
+    private final LoggingAdapter log =
+        Logging.getLogger(getContext().system(), this);
+
+    // By default persistent will be true and can be turned off using the system
+    // property persistent
+    private final boolean persistent;
+
+    private Shard(String name) {
+
+        String setting = System.getProperty("shard.persistent");
+        this.persistent = !"false".equals(setting);
+
+        log.info("Creating shard : {} persistent : {}", name , persistent);
+
+        store = new InMemoryDOMDataStore(name, storeExecutor);
+    }
+
+    public static Props props(final String name) {
+        return Props.create(new Creator<Shard>() {
+
+            @Override
+            public Shard create() throws Exception {
+                return new Shard(name);
+            }
+
+        });
+    }
+
+
+    @Override
+    public void onReceive(Object message) throws Exception {
+        log.debug("Received message {}", message);
+
+        if (message instanceof CreateTransactionChain) {
+            createTransactionChain();
+        } else if (message instanceof RegisterChangeListener) {
+            registerChangeListener((RegisterChangeListener) message);
+        } else if (message instanceof UpdateSchemaContext) {
+            updateSchemaContext((UpdateSchemaContext) message);
+        } else if (message instanceof ForwardedCommitTransaction) {
+            handleForwardedCommit((ForwardedCommitTransaction) message);
+        } else if (message instanceof Persistent) {
+            commit((Modification) ((Persistent) message).payload());
+        } else if (message instanceof CreateTransaction) {
+            createTransaction();
+        } else if(message instanceof NonPersistent){
+            commit((Modification) ((NonPersistent) message).payload());
+        }
+    }
+
+    private void createTransaction() {
+        DOMStoreReadWriteTransaction transaction =
+            store.newReadWriteTransaction();
+        ActorRef transactionActor = getContext().actorOf(
+            ShardTransaction.props(transaction, getSelf()));
+        getSender()
+            .tell(new CreateTransactionReply(transactionActor.path()),
+                getSelf());
+    }
 
-  private final InMemoryDOMDataStore store = new InMemoryDOMDataStore("OPER", storeExecutor);
+    private void commit(Modification modification) {
+        DOMStoreThreePhaseCommitCohort cohort =
+            modificationToCohort.remove(modification);
+        if (cohort == null) {
+            log.error(
+                "Could not find cohort for modification : " + modification);
+            return;
+        }
+        final ListenableFuture<Void> future = cohort.commit();
+        final ActorRef sender = getSender();
+        final ActorRef self = getSelf();
+        future.addListener(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    future.get();
+                    sender.tell(new CommitTransactionReply(), self);
+                } catch (InterruptedException | ExecutionException e) {
+                    log.error(e, "An exception happened when committing");
+                }
+            }
+        }, getContext().dispatcher());
+    }
 
-  LoggingAdapter log = Logging.getLogger(getContext().system(), this);
+    private void handleForwardedCommit(ForwardedCommitTransaction message) {
+        log.info("received forwarded transaction");
+        modificationToCohort
+            .put(message.getModification(), message.getCohort());
+        if(persistent) {
+            getSelf().forward(Persistent.create(message.getModification()),
+                getContext());
+        } else {
+            getSelf().forward(NonPersistent.create(message.getModification()),
+                getContext());
+        }
+    }
+
+    private void updateSchemaContext(UpdateSchemaContext message) {
+        store.onGlobalContextUpdated(message.getSchemaContext());
+    }
+
+    private void registerChangeListener(
+        RegisterChangeListener registerChangeListener) {
+
+        ActorSelection dataChangeListenerPath = getContext()
+            .system().actorSelection(registerChangeListener.getDataChangeListenerPath());
+
+        AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>
+            listener = new DataChangeListenerProxy(dataChangeListenerPath);
+
+        org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>>
+            registration =
+            store.registerChangeListener(registerChangeListener.getPath(),
+                listener, registerChangeListener.getScope());
+        ActorRef listenerRegistration =
+            getContext().actorOf(
+                DataChangeListenerRegistration.props(registration));
+        getSender()
+            .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
+                getSelf());
+    }
 
-  @Override
-  public void onReceive(Object message) throws Exception {
-    if (message instanceof CreateTransactionChain) {
-      createTransactionChain();
-    } else if(message instanceof RegisterChangeListener){
-      registerChangeListener((RegisterChangeListener) message);
-    } else if(message instanceof UpdateSchemaContext){
-      updateSchemaContext((UpdateSchemaContext) message);
+    private void createTransactionChain() {
+        DOMStoreTransactionChain chain = store.createTransactionChain();
+        ActorRef transactionChain =
+            getContext().actorOf(ShardTransactionChain.props(chain));
+        getSender()
+            .tell(new CreateTransactionChainReply(transactionChain.path()),
+                getSelf());
     }
-  }
-
-  private void updateSchemaContext(UpdateSchemaContext message) {
-    store.onGlobalContextUpdated(message.getSchemaContext());
-  }
-
-  private void registerChangeListener(RegisterChangeListener registerChangeListener) {
-    org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>> registration =
-            store.registerChangeListener(registerChangeListener.getPath(), registerChangeListener.getListener(), registerChangeListener.getScope());
-    ActorRef listenerRegistration = getContext().actorOf(ListenerRegistration.props(registration));
-    getSender().tell(new RegisterChangeListenerReply(listenerRegistration.path()), getSelf());
-  }
-
-  private void createTransactionChain() {
-    DOMStoreTransactionChain chain = store.createTransactionChain();
-    ActorRef transactionChain = getContext().actorOf(ShardTransactionChain.props(chain));
-    getSender().tell(new CreateTransactionChainReply(transactionChain.path()), getSelf());
-  }
 }