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.ActorPath;
12 import akka.actor.ActorSelection;
13 import com.google.common.base.Optional;
14 import com.google.common.util.concurrent.ListenableFuture;
15 import com.google.common.util.concurrent.ListenableFutureTask;
16 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
17 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
18 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
19 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
20 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
21 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
22 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
23 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
24 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
25 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
26 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
27 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
28 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
29 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
30 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
32 import java.util.ArrayList;
33 import java.util.HashMap;
34 import java.util.List;
36 import java.util.concurrent.Callable;
37 import java.util.concurrent.Executors;
38 import java.util.concurrent.atomic.AtomicLong;
41 * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
43 * Creating a transaction on the consumer side will create one instance of a transaction proxy. If during
44 * the transaction reads and writes are done on data that belongs to different shards then a separate transaction will
45 * be created on each of those shards by the TransactionProxy
48 * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
49 * shards will be executed.
52 public class TransactionProxy implements DOMStoreReadWriteTransaction {
54 public enum TransactionType {
60 private static final AtomicLong counter = new AtomicLong();
62 private final TransactionType transactionType;
63 private final ActorContext actorContext;
64 private final Map<String, ActorSelection> remoteTransactionPaths = new HashMap<>();
65 private final String identifier;
67 public TransactionProxy(
68 ActorContext actorContext,
69 TransactionType transactionType) {
71 this.identifier = "transaction-" + counter.getAndIncrement();
72 this.transactionType = transactionType;
73 this.actorContext = actorContext;
75 Object response = actorContext.executeShardOperation(Shard.DEFAULT_NAME, new CreateTransaction(), ActorContext.ASK_DURATION);
76 if(response instanceof CreateTransactionReply){
77 CreateTransactionReply reply = (CreateTransactionReply) response;
78 remoteTransactionPaths.put(Shard.DEFAULT_NAME, actorContext.actorSelection(reply.getTransactionPath()));
83 public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final InstanceIdentifier path) {
84 final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
86 Callable<Optional<NormalizedNode<?,?>>> call = new Callable() {
88 @Override public Optional<NormalizedNode<?,?>> call() throws Exception {
89 Object response = actorContext
90 .executeRemoteOperation(remoteTransaction, new ReadData(path),
91 ActorContext.ASK_DURATION);
92 if(response instanceof ReadDataReply){
93 ReadDataReply reply = (ReadDataReply) response;
94 if(reply.getNormalizedNode() == null){
95 return Optional.absent();
97 //FIXME : A cast should not be required here ???
98 return (Optional<NormalizedNode<?, ?>>) Optional.of(reply.getNormalizedNode());
101 return Optional.absent();
105 ListenableFutureTask<Optional<NormalizedNode<?, ?>>>
106 future = ListenableFutureTask.create(call);
108 //FIXME : Use a thread pool here
109 Executors.newSingleThreadExecutor().submit(future);
115 public void write(InstanceIdentifier path, NormalizedNode<?, ?> data) {
116 final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
117 remoteTransaction.tell(new WriteData(path, data), null);
121 public void merge(InstanceIdentifier path, NormalizedNode<?, ?> data) {
122 final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
123 remoteTransaction.tell(new MergeData(path, data), null);
127 public void delete(InstanceIdentifier path) {
128 final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
129 remoteTransaction.tell(new DeleteData(path), null);
133 public DOMStoreThreePhaseCommitCohort ready() {
134 List<ActorPath> cohortPaths = new ArrayList<>();
136 for(ActorSelection remoteTransaction : remoteTransactionPaths.values()) {
137 Object result = actorContext.executeRemoteOperation(remoteTransaction,
138 new ReadyTransaction(),
139 ActorContext.ASK_DURATION
142 if(result instanceof ReadyTransactionReply){
143 ReadyTransactionReply reply = (ReadyTransactionReply) result;
144 cohortPaths.add(reply.getCohortPath());
148 return new ThreePhaseCommitCohortProxy(actorContext, cohortPaths);
152 public Object getIdentifier() {
153 return this.identifier;
157 public void close() {
158 for(ActorSelection remoteTransaction : remoteTransactionPaths.values()) {
159 remoteTransaction.tell(new CloseTransaction(), null);
163 private ActorSelection remoteTransactionFromIdentifier(InstanceIdentifier path){
164 String shardName = shardNameFromIdentifier(path);
165 return remoteTransactionPaths.get(shardName);
168 private String shardNameFromIdentifier(InstanceIdentifier path){
169 return Shard.DEFAULT_NAME;