2421cce26c0afa6f7a7c7cc3cf6b5da5830f6fd0
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / ShardTransaction.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.ActorRef;
12 import akka.actor.PoisonPill;
13 import akka.actor.Props;
14 import akka.actor.ReceiveTimeout;
15 import akka.japi.Creator;
16 import com.google.common.base.Optional;
17 import com.google.common.base.Preconditions;
18 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
19 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActorWithMetering;
20 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
21 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
22 import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionReply;
23 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
24 import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply;
25 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
26 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
27 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
28 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
29 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
30
31 /**
32  * The ShardTransaction Actor represents a remote transaction
33  * <p>
34  * The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction
35  * </p>
36  * <p>
37  * Handles Messages <br/>
38  * ---------------- <br/>
39  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadData}
40  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.CloseTransaction}
41  * </p>
42  */
43 public abstract class ShardTransaction extends AbstractUntypedActorWithMetering {
44     private final ActorRef shardActor;
45     private final ShardStats shardStats;
46     private final TransactionIdentifier transactionID;
47
48     protected ShardTransaction(ActorRef shardActor, ShardStats shardStats, TransactionIdentifier transactionID) {
49         super("shard-tx"); //actor name override used for metering. This does not change the "real" actor name
50         this.shardActor = shardActor;
51         this.shardStats = shardStats;
52         this.transactionID = Preconditions.checkNotNull(transactionID);
53     }
54
55     public static Props props(TransactionType type, AbstractShardDataTreeTransaction<?> transaction, ActorRef shardActor,
56             DatastoreContext datastoreContext, ShardStats shardStats) {
57         return Props.create(new ShardTransactionCreator(type, transaction, shardActor, datastoreContext, shardStats));
58     }
59
60     protected abstract AbstractShardDataTreeTransaction<?> getDOMStoreTransaction();
61
62     protected ActorRef getShardActor() {
63         return shardActor;
64     }
65
66     protected final TransactionIdentifier getTransactionID() {
67         return transactionID;
68     }
69
70     @Override
71     public void handleReceive(Object message) {
72         if (CloseTransaction.isSerializedType(message)) {
73             closeTransaction(true);
74         } else if (message instanceof ReceiveTimeout) {
75             LOG.debug("Got ReceiveTimeout for inactivity - closing transaction {}", transactionID);
76             closeTransaction(false);
77         } else {
78             unknownMessage(message);
79         }
80     }
81
82     protected boolean returnCloseTransactionReply() {
83         return true;
84     }
85
86     private void closeTransaction(boolean sendReply) {
87         getDOMStoreTransaction().abort();
88
89         if(sendReply && returnCloseTransactionReply()) {
90             getSender().tell(new CloseTransactionReply(), getSelf());
91         }
92
93         getSelf().tell(PoisonPill.getInstance(), getSelf());
94     }
95
96     private boolean checkClosed(AbstractShardDataTreeTransaction<?> transaction) {
97         final boolean ret = transaction.isClosed();
98         if (ret) {
99             shardStats.incrementFailedReadTransactionsCount();
100             getSender().tell(new akka.actor.Status.Failure(new ReadFailedException("Transaction is closed")), getSelf());
101         }
102         return ret;
103     }
104
105     protected void readData(AbstractShardDataTreeTransaction<?> transaction, ReadData message) {
106         if (checkClosed(transaction)) {
107             return;
108         }
109
110         final YangInstanceIdentifier path = message.getPath();
111         Optional<NormalizedNode<?, ?>> optional = transaction.getSnapshot().readNode(path);
112         ReadDataReply readDataReply = new ReadDataReply(optional.orNull(), message.getVersion());
113         sender().tell(readDataReply.toSerializable(), self());
114     }
115
116     protected void dataExists(AbstractShardDataTreeTransaction<?> transaction, DataExists message) {
117         if (checkClosed(transaction)) {
118             return;
119         }
120
121         final YangInstanceIdentifier path = message.getPath();
122         boolean exists = transaction.getSnapshot().readNode(path).isPresent();
123         getSender().tell(new DataExistsReply(exists, message.getVersion()).toSerializable(), getSelf());
124     }
125
126     private static class ShardTransactionCreator implements Creator<ShardTransaction> {
127
128         private static final long serialVersionUID = 1L;
129
130         final AbstractShardDataTreeTransaction<?> transaction;
131         final ActorRef shardActor;
132         final DatastoreContext datastoreContext;
133         final ShardStats shardStats;
134         final TransactionType type;
135
136         ShardTransactionCreator(TransactionType type, AbstractShardDataTreeTransaction<?> transaction, ActorRef shardActor,
137                 DatastoreContext datastoreContext, ShardStats shardStats) {
138             this.transaction = Preconditions.checkNotNull(transaction);
139             this.shardActor = shardActor;
140             this.shardStats = shardStats;
141             this.datastoreContext = datastoreContext;
142             this.type = type;
143         }
144
145         @Override
146         public ShardTransaction create() throws Exception {
147             final ShardTransaction tx;
148             switch (type) {
149             case READ_ONLY:
150                 tx = new ShardReadTransaction(transaction, shardActor, shardStats);
151                 break;
152             case READ_WRITE:
153                 tx = new ShardReadWriteTransaction((ReadWriteShardDataTreeTransaction)transaction, shardActor, shardStats);
154                 break;
155             case WRITE_ONLY:
156                 tx = new ShardWriteTransaction((ReadWriteShardDataTreeTransaction)transaction, shardActor, shardStats);
157                 break;
158             default:
159                 throw new IllegalArgumentException("Unhandled transaction type " + type);
160             }
161
162             tx.getContext().setReceiveTimeout(datastoreContext.getShardTransactionIdleTimeout());
163             return tx;
164         }
165     }
166 }

©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.