-public class ShardTransaction extends UntypedActor {
-
- private final DOMStoreReadWriteTransaction transaction;
-
- private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
-
- public ShardTransaction(DOMStoreReadWriteTransaction transaction) {
- this.transaction = transaction;
- }
-
-
- public static Props props(final DOMStoreReadWriteTransaction transaction){
- return Props.create(new Creator<ShardTransaction>(){
-
- @Override
- public ShardTransaction create() throws Exception {
- return new ShardTransaction(transaction);
- }
- });
- }
-
- @Override
- public void onReceive(Object message) throws Exception {
- if(message instanceof ReadData){
- readData((ReadData) message);
- } else if(message instanceof WriteData){
- writeData((WriteData) message);
- } else if(message instanceof MergeData){
- mergeData((MergeData) message);
- } else if(message instanceof DeleteData){
- deleteData((DeleteData) message);
- } else if(message instanceof ReadyTransaction){
- readyTransaction((ReadyTransaction) message);
- } else if(message instanceof CloseTransaction){
- closeTransaction((CloseTransaction) message);
+public abstract class ShardTransaction extends AbstractUntypedActor {
+
+ private final ActorRef shardActor;
+ protected final SchemaContext schemaContext;
+
+ private final MutableCompositeModification modification = new MutableCompositeModification();
+
+ protected ShardTransaction(ActorRef shardActor, SchemaContext schemaContext) {
+ this.shardActor = shardActor;
+ this.schemaContext = schemaContext;
+ }
+
+ public static Props props(DOMStoreTransaction transaction, ActorRef shardActor,
+ SchemaContext schemaContext, ShardContext shardContext) {
+ return Props.create(new ShardTransactionCreator(transaction, shardActor, schemaContext,
+ shardContext));
+ }
+
+ protected abstract DOMStoreTransaction getDOMStoreTransaction();
+
+ @Override
+ public void handleReceive(Object message) throws Exception {
+ if (message.getClass().equals(CloseTransaction.SERIALIZABLE_CLASS)) {
+ closeTransaction(true);
+ } else if (message instanceof GetCompositedModification) {
+ // This is here for testing only
+ getSender().tell(new GetCompositeModificationReply(
+ new ImmutableCompositeModification(modification)), getSelf());
+ } else if (message instanceof ReceiveTimeout) {
+ LOG.debug("Got ReceiveTimeout for inactivity - closing Tx");
+ closeTransaction(false);
+ } else {
+ throw new UnknownMessageException(message);
+ }
+ }
+
+ private void closeTransaction(boolean sendReply) {
+ getDOMStoreTransaction().close();
+
+ if(sendReply) {
+ getSender().tell(new CloseTransactionReply().toSerializable(), getSelf());
+ }
+
+ getSelf().tell(PoisonPill.getInstance(), getSelf());
+ }
+
+ protected void readData(DOMStoreReadTransaction transaction,ReadData message) {
+ final ActorRef sender = getSender();
+ final ActorRef self = getSelf();
+ final YangInstanceIdentifier path = message.getPath();
+ final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> future =
+ transaction.read(path);
+
+ future.addListener(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Optional<NormalizedNode<?, ?>> optional = future.checkedGet();
+ if (optional.isPresent()) {
+ sender.tell(new ReadDataReply(schemaContext,optional.get()).toSerializable(), self);
+ } else {
+ sender.tell(new ReadDataReply(schemaContext,null).toSerializable(), self);
+ }
+ } catch (Exception e) {
+ sender.tell(new akka.actor.Status.Failure(e),self);
+ }
+
+ }
+ }, getContext().dispatcher());