2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
9 package org.opendaylight.controller.cluster.datastore;
11 import akka.actor.ActorSelection;
12 import com.google.common.base.Optional;
13 import com.google.common.util.concurrent.ListenableFuture;
14 import com.google.common.util.concurrent.ListenableFutureTask;
15 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
16 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
17 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
18 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
19 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
20 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
21 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
22 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
23 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
25 import java.util.HashMap;
27 import java.util.concurrent.Callable;
28 import java.util.concurrent.Executors;
31 * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
33 * Creating a transaction on the consumer side will create one instance of a transaction proxy. If during
34 * the transaction reads and writes are done on data that belongs to different shards then a separate transaction will
35 * be created on each of those shards by the TransactionProxy
37 * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
38 * shards will be executed.
41 public class TransactionProxy implements DOMStoreReadWriteTransaction {
43 public enum TransactionType {
49 private final TransactionType readOnly;
50 private final ActorContext actorContext;
51 private final Map<String, ActorSelection> remoteTransactionPaths = new HashMap<>();
53 public TransactionProxy(
54 ActorContext actorContext,
55 TransactionType readOnly) {
57 this.readOnly = readOnly;
58 this.actorContext = actorContext;
60 Object response = actorContext.executeShardOperation(Shard.DEFAULT_NAME, new CreateTransaction(), ActorContext.ASK_DURATION);
61 if(response instanceof CreateTransactionReply){
62 CreateTransactionReply reply = (CreateTransactionReply) response;
63 remoteTransactionPaths.put(Shard.DEFAULT_NAME, actorContext.actorSelection(reply.getTransactionPath()));
68 public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final InstanceIdentifier path) {
69 final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
71 Callable<Optional<NormalizedNode<?,?>>> call = new Callable() {
73 @Override public Optional<NormalizedNode<?,?>> call() throws Exception {
74 Object response = actorContext
75 .executeRemoteOperation(remoteTransaction, new ReadData(path),
76 ActorContext.ASK_DURATION);
77 if(response instanceof ReadDataReply){
78 ReadDataReply reply = (ReadDataReply) response;
79 //FIXME : A cast should not be required here ???
80 return (Optional<NormalizedNode<?, ?>>) Optional.of(reply.getNormalizedNode());
83 return Optional.absent();
87 ListenableFutureTask<Optional<NormalizedNode<?, ?>>>
88 future = ListenableFutureTask.create(call);
90 //FIXME : Use a thread pool here
91 Executors.newSingleThreadExecutor().submit(future);
97 public void write(InstanceIdentifier path, NormalizedNode<?, ?> data) {
98 throw new UnsupportedOperationException("write");
102 public void merge(InstanceIdentifier path, NormalizedNode<?, ?> data) {
103 throw new UnsupportedOperationException("merge");
107 public void delete(InstanceIdentifier path) {
108 throw new UnsupportedOperationException("delete");
112 public DOMStoreThreePhaseCommitCohort ready() {
113 throw new UnsupportedOperationException("ready");
117 public Object getIdentifier() {
118 throw new UnsupportedOperationException("getIdentifier");
122 public void close() {
123 throw new UnsupportedOperationException("close");
126 private ActorSelection remoteTransactionFromIdentifier(InstanceIdentifier path){
127 String shardName = shardNameFromIdentifier(path);
128 return remoteTransactionPaths.get(shardName);
131 private String shardNameFromIdentifier(InstanceIdentifier path){
132 return Shard.DEFAULT_NAME;