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.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.util.concurrent.CheckedFuture;
18 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActor;
19 import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException;
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.controller.sal.core.spi.data.DOMStoreReadTransaction;
29 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
30 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
31 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
32 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
33 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
34 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
37 * The ShardTransaction Actor represents a remote transaction
39 * The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction
42 * Even though the DOMStore and the DOMStoreTransactionChain implement multiple types of transactions
43 * the ShardTransaction Actor only works with read-write transactions. This is just to keep the logic simple. At this
44 * time there are no known advantages for creating a read-only or write-only transaction which may change over time
45 * at which point we can optimize things in the distributed store as well.
48 * Handles Messages <br/>
49 * ---------------- <br/>
50 * <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadData}
51 * <li> {@link org.opendaylight.controller.cluster.datastore.messages.WriteData}
52 * <li> {@link org.opendaylight.controller.cluster.datastore.messages.MergeData}
53 * <li> {@link org.opendaylight.controller.cluster.datastore.messages.DeleteData}
54 * <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction}
55 * <li> {@link org.opendaylight.controller.cluster.datastore.messages.CloseTransaction}
58 public abstract class ShardTransaction extends AbstractUntypedActor {
60 private final ActorRef shardActor;
61 private final SchemaContext schemaContext;
62 private final ShardStats shardStats;
63 private final String transactionID;
64 protected static final boolean SERIALIZED_REPLY = true;
66 protected ShardTransaction(ActorRef shardActor, SchemaContext schemaContext,
67 ShardStats shardStats, String transactionID) {
68 this.shardActor = shardActor;
69 this.schemaContext = schemaContext;
70 this.shardStats = shardStats;
71 this.transactionID = transactionID;
74 public static Props props(DOMStoreTransaction transaction, ActorRef shardActor,
75 SchemaContext schemaContext,DatastoreContext datastoreContext, ShardStats shardStats,
76 String transactionID) {
77 return Props.create(new ShardTransactionCreator(transaction, shardActor, schemaContext,
78 datastoreContext, shardStats, transactionID));
81 protected abstract DOMStoreTransaction getDOMStoreTransaction();
83 protected ActorRef getShardActor() {
87 protected String getTransactionID() {
91 protected SchemaContext getSchemaContext() {
96 public void handleReceive(Object message) throws Exception {
97 if (message.getClass().equals(CloseTransaction.SERIALIZABLE_CLASS)) {
98 closeTransaction(true);
99 } else if (message instanceof ReceiveTimeout) {
100 if(LOG.isDebugEnabled()) {
101 LOG.debug("Got ReceiveTimeout for inactivity - closing Tx");
103 closeTransaction(false);
105 throw new UnknownMessageException(message);
109 private void closeTransaction(boolean sendReply) {
110 getDOMStoreTransaction().close();
113 getSender().tell(new CloseTransactionReply().toSerializable(), getSelf());
116 getSelf().tell(PoisonPill.getInstance(), getSelf());
119 protected void readData(DOMStoreReadTransaction transaction, ReadData message, final boolean returnSerialized) {
120 final ActorRef sender = getSender();
121 final ActorRef self = getSelf();
122 final YangInstanceIdentifier path = message.getPath();
123 final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> future =
124 transaction.read(path);
127 future.addListener(new Runnable() {
131 Optional<NormalizedNode<?, ?>> optional = future.checkedGet();
132 ReadDataReply readDataReply = new ReadDataReply(schemaContext, optional.orNull());
134 sender.tell((returnSerialized ? readDataReply.toSerializable():
135 readDataReply), self);
137 } catch (Exception e) {
138 shardStats.incrementFailedReadTransactionsCount();
139 sender.tell(new akka.actor.Status.Failure(e), self);
143 }, getContext().dispatcher());
146 protected void dataExists(DOMStoreReadTransaction transaction, DataExists message,
147 final boolean returnSerialized) {
148 final YangInstanceIdentifier path = message.getPath();
151 Boolean exists = transaction.exists(path).checkedGet();
152 DataExistsReply dataExistsReply = new DataExistsReply(exists);
153 getSender().tell(returnSerialized ? dataExistsReply.toSerializable() :
154 dataExistsReply, getSelf());
155 } catch (ReadFailedException e) {
156 getSender().tell(new akka.actor.Status.Failure(e),getSelf());
161 private static class ShardTransactionCreator implements Creator<ShardTransaction> {
163 private static final long serialVersionUID = 1L;
165 final DOMStoreTransaction transaction;
166 final ActorRef shardActor;
167 final SchemaContext schemaContext;
168 final DatastoreContext datastoreContext;
169 final ShardStats shardStats;
170 final String transactionID;
172 ShardTransactionCreator(DOMStoreTransaction transaction, ActorRef shardActor,
173 SchemaContext schemaContext, DatastoreContext datastoreContext,
174 ShardStats shardStats, String transactionID) {
175 this.transaction = transaction;
176 this.shardActor = shardActor;
177 this.shardStats = shardStats;
178 this.schemaContext = schemaContext;
179 this.datastoreContext = datastoreContext;
180 this.transactionID = transactionID;
184 public ShardTransaction create() throws Exception {
186 if(transaction instanceof DOMStoreReadWriteTransaction) {
187 tx = new ShardReadWriteTransaction((DOMStoreReadWriteTransaction)transaction,
188 shardActor, schemaContext, shardStats, transactionID);
189 } else if(transaction instanceof DOMStoreReadTransaction) {
190 tx = new ShardReadTransaction((DOMStoreReadTransaction)transaction, shardActor,
191 schemaContext, shardStats, transactionID);
193 tx = new ShardWriteTransaction((DOMStoreWriteTransaction)transaction,
194 shardActor, schemaContext, shardStats, transactionID);
197 tx.getContext().setReceiveTimeout(datastoreContext.getShardTransactionIdleTimeout());