Implement TransactionProxy#read

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / TransactionProxy.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java
index 609dea0b360819ebc576e636e2237b8cb9868172..c3c7e7c00cb0d6ee4077042fa11fb548c2d74e1f 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java
@@ -8,13 +8,25 @@
 
 package org.opendaylight.controller.cluster.datastore;
 
+import akka.actor.ActorSelection;
 import com.google.common.base.Optional;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListenableFutureTask;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.ReadData;
+import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
+import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
 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.Callable;
+import java.util.concurrent.Executors;
+
 /**
  * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
  *
@@ -27,9 +39,58 @@ import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  *
  */
 public class TransactionProxy implements DOMStoreReadWriteTransaction {
+
+    public enum TransactionType {
+        READ_ONLY,
+        WRITE_ONLY,
+        READ_WRITE
+    }
+
+    private final TransactionType readOnly;
+    private final ActorContext actorContext;
+    private final Map<String, ActorSelection> remoteTransactionPaths = new HashMap<>();
+
+    public TransactionProxy(
+        ActorContext actorContext,
+        TransactionType readOnly) {
+
+        this.readOnly = readOnly;
+        this.actorContext = actorContext;
+
+        Object response = actorContext.executeShardOperation(Shard.DEFAULT_NAME, new CreateTransaction(), ActorContext.ASK_DURATION);
+        if(response instanceof CreateTransactionReply){
+            CreateTransactionReply reply = (CreateTransactionReply) response;
+            remoteTransactionPaths.put(Shard.DEFAULT_NAME, actorContext.actorSelection(reply.getTransactionPath()));
+        }
+    }
+
     @Override
-    public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(InstanceIdentifier path) {
-        throw new UnsupportedOperationException("read");
+    public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final InstanceIdentifier path) {
+        final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
+
+        Callable<Optional<NormalizedNode<?,?>>> call = new Callable() {
+
+            @Override public Optional<NormalizedNode<?,?>> call() throws Exception {
+                Object response = actorContext
+                    .executeRemoteOperation(remoteTransaction, new ReadData(path),
+                        ActorContext.ASK_DURATION);
+                if(response instanceof ReadDataReply){
+                    ReadDataReply reply = (ReadDataReply) response;
+                    //FIXME : A cast should not be required here ???
+                    return (Optional<NormalizedNode<?, ?>>) Optional.of(reply.getNormalizedNode());
+                }
+
+                return Optional.absent();
+            }
+        };
+
+        ListenableFutureTask<Optional<NormalizedNode<?, ?>>>
+            future = ListenableFutureTask.create(call);
+
+        //FIXME : Use a thread pool here
+        Executors.newSingleThreadExecutor().submit(future);
+
+        return future;
     }
 
     @Override
@@ -61,4 +122,13 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
     public void close() {
         throw new UnsupportedOperationException("close");
     }
+
+    private ActorSelection remoteTransactionFromIdentifier(InstanceIdentifier path){
+        String shardName = shardNameFromIdentifier(path);
+        return remoteTransactionPaths.get(shardName);
+    }
+
+    private String shardNameFromIdentifier(InstanceIdentifier path){
+        return Shard.DEFAULT_NAME;
+    }
 }