import akka.actor.Props;
import akka.actor.ReceiveTimeout;
import akka.japi.Creator;
+
import com.google.common.base.Optional;
import com.google.common.util.concurrent.CheckedFuture;
+import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActor;
+
+
import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException;
-import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
+import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.DataExists;
private final ActorRef shardActor;
protected final SchemaContext schemaContext;
- private final String shardName;
-
+ private final ShardStats shardStats;
private final MutableCompositeModification modification = new MutableCompositeModification();
protected ShardTransaction(ActorRef shardActor, SchemaContext schemaContext,
- String shardName) {
+ ShardStats shardStats) {
this.shardActor = shardActor;
this.schemaContext = schemaContext;
- this.shardName = shardName;
+ this.shardStats = shardStats;
}
public static Props props(DOMStoreTransaction transaction, ActorRef shardActor,
- SchemaContext schemaContext,DatastoreContext datastoreContext, String shardName) {
+ SchemaContext schemaContext,DatastoreContext datastoreContext, ShardStats shardStats) {
return Props.create(new ShardTransactionCreator(transaction, shardActor, schemaContext,
- datastoreContext, shardName));
+ datastoreContext, shardStats));
}
protected abstract DOMStoreTransaction getDOMStoreTransaction();
getSender().tell(new GetCompositeModificationReply(
new ImmutableCompositeModification(modification)), getSelf());
} else if (message instanceof ReceiveTimeout) {
- LOG.debug("Got ReceiveTimeout for inactivity - closing Tx");
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Got ReceiveTimeout for inactivity - closing Tx");
+ }
closeTransaction(false);
} else {
throw new UnknownMessageException(message);
sender.tell(new ReadDataReply(schemaContext,null).toSerializable(), self);
}
} catch (Exception e) {
- ShardMBeanFactory.getShardStatsMBean(shardName).incrementFailedReadTransactionsCount();
+ shardStats.incrementFailedReadTransactionsCount();
sender.tell(new akka.actor.Status.Failure(e), self);
}
protected void writeData(DOMStoreWriteTransaction transaction, WriteData message) {
modification.addModification(
new WriteModification(message.getPath(), message.getData(),schemaContext));
- LOG.debug("writeData at path : " + message.getPath().toString());
-
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("writeData at path : " + message.getPath().toString());
+ }
try {
transaction.write(message.getPath(), message.getData());
getSender().tell(new WriteDataReply().toSerializable(), getSelf());
protected void mergeData(DOMStoreWriteTransaction transaction, MergeData message) {
modification.addModification(
new MergeModification(message.getPath(), message.getData(), schemaContext));
- LOG.debug("mergeData at path : " + message.getPath().toString());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("mergeData at path : " + message.getPath().toString());
+ }
try {
transaction.merge(message.getPath(), message.getData());
getSender().tell(new MergeDataReply().toSerializable(), getSelf());
}
protected void deleteData(DOMStoreWriteTransaction transaction, DeleteData message) {
- LOG.debug("deleteData at path : " + message.getPath().toString());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("deleteData at path : " + message.getPath().toString());
+ }
modification.addModification(new DeleteModification(message.getPath()));
try {
transaction.delete(message.getPath());
protected void readyTransaction(DOMStoreWriteTransaction transaction, ReadyTransaction message) {
DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
ActorRef cohortActor = getContext().actorOf(
- ThreePhaseCommitCohort.props(cohort, shardActor, modification, shardName), "cohort");
+ ThreePhaseCommitCohort.props(cohort, shardActor, modification, shardStats), "cohort");
getSender()
.tell(new ReadyTransactionReply(cohortActor.path()).toSerializable(), getSelf());
final ActorRef shardActor;
final SchemaContext schemaContext;
final DatastoreContext datastoreContext;
- final String shardName;
+ final ShardStats shardStats;
ShardTransactionCreator(DOMStoreTransaction transaction, ActorRef shardActor,
- SchemaContext schemaContext, DatastoreContext datastoreContext, String shardName) {
+ SchemaContext schemaContext, DatastoreContext datastoreContext,
+ ShardStats shardStats) {
this.transaction = transaction;
this.shardActor = shardActor;
- this.shardName = shardName;
+ this.shardStats = shardStats;
this.schemaContext = schemaContext;
this.datastoreContext = datastoreContext;
}
ShardTransaction tx;
if(transaction instanceof DOMStoreReadWriteTransaction) {
tx = new ShardReadWriteTransaction((DOMStoreReadWriteTransaction)transaction,
- shardActor, schemaContext, shardName);
+ shardActor, schemaContext, shardStats);
} else if(transaction instanceof DOMStoreReadTransaction) {
tx = new ShardReadTransaction((DOMStoreReadTransaction)transaction, shardActor,
- schemaContext, shardName);
+ schemaContext, shardStats);
} else {
tx = new ShardWriteTransaction((DOMStoreWriteTransaction)transaction,
- shardActor, schemaContext, shardName);
+ shardActor, schemaContext, shardStats);
}
tx.getContext().setReceiveTimeout(datastoreContext.getShardTransactionIdleTimeout());