opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/BasicIntegrationTest.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.ActorPath;
  12 import akka.actor.ActorRef;
  13 import akka.actor.ActorSelection;
  14 import akka.actor.Props;
  15 import akka.actor.Terminated;
  16 import akka.testkit.JavaTestKit;
  17 import junit.framework.Assert;
  18 import org.junit.Test;
  19 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
  20 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  21 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
  22 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
  23 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
  24 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
  25 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransaction;
  26 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply;
  27 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
  28 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  29 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
  30 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
  31 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
  32 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  33 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  34 import scala.concurrent.Await;
  35 import scala.concurrent.Future;
  36 import scala.concurrent.duration.FiniteDuration;
  37
  38 public class BasicIntegrationTest extends AbstractActorTest {
  39
  40     @Test
  41     public void integrationTest() throws Exception{
  42         // This test will
  43         // - create a Shard
  44         // - initiate a transaction
  45         // - write something
  46         // - read the transaction for commit
  47         // - commit the transaction
  48
  49
  50         new JavaTestKit(getSystem()) {{
  51             final Props props = Shard.props("config");
  52             final ActorRef shard = getSystem().actorOf(props);
  53
  54             new Within(duration("5 seconds")) {
  55                 protected void run() {
  56
  57                     shard.tell(
  58                         new UpdateSchemaContext(TestModel.createTestContext()),
  59                         getRef());
  60
  61                     shard.tell(new CreateTransactionChain(), getRef());
  62
  63                     final ActorSelection transactionChain =
  64                         new ExpectMsg<ActorSelection>("CreateTransactionChainReply") {
  65                             protected ActorSelection match(Object in) {
  66                                 if (in instanceof CreateTransactionChainReply) {
  67                                     ActorPath transactionChainPath =
  68                                         ((CreateTransactionChainReply) in)
  69                                             .getTransactionChainPath();
  70                                     return getSystem()
  71                                         .actorSelection(transactionChainPath);
  72                                 } else {
  73                                     throw noMatch();
  74                                 }
  75                             }
  76                         }.get(); // this extracts the received message
  77
  78                     Assert.assertNotNull(transactionChain);
  79
  80                     transactionChain.tell(new CreateTransaction("txn-1"), getRef());
  81
  82                     final ActorSelection transaction =
  83                         new ExpectMsg<ActorSelection>("CreateTransactionReply") {
  84                             protected ActorSelection match(Object in) {
  85                                 if (in instanceof CreateTransactionReply) {
  86                                     return getSystem()
  87                                         .actorSelection((((CreateTransactionReply) in)
  88                                             .getTransactionActorPath()));
  89                                 } else {
  90                                     throw noMatch();
  91                                 }
  92                             }
  93                         }.get(); // this extracts the received message
  94
  95                     Assert.assertNotNull(transaction);
  96
  97                     // Add a watch on the transaction actor so that we are notified when it dies
  98                     final ActorRef transactionActorRef = watchActor(transaction);
  99
 100                     transaction.tell(new WriteData(TestModel.TEST_PATH,
 101                         ImmutableNodes.containerNode(TestModel.TEST_QNAME)),
 102                         getRef());
 103
 104                     Boolean writeDone = new ExpectMsg<Boolean>("WriteDataReply") {
 105                         protected Boolean match(Object in) {
 106                             if (in instanceof WriteDataReply) {
 107                                 return true;
 108                             } else {
 109                                 throw noMatch();
 110                             }
 111                         }
 112                     }.get(); // this extracts the received message
 113
 114                     Assert.assertTrue(writeDone);
 115
 116                     transaction.tell(new ReadyTransaction(), getRef());
 117
 118                     final ActorSelection cohort =
 119                         new ExpectMsg<ActorSelection>("ReadyTransactionReply") {
 120                             protected ActorSelection match(Object in) {
 121                                 if (in instanceof ReadyTransactionReply) {
 122                                     ActorPath cohortPath =
 123                                         ((ReadyTransactionReply) in)
 124                                             .getCohortPath();
 125                                     return getSystem()
 126                                         .actorSelection(cohortPath);
 127                                 } else {
 128                                     throw noMatch();
 129                                 }
 130                             }
 131                         }.get(); // this extracts the received message
 132
 133                     Assert.assertNotNull(cohort);
 134
 135                     // Add a watch on the transaction actor so that we are notified when it dies
 136                     final ActorRef cohorActorRef = watchActor(cohort);
 137
 138                     cohort.tell(new PreCommitTransaction(), getRef());
 139
 140                     Boolean preCommitDone =
 141                         new ExpectMsg<Boolean>("PreCommitTransactionReply") {
 142                             protected Boolean match(Object in) {
 143                                 if (in instanceof PreCommitTransactionReply) {
 144                                     return true;
 145                                 } else {
 146                                     throw noMatch();
 147                                 }
 148                             }
 149                         }.get(); // this extracts the received message
 150
 151                     Assert.assertTrue(preCommitDone);
 152
 153                     // FIXME : When we commit on the cohort it "kills" the Transaction.
 154                     // This in turn kills the child of Transaction as well.
 155                     // The order in which we receive the terminated event for both
 156                     // these actors is not fixed which may cause this test to fail
 157                     cohort.tell(new CommitTransaction(), getRef());
 158
 159                     final Boolean terminatedCohort =
 160                         new ExpectMsg<Boolean>("Terminated Cohort") {
 161                             protected Boolean match(Object in) {
 162                                 if (in instanceof Terminated) {
 163                                     return cohorActorRef.equals(((Terminated) in).actor());
 164                                 } else {
 165                                     throw noMatch();
 166                                 }
 167                             }
 168                         }.get(); // this extracts the received message
 169
 170                     Assert.assertTrue(terminatedCohort);
 171
 172
 173                     final Boolean terminatedTransaction =
 174                         new ExpectMsg<Boolean>("Terminated Transaction") {
 175                             protected Boolean match(Object in) {
 176                                 if (in instanceof Terminated) {
 177                                     return transactionActorRef.equals(((Terminated) in).actor());
 178                                 } else {
 179                                     throw noMatch();
 180                                 }
 181                             }
 182                         }.get(); // this extracts the received message
 183
 184                     Assert.assertTrue(terminatedTransaction);
 185
 186                     final Boolean commitDone =
 187                         new ExpectMsg<Boolean>("CommitTransactionReply") {
 188                             protected Boolean match(Object in) {
 189                                 if (in instanceof CommitTransactionReply) {
 190                                     return true;
 191                                 } else {
 192                                     throw noMatch();
 193                                 }
 194                             }
 195                         }.get(); // this extracts the received message
 196
 197                     Assert.assertTrue(commitDone);
 198
 199                 }
 200
 201
 202             };
 203         }
 204
 205             private ActorRef watchActor(ActorSelection actor) {
 206                 Future<ActorRef> future = actor
 207                     .resolveOne(FiniteDuration.apply(100, "milliseconds"));
 208
 209                 try {
 210                     ActorRef actorRef = Await.result(future,
 211                         FiniteDuration.apply(100, "milliseconds"));
 212
 213                     watch(actorRef);
 214
 215                     return actorRef;
 216                 } catch (Exception e) {
 217                     throw new RuntimeException(e);
 218                 }
 219
 220             }
 221         };
 222
 223
 224     }
 225 }