2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
3 * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved.
5 * This program and the accompanying materials are made available under the
6 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
7 * and is available at http://www.eclipse.org/legal/epl-v10.html
11 package org.opendaylight.controller.cluster.datastore;
13 import akka.actor.ActorRef;
14 import akka.actor.PoisonPill;
15 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
16 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
17 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
18 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
19 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
20 import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply;
21 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
22 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
23 import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply;
24 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
25 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
26 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
27 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
28 import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
29 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
30 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
31 import org.opendaylight.controller.cluster.datastore.modification.Modification;
32 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
33 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
34 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
40 public class ShardWriteTransaction extends ShardTransaction {
42 private final MutableCompositeModification compositeModification = new MutableCompositeModification();
43 private int totalBatchedModificationsReceived;
44 private Exception lastBatchedModificationsException;
45 private final ReadWriteShardDataTreeTransaction transaction;
47 public ShardWriteTransaction(ReadWriteShardDataTreeTransaction transaction, ActorRef shardActor,
48 ShardStats shardStats, String transactionID, short clientTxVersion) {
49 super(shardActor, shardStats, transactionID, clientTxVersion);
50 this.transaction = transaction;
54 protected ReadWriteShardDataTreeTransaction getDOMStoreTransaction() {
59 public void handleReceive(Object message) throws Exception {
61 if (message instanceof BatchedModifications) {
62 batchedModifications((BatchedModifications)message);
63 } else if (message instanceof ReadyTransaction) {
64 readyTransaction(!SERIALIZED_REPLY, false);
65 } else if(ReadyTransaction.SERIALIZABLE_CLASS.equals(message.getClass())) {
66 readyTransaction(SERIALIZED_REPLY, false);
67 } else if(WriteData.isSerializedType(message)) {
68 writeData(WriteData.fromSerializable(message), SERIALIZED_REPLY);
70 } else if(MergeData.isSerializedType(message)) {
71 mergeData(MergeData.fromSerializable(message), SERIALIZED_REPLY);
73 } else if(DeleteData.isSerializedType(message)) {
74 deleteData(DeleteData.fromSerializable(message), SERIALIZED_REPLY);
76 } else if (message instanceof GetCompositedModification) {
77 // This is here for testing only
78 getSender().tell(new GetCompositeModificationReply(compositeModification), getSelf());
80 super.handleReceive(message);
84 private void batchedModifications(BatchedModifications batched) {
86 if (batched.isReady()) {
87 getSelf().tell(PoisonPill.getInstance(), getSelf());
93 for(Modification modification: batched.getModifications()) {
94 compositeModification.addModification(modification);
95 modification.apply(transaction.getSnapshot());
98 totalBatchedModificationsReceived++;
99 if(batched.isReady()) {
100 if(lastBatchedModificationsException != null) {
101 throw lastBatchedModificationsException;
104 if(totalBatchedModificationsReceived != batched.getTotalMessagesSent()) {
105 throw new IllegalStateException(String.format(
106 "The total number of batched messages received %d does not match the number sent %d",
107 totalBatchedModificationsReceived, batched.getTotalMessagesSent()));
110 readyTransaction(false, batched.isDoCommitOnReady());
112 getSender().tell(new BatchedModificationsReply(batched.getModifications().size()), getSelf());
114 } catch (Exception e) {
115 lastBatchedModificationsException = e;
116 getSender().tell(new akka.actor.Status.Failure(e), getSelf());
118 if(batched.isReady()) {
119 getSelf().tell(PoisonPill.getInstance(), getSelf());
124 protected final void dataExists(DataExists message, final boolean returnSerialized) {
125 super.dataExists(transaction, message, returnSerialized);
128 protected final void readData(ReadData message, final boolean returnSerialized) {
129 super.readData(transaction, message, returnSerialized);
132 private boolean checkClosed() {
133 if (transaction.isClosed()) {
134 getSender().tell(new akka.actor.Status.Failure(new IllegalStateException("Transaction is closed, no modifications allowed")), getSelf());
141 private void writeData(WriteData message, boolean returnSerialized) {
142 LOG.debug("writeData at path : {}", message.getPath());
147 compositeModification.addModification(
148 new WriteModification(message.getPath(), message.getData()));
150 transaction.getSnapshot().write(message.getPath(), message.getData());
151 WriteDataReply writeDataReply = WriteDataReply.INSTANCE;
152 getSender().tell(returnSerialized ? writeDataReply.toSerializable(message.getVersion()) :
153 writeDataReply, getSelf());
155 getSender().tell(new akka.actor.Status.Failure(e), getSelf());
159 private void mergeData(MergeData message, boolean returnSerialized) {
160 LOG.debug("mergeData at path : {}", message.getPath());
165 compositeModification.addModification(
166 new MergeModification(message.getPath(), message.getData()));
169 transaction.getSnapshot().merge(message.getPath(), message.getData());
170 MergeDataReply mergeDataReply = MergeDataReply.INSTANCE;
171 getSender().tell(returnSerialized ? mergeDataReply.toSerializable(message.getVersion()) :
172 mergeDataReply, getSelf());
174 getSender().tell(new akka.actor.Status.Failure(e), getSelf());
178 private void deleteData(DeleteData message, boolean returnSerialized) {
179 LOG.debug("deleteData at path : {}", message.getPath());
184 compositeModification.addModification(new DeleteModification(message.getPath()));
186 transaction.getSnapshot().delete(message.getPath());
187 DeleteDataReply deleteDataReply = DeleteDataReply.INSTANCE;
188 getSender().tell(returnSerialized ? deleteDataReply.toSerializable(message.getVersion()) :
189 deleteDataReply, getSelf());
190 } catch(Exception e) {
191 getSender().tell(new akka.actor.Status.Failure(e), getSelf());
195 private void readyTransaction(boolean returnSerialized, boolean doImmediateCommit) {
196 String transactionID = getTransactionID();
198 LOG.debug("readyTransaction : {}", transactionID);
200 DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
202 getShardActor().forward(new ForwardedReadyTransaction(transactionID, getClientTxVersion(),
203 cohort, compositeModification, returnSerialized, doImmediateCommit), getContext());
205 // The shard will handle the commit from here so we're no longer needed - self-destruct.
206 getSelf().tell(PoisonPill.getInstance(), getSelf());
209 // These classes are in here for test purposes only
211 static class GetCompositedModification {
214 static class GetCompositeModificationReply {
215 private final CompositeModification modification;
218 GetCompositeModificationReply(CompositeModification modification) {
219 this.modification = modification;
222 public CompositeModification getModification() {