3b175a9c05956174896711412474abac649e0141
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / ShardWriteTransaction.java
1 /*
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.
4  *
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
8  */
9
10 package org.opendaylight.controller.cluster.datastore;
11
12 import akka.actor.ActorRef;
13 import akka.actor.PoisonPill;
14 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
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.ForwardedReadyTransaction;
20 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
21 import org.opendaylight.controller.cluster.datastore.modification.Modification;
22
23 /**
24  * @author: syedbahm
25  * Date: 8/6/14
26  */
27 public class ShardWriteTransaction extends ShardTransaction {
28
29     private int totalBatchedModificationsReceived;
30     private Exception lastBatchedModificationsException;
31     private final ReadWriteShardDataTreeTransaction transaction;
32
33     public ShardWriteTransaction(ReadWriteShardDataTreeTransaction transaction, ActorRef shardActor,
34             ShardStats shardStats) {
35         super(shardActor, shardStats, transaction.getId());
36         this.transaction = transaction;
37     }
38
39     @Override
40     protected ReadWriteShardDataTreeTransaction getDOMStoreTransaction() {
41         return transaction;
42     }
43
44     @Override
45     public void handleReceive(Object message) {
46         if (message instanceof BatchedModifications) {
47             batchedModifications((BatchedModifications)message);
48         } else {
49             super.handleReceive(message);
50         }
51     }
52
53     private void batchedModifications(BatchedModifications batched) {
54         if (checkClosed()) {
55             if (batched.isReady()) {
56                 getSelf().tell(PoisonPill.getInstance(), getSelf());
57             }
58             return;
59         }
60
61         try {
62             for(Modification modification: batched.getModifications()) {
63                 modification.apply(transaction.getSnapshot());
64             }
65
66             totalBatchedModificationsReceived++;
67             if(batched.isReady()) {
68                 if(lastBatchedModificationsException != null) {
69                     throw lastBatchedModificationsException;
70                 }
71
72                 if(totalBatchedModificationsReceived != batched.getTotalMessagesSent()) {
73                     throw new IllegalStateException(String.format(
74                             "The total number of batched messages received %d does not match the number sent %d",
75                             totalBatchedModificationsReceived, batched.getTotalMessagesSent()));
76                 }
77
78                 readyTransaction(false, batched.isDoCommitOnReady(), batched.getVersion());
79             } else {
80                 getSender().tell(new BatchedModificationsReply(batched.getModifications().size()), getSelf());
81             }
82         } catch (Exception e) {
83             lastBatchedModificationsException = e;
84             getSender().tell(new akka.actor.Status.Failure(e), getSelf());
85
86             if(batched.isReady()) {
87                 getSelf().tell(PoisonPill.getInstance(), getSelf());
88             }
89         }
90     }
91
92     protected final void dataExists(DataExists message) {
93         super.dataExists(transaction, message);
94     }
95
96     protected final void readData(ReadData message) {
97         super.readData(transaction, message);
98     }
99
100     private boolean checkClosed() {
101         if (transaction.isClosed()) {
102             getSender().tell(new akka.actor.Status.Failure(new IllegalStateException("Transaction is closed, no modifications allowed")), getSelf());
103             return true;
104         } else {
105             return false;
106         }
107     }
108
109     private void readyTransaction(boolean returnSerialized, boolean doImmediateCommit, short clientTxVersion) {
110         TransactionIdentifier transactionID = getTransactionID();
111
112         LOG.debug("readyTransaction : {}", transactionID);
113
114         getShardActor().forward(new ForwardedReadyTransaction(transactionID, clientTxVersion,
115                 transaction, doImmediateCommit), getContext());
116
117         // The shard will handle the commit from here so we're no longer needed - self-destruct.
118         getSelf().tell(PoisonPill.getInstance(), getSelf());
119     }
120 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.