Merge "Bug 1027: Sort map of instance-identifiers and data by length."
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / TransactionProxy.java
1 /*
2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
3  *
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
7  */
8
9 package org.opendaylight.controller.cluster.datastore;
10
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;
31
32 import java.util.ArrayList;
33 import java.util.HashMap;
34 import java.util.List;
35 import java.util.Map;
36 import java.util.concurrent.Callable;
37 import java.util.concurrent.Executors;
38 import java.util.concurrent.atomic.AtomicLong;
39
40 /**
41  * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
42  * <p>
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
46  *</p>
47  * <p>
48  * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
49  * shards will be executed.
50  * </p>
51  */
52 public class TransactionProxy implements DOMStoreReadWriteTransaction {
53
54     public enum TransactionType {
55         READ_ONLY,
56         WRITE_ONLY,
57         READ_WRITE
58     }
59
60     private static final AtomicLong counter = new AtomicLong();
61
62     private final TransactionType readOnly;
63     private final ActorContext actorContext;
64     private final Map<String, ActorSelection> remoteTransactionPaths = new HashMap<>();
65     private final String identifier;
66
67     public TransactionProxy(
68         ActorContext actorContext,
69         TransactionType readOnly) {
70
71         this.identifier = "transaction-" + counter.getAndIncrement();
72         this.readOnly = readOnly;
73         this.actorContext = actorContext;
74
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()));
79         }
80     }
81
82     @Override
83     public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final InstanceIdentifier path) {
84         final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
85
86         Callable<Optional<NormalizedNode<?,?>>> call = new Callable() {
87
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                     //FIXME : A cast should not be required here ???
95                     return (Optional<NormalizedNode<?, ?>>) Optional.of(reply.getNormalizedNode());
96                 }
97
98                 return Optional.absent();
99             }
100         };
101
102         ListenableFutureTask<Optional<NormalizedNode<?, ?>>>
103             future = ListenableFutureTask.create(call);
104
105         //FIXME : Use a thread pool here
106         Executors.newSingleThreadExecutor().submit(future);
107
108         return future;
109     }
110
111     @Override
112     public void write(InstanceIdentifier path, NormalizedNode<?, ?> data) {
113         final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
114         remoteTransaction.tell(new WriteData(path, data), null);
115     }
116
117     @Override
118     public void merge(InstanceIdentifier path, NormalizedNode<?, ?> data) {
119         final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
120         remoteTransaction.tell(new MergeData(path, data), null);
121     }
122
123     @Override
124     public void delete(InstanceIdentifier path) {
125         final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
126         remoteTransaction.tell(new DeleteData(path), null);
127     }
128
129     @Override
130     public DOMStoreThreePhaseCommitCohort ready() {
131         List<ActorPath> cohortPaths = new ArrayList<>();
132
133         for(ActorSelection remoteTransaction : remoteTransactionPaths.values()) {
134             Object result = actorContext.executeRemoteOperation(remoteTransaction,
135                 new ReadyTransaction(),
136                 ActorContext.ASK_DURATION
137             );
138
139             if(result instanceof ReadyTransactionReply){
140                 ReadyTransactionReply reply = (ReadyTransactionReply) result;
141                 cohortPaths.add(reply.getCohortPath());
142             }
143         }
144
145         return new ThreePhaseCommitCohortProxy(cohortPaths);
146     }
147
148     @Override
149     public Object getIdentifier() {
150         return this.identifier;
151     }
152
153     @Override
154     public void close() {
155         for(ActorSelection remoteTransaction : remoteTransactionPaths.values()) {
156             remoteTransaction.tell(new CloseTransaction(), null);
157         }
158     }
159
160     private ActorSelection remoteTransactionFromIdentifier(InstanceIdentifier path){
161         String shardName = shardNameFromIdentifier(path);
162         return remoteTransactionPaths.get(shardName);
163     }
164
165     private String shardNameFromIdentifier(InstanceIdentifier path){
166         return Shard.DEFAULT_NAME;
167     }
168 }