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.AbstractUntypedActorWithMetering;
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 AbstractUntypedActorWithMetering {
60 private final ActorRef shardActor;
61 private final SchemaContext schemaContext;
62 private final ShardStats shardStats;
63 private final String transactionID;
65 protected ShardTransaction(ActorRef shardActor, SchemaContext schemaContext,
66 ShardStats shardStats, String transactionID) {
67 super("shard-tx"); //actor name override used for metering. This does not change the "real" actor name
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) {
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);
126 future.addListener(new Runnable() {
130 Optional<NormalizedNode<?, ?>> optional = future.checkedGet();
131 if (optional.isPresent()) {
132 sender.tell(new ReadDataReply(schemaContext,optional.get()).toSerializable(), self);
134 sender.tell(new ReadDataReply(schemaContext,null).toSerializable(), self);
136 } catch (Exception e) {
137 shardStats.incrementFailedReadTransactionsCount();
138 sender.tell(new akka.actor.Status.Failure(e), self);
142 }, getContext().dispatcher());
145 protected void dataExists(DOMStoreReadTransaction transaction, DataExists message) {
146 final YangInstanceIdentifier path = message.getPath();
149 Boolean exists = transaction.exists(path).checkedGet();
150 getSender().tell(new DataExistsReply(exists).toSerializable(), getSelf());
151 } catch (ReadFailedException e) {
152 getSender().tell(new akka.actor.Status.Failure(e),getSelf());
157 private static class ShardTransactionCreator implements Creator<ShardTransaction> {
159 private static final long serialVersionUID = 1L;
161 final DOMStoreTransaction transaction;
162 final ActorRef shardActor;
163 final SchemaContext schemaContext;
164 final DatastoreContext datastoreContext;
165 final ShardStats shardStats;
166 final String transactionID;
168 ShardTransactionCreator(DOMStoreTransaction transaction, ActorRef shardActor,
169 SchemaContext schemaContext, DatastoreContext datastoreContext,
170 ShardStats shardStats, String transactionID) {
171 this.transaction = transaction;
172 this.shardActor = shardActor;
173 this.shardStats = shardStats;
174 this.schemaContext = schemaContext;
175 this.datastoreContext = datastoreContext;
176 this.transactionID = transactionID;
180 public ShardTransaction create() throws Exception {
182 if(transaction instanceof DOMStoreReadWriteTransaction) {
183 tx = new ShardReadWriteTransaction((DOMStoreReadWriteTransaction)transaction,
184 shardActor, schemaContext, shardStats, transactionID);
185 } else if(transaction instanceof DOMStoreReadTransaction) {
186 tx = new ShardReadTransaction((DOMStoreReadTransaction)transaction, shardActor,
187 schemaContext, shardStats, transactionID);
189 tx = new ShardWriteTransaction((DOMStoreWriteTransaction)transaction,
190 shardActor, schemaContext, shardStats, transactionID);
193 tx.getContext().setReceiveTimeout(datastoreContext.getShardTransactionIdleTimeout());