opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java

   1 package org.opendaylight.controller.cluster.datastore;
   2
   3 import static org.junit.Assert.assertEquals;
   4 import static org.junit.Assert.assertNotNull;
   5 import static org.junit.Assert.assertTrue;
   6 import static org.junit.Assert.fail;
   7 import static org.mockito.Matchers.any;
   8 import static org.mockito.Matchers.anyString;
   9 import static org.mockito.Matchers.argThat;
  10 import static org.mockito.Matchers.eq;
  11 import static org.mockito.Matchers.isA;
  12 import static org.mockito.Mockito.doReturn;
  13 import static org.mockito.Mockito.mock;
  14 import static org.mockito.Mockito.times;
  15 import static org.mockito.Mockito.verify;
  16 import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_ONLY;
  17 import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_WRITE;
  18 import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.WRITE_ONLY;
  19 import akka.actor.ActorRef;
  20 import akka.actor.ActorSelection;
  21 import akka.actor.ActorSystem;
  22 import akka.actor.Props;
  23 import akka.dispatch.Futures;
  24 import akka.testkit.JavaTestKit;
  25 import com.google.common.base.Optional;
  26 import com.google.common.collect.ImmutableMap;
  27 import com.google.common.util.concurrent.CheckedFuture;
  28 import com.google.common.util.concurrent.FutureCallback;
  29 import com.google.common.util.concurrent.Uninterruptibles;
  30 import com.typesafe.config.Config;
  31 import com.typesafe.config.ConfigFactory;
  32 import java.io.IOException;
  33 import java.util.ArrayList;
  34 import java.util.List;
  35 import java.util.concurrent.CountDownLatch;
  36 import java.util.concurrent.TimeUnit;
  37 import java.util.concurrent.atomic.AtomicReference;
  38 import org.junit.AfterClass;
  39 import org.junit.Assert;
  40 import org.junit.Before;
  41 import org.junit.BeforeClass;
  42 import org.junit.Test;
  43 import org.mockito.ArgumentCaptor;
  44 import org.mockito.ArgumentMatcher;
  45 import org.mockito.InOrder;
  46 import org.mockito.Mock;
  47 import org.mockito.Mockito;
  48 import org.mockito.MockitoAnnotations;
  49 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
  50 import org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType;
  51 import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
  52 import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
  53 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  54 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
  55 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  56 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
  57 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
  58 import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply;
  59 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
  60 import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply;
  61 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
  62 import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply;
  63 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
  64 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
  65 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
  66 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  67 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
  68 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
  69 import org.opendaylight.controller.cluster.datastore.modification.AbstractModification;
  70 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
  71 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
  72 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  73 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  74 import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
  75 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
  76 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  77 import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor;
  78 import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
  79 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
  80 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  81 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  82 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages;
  83 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
  84 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  85 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  86 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  87 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  88 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  89 import scala.concurrent.Await;
  90 import scala.concurrent.Future;
  91 import scala.concurrent.Promise;
  92 import scala.concurrent.duration.Duration;
  93
  94 @SuppressWarnings("resource")
  95 public class TransactionProxyTest {
  96
  97     @SuppressWarnings("serial")
  98     static class TestException extends RuntimeException {
  99     }
 100
 101     static interface Invoker {
 102         CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception;
 103     }
 104
 105     private static ActorSystem system;
 106
 107     private final Configuration configuration = new MockConfiguration();
 108
 109     @Mock
 110     private ActorContext mockActorContext;
 111
 112     private SchemaContext schemaContext;
 113
 114     @Mock
 115     private ClusterWrapper mockClusterWrapper;
 116
 117     private final String memberName = "mock-member";
 118
 119     private final Builder dataStoreContextBuilder = DatastoreContext.newBuilder().operationTimeoutInSeconds(2).
 120             shardBatchedModificationCount(1);
 121
 122     @BeforeClass
 123     public static void setUpClass() throws IOException {
 124
 125         Config config = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder().
 126                 put("akka.actor.default-dispatcher.type",
 127                         "akka.testkit.CallingThreadDispatcherConfigurator").build()).
 128                 withFallback(ConfigFactory.load());
 129         system = ActorSystem.create("test", config);
 130     }
 131
 132     @AfterClass
 133     public static void tearDownClass() throws IOException {
 134         JavaTestKit.shutdownActorSystem(system);
 135         system = null;
 136     }
 137
 138     @Before
 139     public void setUp(){
 140         MockitoAnnotations.initMocks(this);
 141
 142         schemaContext = TestModel.createTestContext();
 143
 144         doReturn(getSystem()).when(mockActorContext).getActorSystem();
 145         doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(mockActorContext).getClientDispatcher();
 146         doReturn(memberName).when(mockActorContext).getCurrentMemberName();
 147         doReturn(schemaContext).when(mockActorContext).getSchemaContext();
 148         doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper();
 149         doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper();
 150         doReturn(dataStoreContextBuilder.build()).when(mockActorContext).getDatastoreContext();
 151         doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit();
 152
 153         ShardStrategyFactory.setConfiguration(configuration);
 154     }
 155
 156     private ActorSystem getSystem() {
 157         return system;
 158     }
 159
 160     private CreateTransaction eqCreateTransaction(final String memberName,
 161             final TransactionType type) {
 162         ArgumentMatcher<CreateTransaction> matcher = new ArgumentMatcher<CreateTransaction>() {
 163             @Override
 164             public boolean matches(Object argument) {
 165                 if(CreateTransaction.SERIALIZABLE_CLASS.equals(argument.getClass())) {
 166                     CreateTransaction obj = CreateTransaction.fromSerializable(argument);
 167                     return obj.getTransactionId().startsWith(memberName) &&
 168                             obj.getTransactionType() == type.ordinal();
 169                 }
 170
 171                 return false;
 172             }
 173         };
 174
 175         return argThat(matcher);
 176     }
 177
 178     private DataExists eqSerializedDataExists() {
 179         ArgumentMatcher<DataExists> matcher = new ArgumentMatcher<DataExists>() {
 180             @Override
 181             public boolean matches(Object argument) {
 182                 return DataExists.SERIALIZABLE_CLASS.equals(argument.getClass()) &&
 183                        DataExists.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH);
 184             }
 185         };
 186
 187         return argThat(matcher);
 188     }
 189
 190     private DataExists eqDataExists() {
 191         ArgumentMatcher<DataExists> matcher = new ArgumentMatcher<DataExists>() {
 192             @Override
 193             public boolean matches(Object argument) {
 194                 return (argument instanceof DataExists) &&
 195                     ((DataExists)argument).getPath().equals(TestModel.TEST_PATH);
 196             }
 197         };
 198
 199         return argThat(matcher);
 200     }
 201
 202     private ReadData eqSerializedReadData() {
 203         return eqSerializedReadData(TestModel.TEST_PATH);
 204     }
 205
 206     private ReadData eqSerializedReadData(final YangInstanceIdentifier path) {
 207         ArgumentMatcher<ReadData> matcher = new ArgumentMatcher<ReadData>() {
 208             @Override
 209             public boolean matches(Object argument) {
 210                 return ReadData.SERIALIZABLE_CLASS.equals(argument.getClass()) &&
 211                        ReadData.fromSerializable(argument).getPath().equals(path);
 212             }
 213         };
 214
 215         return argThat(matcher);
 216     }
 217
 218     private ReadData eqReadData() {
 219         ArgumentMatcher<ReadData> matcher = new ArgumentMatcher<ReadData>() {
 220             @Override
 221             public boolean matches(Object argument) {
 222                 return (argument instanceof ReadData) &&
 223                     ((ReadData)argument).getPath().equals(TestModel.TEST_PATH);
 224             }
 225         };
 226
 227         return argThat(matcher);
 228     }
 229
 230     private WriteData eqLegacyWriteData(final NormalizedNode<?, ?> nodeToWrite) {
 231         ArgumentMatcher<WriteData> matcher = new ArgumentMatcher<WriteData>() {
 232             @Override
 233             public boolean matches(Object argument) {
 234                 if(ShardTransactionMessages.WriteData.class.equals(argument.getClass())) {
 235                     WriteData obj = WriteData.fromSerializable(argument);
 236                     return obj.getPath().equals(TestModel.TEST_PATH) && obj.getData().equals(nodeToWrite);
 237                 }
 238
 239                 return false;
 240             }
 241         };
 242
 243         return argThat(matcher);
 244     }
 245
 246     private MergeData eqLegacyMergeData(final NormalizedNode<?, ?> nodeToWrite) {
 247         ArgumentMatcher<MergeData> matcher = new ArgumentMatcher<MergeData>() {
 248             @Override
 249             public boolean matches(Object argument) {
 250                 if(ShardTransactionMessages.MergeData.class.equals(argument.getClass())) {
 251                     MergeData obj = MergeData.fromSerializable(argument);
 252                     return obj.getPath().equals(TestModel.TEST_PATH) && obj.getData().equals(nodeToWrite);
 253                 }
 254
 255                 return false;
 256             }
 257         };
 258
 259         return argThat(matcher);
 260     }
 261
 262     private DeleteData eqLegacyDeleteData(final YangInstanceIdentifier expPath) {
 263         ArgumentMatcher<DeleteData> matcher = new ArgumentMatcher<DeleteData>() {
 264             @Override
 265             public boolean matches(Object argument) {
 266                 return ShardTransactionMessages.DeleteData.class.equals(argument.getClass()) &&
 267                        DeleteData.fromSerializable(argument).getPath().equals(expPath);
 268             }
 269         };
 270
 271         return argThat(matcher);
 272     }
 273
 274     private Future<Object> readySerializedTxReply(String path) {
 275         return Futures.successful((Object)new ReadyTransactionReply(path).toSerializable());
 276     }
 277
 278     private Future<Object> readyTxReply(String path) {
 279         return Futures.successful((Object)new ReadyTransactionReply(path));
 280     }
 281
 282     private Future<Object> readSerializedDataReply(NormalizedNode<?, ?> data,
 283             short transactionVersion) {
 284         return Futures.successful(new ReadDataReply(data, transactionVersion).toSerializable());
 285     }
 286
 287     private Future<Object> readSerializedDataReply(NormalizedNode<?, ?> data) {
 288         return readSerializedDataReply(data, DataStoreVersions.CURRENT_VERSION);
 289     }
 290
 291     private Future<ReadDataReply> readDataReply(NormalizedNode<?, ?> data) {
 292         return Futures.successful(new ReadDataReply(data, DataStoreVersions.CURRENT_VERSION));
 293     }
 294
 295     private Future<Object> dataExistsSerializedReply(boolean exists) {
 296         return Futures.successful(new DataExistsReply(exists).toSerializable());
 297     }
 298
 299     private Future<DataExistsReply> dataExistsReply(boolean exists) {
 300         return Futures.successful(new DataExistsReply(exists));
 301     }
 302
 303     private Future<BatchedModificationsReply> batchedModificationsReply(int count) {
 304         return Futures.successful(new BatchedModificationsReply(count));
 305     }
 306
 307     private Future<Object> incompleteFuture(){
 308         return mock(Future.class);
 309     }
 310
 311     private ActorSelection actorSelection(ActorRef actorRef) {
 312         return getSystem().actorSelection(actorRef.path());
 313     }
 314
 315     private void expectBatchedModifications(ActorRef actorRef, int count) {
 316         doReturn(batchedModificationsReply(count)).when(mockActorContext).executeOperationAsync(
 317                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
 318     }
 319
 320     private void expectBatchedModifications(int count) {
 321         doReturn(batchedModificationsReply(count)).when(mockActorContext).executeOperationAsync(
 322                 any(ActorSelection.class), isA(BatchedModifications.class));
 323     }
 324
 325     private void expectIncompleteBatchedModifications() {
 326         doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
 327                 any(ActorSelection.class), isA(BatchedModifications.class));
 328     }
 329
 330     private void expectReadyTransaction(ActorRef actorRef) {
 331         doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync(
 332                 eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS));
 333     }
 334
 335     private void expectFailedBatchedModifications(ActorRef actorRef) {
 336         doReturn(Futures.failed(new TestException())).when(mockActorContext).executeOperationAsync(
 337                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
 338     }
 339
 340     private CreateTransactionReply createTransactionReply(ActorRef actorRef, int transactionVersion){
 341         return CreateTransactionReply.newBuilder()
 342             .setTransactionActorPath(actorRef.path().toString())
 343             .setTransactionId("txn-1")
 344             .setMessageVersion(transactionVersion)
 345             .build();
 346     }
 347
 348     private ActorRef setupActorContextWithoutInitialCreateTransaction(ActorSystem actorSystem) {
 349         ActorRef actorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
 350         doReturn(actorSystem.actorSelection(actorRef.path())).
 351                 when(mockActorContext).actorSelection(actorRef.path().toString());
 352
 353         doReturn(Futures.successful(actorSystem.actorSelection(actorRef.path()))).
 354                 when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 355
 356         doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString());
 357
 358         doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit();
 359
 360         return actorRef;
 361     }
 362
 363     private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem,
 364             TransactionType type, int transactionVersion) {
 365         ActorRef actorRef = setupActorContextWithoutInitialCreateTransaction(actorSystem);
 366
 367         doReturn(Futures.successful(createTransactionReply(actorRef, transactionVersion))).when(mockActorContext).
 368                 executeOperationAsync(eq(actorSystem.actorSelection(actorRef.path())),
 369                         eqCreateTransaction(memberName, type));
 370
 371         return actorRef;
 372     }
 373
 374     private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) {
 375         return setupActorContextWithInitialCreateTransaction(actorSystem, type, DataStoreVersions.CURRENT_VERSION);
 376     }
 377
 378
 379     private void propagateReadFailedExceptionCause(CheckedFuture<?, ReadFailedException> future)
 380             throws Throwable {
 381
 382         try {
 383             future.checkedGet(5, TimeUnit.SECONDS);
 384             fail("Expected ReadFailedException");
 385         } catch(ReadFailedException e) {
 386             throw e.getCause();
 387         }
 388     }
 389
 390     @Test
 391     public void testRead() throws Exception {
 392         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 393
 394         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
 395
 396         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 397                 eq(actorSelection(actorRef)), eqSerializedReadData());
 398
 399         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
 400                 TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 401
 402         assertEquals("NormalizedNode isPresent", false, readOptional.isPresent());
 403
 404         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 405
 406         doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
 407                 eq(actorSelection(actorRef)), eqSerializedReadData());
 408
 409         readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 410
 411         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
 412
 413         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
 414     }
 415
 416     @Test(expected = ReadFailedException.class)
 417     public void testReadWithInvalidReplyMessageType() throws Exception {
 418         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 419
 420         doReturn(Futures.successful(new Object())).when(mockActorContext).
 421                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData());
 422
 423         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
 424
 425         transactionProxy.read(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS);
 426     }
 427
 428     @Test(expected = TestException.class)
 429     public void testReadWithAsyncRemoteOperatonFailure() throws Throwable {
 430         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 431
 432         doReturn(Futures.failed(new TestException())).when(mockActorContext).
 433                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData());
 434
 435         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
 436
 437         propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
 438     }
 439
 440     private void testExceptionOnInitialCreateTransaction(Exception exToThrow, Invoker invoker)
 441             throws Throwable {
 442         ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
 443
 444         if (exToThrow instanceof PrimaryNotFoundException) {
 445             doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString());
 446         } else {
 447             doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))).
 448                     when(mockActorContext).findPrimaryShardAsync(anyString());
 449         }
 450
 451         doReturn(Futures.failed(exToThrow)).when(mockActorContext).executeOperationAsync(
 452                 any(ActorSelection.class), any());
 453
 454         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
 455
 456         propagateReadFailedExceptionCause(invoker.invoke(transactionProxy));
 457     }
 458
 459     private void testReadWithExceptionOnInitialCreateTransaction(Exception exToThrow) throws Throwable {
 460         testExceptionOnInitialCreateTransaction(exToThrow, new Invoker() {
 461             @Override
 462             public CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception {
 463                 return proxy.read(TestModel.TEST_PATH);
 464             }
 465         });
 466     }
 467
 468     @Test(expected = PrimaryNotFoundException.class)
 469     public void testReadWhenAPrimaryNotFoundExceptionIsThrown() throws Throwable {
 470         testReadWithExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"));
 471     }
 472
 473     @Test(expected = TimeoutException.class)
 474     public void testReadWhenATimeoutExceptionIsThrown() throws Throwable {
 475         testReadWithExceptionOnInitialCreateTransaction(new TimeoutException("test",
 476                 new Exception("reason")));
 477     }
 478
 479     @Test(expected = TestException.class)
 480     public void testReadWhenAnyOtherExceptionIsThrown() throws Throwable {
 481         testReadWithExceptionOnInitialCreateTransaction(new TestException());
 482     }
 483
 484     @Test(expected = TestException.class)
 485     public void testReadWithPriorRecordingOperationFailure() throws Throwable {
 486         doReturn(dataStoreContextBuilder.shardBatchedModificationCount(2).build()).
 487                 when(mockActorContext).getDatastoreContext();
 488
 489         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 490
 491         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 492
 493         expectFailedBatchedModifications(actorRef);
 494
 495         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 496                 eq(actorSelection(actorRef)), eqSerializedReadData());
 497
 498         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
 499
 500         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 501
 502         transactionProxy.delete(TestModel.TEST_PATH);
 503
 504         try {
 505             propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
 506         } finally {
 507             verify(mockActorContext, times(0)).executeOperationAsync(
 508                     eq(actorSelection(actorRef)), eqSerializedReadData());
 509         }
 510     }
 511
 512     @Test
 513     public void testReadWithPriorRecordingOperationSuccessful() throws Throwable {
 514         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 515
 516         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 517
 518         expectBatchedModifications(actorRef, 1);
 519
 520         doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
 521                 eq(actorSelection(actorRef)), eqSerializedReadData());
 522
 523         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
 524
 525         transactionProxy.write(TestModel.TEST_PATH, expectedNode);
 526
 527         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
 528                 TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 529
 530         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
 531         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
 532
 533         InOrder inOrder = Mockito.inOrder(mockActorContext);
 534         inOrder.verify(mockActorContext).executeOperationAsync(
 535                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
 536
 537         inOrder.verify(mockActorContext).executeOperationAsync(
 538                 eq(actorSelection(actorRef)), eqSerializedReadData());
 539     }
 540
 541     @Test(expected=IllegalStateException.class)
 542     public void testReadPreConditionCheck() {
 543         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
 544         transactionProxy.read(TestModel.TEST_PATH);
 545     }
 546
 547     @Test(expected=IllegalArgumentException.class)
 548     public void testInvalidCreateTransactionReply() throws Throwable {
 549         ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
 550
 551         doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext).
 552             actorSelection(actorRef.path().toString());
 553
 554         doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))).
 555             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 556
 557         doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync(
 558             eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY));
 559
 560         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
 561
 562         propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
 563     }
 564
 565     @Test
 566     public void testExists() throws Exception {
 567         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 568
 569         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
 570
 571         doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync(
 572                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 573
 574         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
 575
 576         assertEquals("Exists response", false, exists);
 577
 578         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
 579                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 580
 581         exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
 582
 583         assertEquals("Exists response", true, exists);
 584     }
 585
 586     @Test(expected = PrimaryNotFoundException.class)
 587     public void testExistsWhenAPrimaryNotFoundExceptionIsThrown() throws Throwable {
 588         testExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"), new Invoker() {
 589             @Override
 590             public CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception {
 591                 return proxy.exists(TestModel.TEST_PATH);
 592             }
 593         });
 594     }
 595
 596     @Test(expected = ReadFailedException.class)
 597     public void testExistsWithInvalidReplyMessageType() throws Exception {
 598         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 599
 600         doReturn(Futures.successful(new Object())).when(mockActorContext).
 601                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists());
 602
 603         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 604                 READ_ONLY);
 605
 606         transactionProxy.exists(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS);
 607     }
 608
 609     @Test(expected = TestException.class)
 610     public void testExistsWithAsyncRemoteOperatonFailure() throws Throwable {
 611         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 612
 613         doReturn(Futures.failed(new TestException())).when(mockActorContext).
 614                 executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists());
 615
 616         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
 617
 618         propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH));
 619     }
 620
 621     @Test(expected = TestException.class)
 622     public void testExistsWithPriorRecordingOperationFailure() throws Throwable {
 623         doReturn(dataStoreContextBuilder.shardBatchedModificationCount(2).build()).
 624                 when(mockActorContext).getDatastoreContext();
 625
 626         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 627
 628         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 629
 630         expectFailedBatchedModifications(actorRef);
 631
 632         doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync(
 633                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 634
 635         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
 636                 READ_WRITE);
 637
 638         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 639
 640         transactionProxy.delete(TestModel.TEST_PATH);
 641
 642         try {
 643             propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH));
 644         } finally {
 645             verify(mockActorContext, times(0)).executeOperationAsync(
 646                     eq(actorSelection(actorRef)), eqSerializedDataExists());
 647         }
 648     }
 649
 650     @Test
 651     public void testExistsWithPriorRecordingOperationSuccessful() throws Throwable {
 652         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 653
 654         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 655
 656         expectBatchedModifications(actorRef, 1);
 657
 658         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
 659                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 660
 661         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
 662
 663         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 664
 665         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
 666
 667         assertEquals("Exists response", true, exists);
 668
 669         InOrder inOrder = Mockito.inOrder(mockActorContext);
 670         inOrder.verify(mockActorContext).executeOperationAsync(
 671                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
 672
 673         inOrder.verify(mockActorContext).executeOperationAsync(
 674                 eq(actorSelection(actorRef)), eqSerializedDataExists());
 675     }
 676
 677     @Test(expected=IllegalStateException.class)
 678     public void testExistsPreConditionCheck() {
 679         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
 680         transactionProxy.exists(TestModel.TEST_PATH);
 681     }
 682
 683     private void verifyRecordingOperationFutures(List<Future<Object>> futures,
 684             Class<?>... expResultTypes) throws Exception {
 685         assertEquals("getRecordingOperationFutures size", expResultTypes.length, futures.size());
 686
 687         int i = 0;
 688         for( Future<Object> future: futures) {
 689             assertNotNull("Recording operation Future is null", future);
 690
 691             Class<?> expResultType = expResultTypes[i++];
 692             if(Throwable.class.isAssignableFrom(expResultType)) {
 693                 try {
 694                     Await.result(future, Duration.create(5, TimeUnit.SECONDS));
 695                     fail("Expected exception from recording operation Future");
 696                 } catch(Exception e) {
 697                     // Expected
 698                 }
 699             } else {
 700                 assertEquals(String.format("Recording operation %d Future result type", i +1 ), expResultType,
 701                              Await.result(future, Duration.create(5, TimeUnit.SECONDS)).getClass());
 702             }
 703         }
 704     }
 705
 706     @Test
 707     public void testWrite() throws Exception {
 708         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 709
 710         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 711
 712         expectBatchedModifications(actorRef, 1);
 713         expectReadyTransaction(actorRef);
 714
 715         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
 716
 717         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 718
 719         // This sends the batched modification.
 720         transactionProxy.ready();
 721
 722         verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite));
 723
 724         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 725                 BatchedModificationsReply.class);
 726     }
 727
 728     @Test
 729     public void testWriteAfterAsyncRead() throws Throwable {
 730         ActorRef actorRef = setupActorContextWithoutInitialCreateTransaction(getSystem());
 731
 732         Promise<Object> createTxPromise = akka.dispatch.Futures.promise();
 733         doReturn(createTxPromise).when(mockActorContext).executeOperationAsync(
 734                 eq(getSystem().actorSelection(actorRef.path())),
 735                 eqCreateTransaction(memberName, READ_WRITE));
 736
 737         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 738                 eq(actorSelection(actorRef)), eqSerializedReadData());
 739
 740         expectBatchedModifications(actorRef, 1);
 741         expectReadyTransaction(actorRef);
 742
 743         final NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 744
 745         final TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
 746
 747         final CountDownLatch readComplete = new CountDownLatch(1);
 748         final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
 749         com.google.common.util.concurrent.Futures.addCallback(transactionProxy.read(TestModel.TEST_PATH),
 750                 new  FutureCallback<Optional<NormalizedNode<?, ?>>>() {
 751                     @Override
 752                     public void onSuccess(Optional<NormalizedNode<?, ?>> result) {
 753                         try {
 754                             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 755                         } catch (Exception e) {
 756                             caughtEx.set(e);
 757                         } finally {
 758                             readComplete.countDown();
 759                         }
 760                     }
 761
 762                     @Override
 763                     public void onFailure(Throwable t) {
 764                         caughtEx.set(t);
 765                         readComplete.countDown();
 766                     }
 767                 });
 768
 769         createTxPromise.success(createTransactionReply(actorRef, DataStoreVersions.CURRENT_VERSION));
 770
 771         Uninterruptibles.awaitUninterruptibly(readComplete, 5, TimeUnit.SECONDS);
 772
 773         if(caughtEx.get() != null) {
 774             throw caughtEx.get();
 775         }
 776
 777         // This sends the batched modification.
 778         transactionProxy.ready();
 779
 780         verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite));
 781
 782         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 783                 BatchedModificationsReply.class);
 784     }
 785
 786     @Test(expected=IllegalStateException.class)
 787     public void testWritePreConditionCheck() {
 788         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY);
 789         transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 790     }
 791
 792     @Test(expected=IllegalStateException.class)
 793     public void testWriteAfterReadyPreConditionCheck() {
 794         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
 795
 796         transactionProxy.ready();
 797
 798         transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 799     }
 800
 801     @Test
 802     public void testMerge() throws Exception {
 803         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 804
 805         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 806
 807         expectBatchedModifications(actorRef, 1);
 808         expectReadyTransaction(actorRef);
 809
 810         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
 811
 812         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 813
 814         // This sends the batched modification.
 815         transactionProxy.ready();
 816
 817         verifyOneBatchedModification(actorRef, new MergeModification(TestModel.TEST_PATH, nodeToWrite));
 818
 819         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 820                 BatchedModificationsReply.class);
 821     }
 822
 823     @Test
 824     public void testDelete() throws Exception {
 825         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 826
 827         expectBatchedModifications(actorRef, 1);
 828         expectReadyTransaction(actorRef);
 829
 830         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
 831
 832         transactionProxy.delete(TestModel.TEST_PATH);
 833
 834         // This sends the batched modification.
 835         transactionProxy.ready();
 836
 837         verifyOneBatchedModification(actorRef, new DeleteModification(TestModel.TEST_PATH));
 838
 839         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 840                 BatchedModificationsReply.class);
 841     }
 842
 843     private void verifyCohortFutures(ThreePhaseCommitCohortProxy proxy,
 844         Object... expReplies) throws Exception {
 845         assertEquals("getReadyOperationFutures size", expReplies.length,
 846                 proxy.getCohortFutures().size());
 847
 848         int i = 0;
 849         for( Future<ActorSelection> future: proxy.getCohortFutures()) {
 850             assertNotNull("Ready operation Future is null", future);
 851
 852             Object expReply = expReplies[i++];
 853             if(expReply instanceof ActorSelection) {
 854                 ActorSelection actual = Await.result(future, Duration.create(5, TimeUnit.SECONDS));
 855                 assertEquals("Cohort actor path", expReply, actual);
 856             } else {
 857                 // Expecting exception.
 858                 try {
 859                     Await.result(future, Duration.create(5, TimeUnit.SECONDS));
 860                     fail("Expected exception from ready operation Future");
 861                 } catch(Exception e) {
 862                     // Expected
 863                 }
 864             }
 865         }
 866     }
 867
 868     @Test
 869     public void testReady() throws Exception {
 870         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 871
 872         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 873
 874         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
 875                 eq(actorSelection(actorRef)), eqSerializedReadData());
 876
 877         expectBatchedModifications(actorRef, 1);
 878         expectReadyTransaction(actorRef);
 879
 880         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
 881
 882         transactionProxy.read(TestModel.TEST_PATH);
 883
 884         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 885
 886         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 887
 888         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 889
 890         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
 891
 892         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 893                 BatchedModificationsReply.class);
 894
 895         verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path()));
 896
 897         verify(mockActorContext).executeOperationAsync(eq(actorSelection(actorRef)),
 898                 isA(BatchedModifications.class));
 899     }
 900
 901     private ActorRef testCompatibilityWithHeliumVersion(short version) throws Exception {
 902         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(),
 903                 READ_WRITE, version);
 904
 905         NormalizedNode<?, ?> testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 906
 907         doReturn(readSerializedDataReply(testNode, version)).when(mockActorContext).executeOperationAsync(
 908                 eq(actorSelection(actorRef)), eqSerializedReadData());
 909
 910         doReturn(Futures.successful(new WriteDataReply().toSerializable(version))).when(mockActorContext).
 911                 executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyWriteData(testNode));
 912
 913         doReturn(Futures.successful(new MergeDataReply().toSerializable(version))).when(mockActorContext).
 914                 executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyMergeData(testNode));
 915
 916         doReturn(Futures.successful(new DeleteDataReply().toSerializable(version))).when(mockActorContext).
 917                 executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyDeleteData(TestModel.TEST_PATH));
 918
 919         expectReadyTransaction(actorRef);
 920
 921         doReturn(actorRef.path().toString()).when(mockActorContext).resolvePath(eq(actorRef.path().toString()),
 922                 eq(actorRef.path().toString()));
 923
 924         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
 925
 926         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(TestModel.TEST_PATH).
 927                 get(5, TimeUnit.SECONDS);
 928
 929         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
 930         assertEquals("Response NormalizedNode", testNode, readOptional.get());
 931
 932         transactionProxy.write(TestModel.TEST_PATH, testNode);
 933
 934         transactionProxy.merge(TestModel.TEST_PATH, testNode);
 935
 936         transactionProxy.delete(TestModel.TEST_PATH);
 937
 938         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 939
 940         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 941
 942         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
 943
 944         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
 945                 ShardTransactionMessages.WriteDataReply.class, ShardTransactionMessages.MergeDataReply.class,
 946                 ShardTransactionMessages.DeleteDataReply.class);
 947
 948         verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path()));
 949
 950         return actorRef;
 951     }
 952
 953     @Test
 954     public void testCompatibilityWithBaseHeliumVersion() throws Exception {
 955         ActorRef actorRef = testCompatibilityWithHeliumVersion(DataStoreVersions.BASE_HELIUM_VERSION);
 956
 957         verify(mockActorContext).resolvePath(eq(actorRef.path().toString()),
 958                 eq(actorRef.path().toString()));
 959     }
 960
 961     @Test
 962     public void testCompatibilityWithHeliumR1Version() throws Exception {
 963         ActorRef actorRef = testCompatibilityWithHeliumVersion(DataStoreVersions.HELIUM_1_VERSION);
 964
 965         verify(mockActorContext, Mockito.never()).resolvePath(eq(actorRef.path().toString()),
 966                 eq(actorRef.path().toString()));
 967     }
 968
 969     @Test
 970     public void testReadyWithRecordingOperationFailure() throws Exception {
 971         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 972
 973         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 974
 975         expectFailedBatchedModifications(actorRef);
 976
 977         expectReadyTransaction(actorRef);
 978
 979         doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString());
 980
 981         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
 982
 983         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 984
 985         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 986
 987         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 988
 989         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
 990
 991         verifyCohortFutures(proxy, TestException.class);
 992
 993         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), TestException.class);
 994     }
 995
 996     @Test
 997     public void testReadyWithReplyFailure() throws Exception {
 998         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 999
1000         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1001
1002         expectBatchedModifications(actorRef, 1);
1003
1004         doReturn(Futures.failed(new TestException())).when(mockActorContext).
1005                 executeOperationAsync(eq(actorSelection(actorRef)),
1006                         isA(ReadyTransaction.SERIALIZABLE_CLASS));
1007
1008         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
1009
1010         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
1011
1012         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
1013
1014         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
1015
1016         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
1017
1018         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
1019                 BatchedModificationsReply.class);
1020
1021         verifyCohortFutures(proxy, TestException.class);
1022     }
1023
1024     @Test
1025     public void testReadyWithInitialCreateTransactionFailure() throws Exception {
1026
1027         doReturn(Futures.failed(new PrimaryNotFoundException("mock"))).when(
1028                 mockActorContext).findPrimaryShardAsync(anyString());
1029
1030         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
1031
1032         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1033
1034         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
1035
1036         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1037
1038         transactionProxy.delete(TestModel.TEST_PATH);
1039
1040         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
1041
1042         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
1043
1044         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
1045
1046         verifyCohortFutures(proxy, PrimaryNotFoundException.class);
1047     }
1048
1049     @Test
1050     public void testReadyWithInvalidReplyMessageType() throws Exception {
1051         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
1052
1053         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1054
1055         expectBatchedModifications(actorRef, 1);
1056
1057         doReturn(Futures.successful(new Object())).when(mockActorContext).
1058                 executeOperationAsync(eq(actorSelection(actorRef)),
1059                         isA(ReadyTransaction.SERIALIZABLE_CLASS));
1060
1061         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
1062
1063         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1064
1065         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
1066
1067         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
1068
1069         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
1070
1071         verifyCohortFutures(proxy, IllegalArgumentException.class);
1072     }
1073
1074     @Test
1075     public void testGetIdentifier() {
1076         setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
1077         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
1078                 TransactionProxy.TransactionType.READ_ONLY);
1079
1080         Object id = transactionProxy.getIdentifier();
1081         assertNotNull("getIdentifier returned null", id);
1082         assertTrue("Invalid identifier: " + id, id.toString().startsWith(memberName));
1083     }
1084
1085     @Test
1086     public void testClose() throws Exception{
1087         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
1088
1089         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
1090                 eq(actorSelection(actorRef)), eqSerializedReadData());
1091
1092         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1093
1094         transactionProxy.read(TestModel.TEST_PATH);
1095
1096         transactionProxy.close();
1097
1098         verify(mockActorContext).sendOperationAsync(
1099                 eq(actorSelection(actorRef)), isA(CloseTransaction.SERIALIZABLE_CLASS));
1100     }
1101
1102
1103     /**
1104      * Method to test a local Tx actor. The Tx paths are matched to decide if the
1105      * Tx actor is local or not. This is done by mocking the Tx actor path
1106      * and the caller paths and ensuring that the paths have the remote-address format
1107      *
1108      * Note: Since the default akka provider for test is not a RemoteActorRefProvider,
1109      * the paths returned for the actors for all the tests are not qualified remote paths.
1110      * Hence are treated as non-local/remote actors. In short, all tests except
1111      * few below run for remote actors
1112      *
1113      * @throws Exception
1114      */
1115     @Test
1116     public void testLocalTxActorRead() throws Exception {
1117         ActorSystem actorSystem = getSystem();
1118         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1119
1120         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1121             when(mockActorContext).actorSelection(shardActorRef.path().toString());
1122
1123         doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1124             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1125
1126         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1127         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder()
1128             .setTransactionId("txn-1").setTransactionActorPath(actorPath).build();
1129
1130         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1131             executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1132                 eqCreateTransaction(memberName, READ_ONLY));
1133
1134         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
1135
1136         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,READ_ONLY);
1137
1138         // negative test case with null as the reply
1139         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
1140             any(ActorSelection.class), eqReadData());
1141
1142         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
1143             TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
1144
1145         assertEquals("NormalizedNode isPresent", false, readOptional.isPresent());
1146
1147         // test case with node as read data reply
1148         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1149
1150         doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
1151             any(ActorSelection.class), eqReadData());
1152
1153         readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
1154
1155         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1156
1157         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
1158
1159         // test for local data exists
1160         doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1161             any(ActorSelection.class), eqDataExists());
1162
1163         boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
1164
1165         assertEquals("Exists response", true, exists);
1166     }
1167
1168     @Test
1169     public void testLocalTxActorReady() throws Exception {
1170         ActorSystem actorSystem = getSystem();
1171         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1172
1173         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1174             when(mockActorContext).actorSelection(shardActorRef.path().toString());
1175
1176         doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1177             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1178
1179         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1180         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder().
1181             setTransactionId("txn-1").setTransactionActorPath(actorPath).
1182             setMessageVersion(DataStoreVersions.CURRENT_VERSION).build();
1183
1184         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1185             executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1186                 eqCreateTransaction(memberName, WRITE_ONLY));
1187
1188         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
1189
1190         doReturn(batchedModificationsReply(1)).when(mockActorContext).executeOperationAsync(
1191                 any(ActorSelection.class), isA(BatchedModifications.class));
1192
1193         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
1194
1195         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1196         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1197
1198         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
1199                 BatchedModificationsReply.class);
1200
1201         // testing ready
1202         doReturn(readyTxReply(shardActorRef.path().toString())).when(mockActorContext).executeOperationAsync(
1203             any(ActorSelection.class), isA(ReadyTransaction.class));
1204
1205         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
1206
1207         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
1208
1209         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
1210
1211         verifyCohortFutures(proxy, getSystem().actorSelection(shardActorRef.path()));
1212     }
1213
1214     private static interface TransactionProxyOperation {
1215         void run(TransactionProxy transactionProxy);
1216     }
1217
1218     private void throttleOperation(TransactionProxyOperation operation) {
1219         throttleOperation(operation, 1, true);
1220     }
1221
1222     private void throttleOperation(TransactionProxyOperation operation, int outstandingOpsLimit, boolean shardFound){
1223         ActorSystem actorSystem = getSystem();
1224         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1225
1226         doReturn(outstandingOpsLimit).when(mockActorContext).getTransactionOutstandingOperationLimit();
1227
1228         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1229                 when(mockActorContext).actorSelection(shardActorRef.path().toString());
1230
1231         if(shardFound) {
1232             doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1233                     when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1234         } else {
1235             doReturn(Futures.failed(new Exception("not found")))
1236                     .when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1237         }
1238
1239         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1240         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder().
1241                 setTransactionId("txn-1").setTransactionActorPath(actorPath).
1242                 setMessageVersion(DataStoreVersions.CURRENT_VERSION).build();
1243
1244         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1245                 executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1246                         eqCreateTransaction(memberName, READ_WRITE));
1247
1248         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
1249
1250         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1251
1252         long start = System.nanoTime();
1253
1254         operation.run(transactionProxy);
1255
1256         long end = System.nanoTime();
1257
1258         long expected = TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds());
1259         Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
1260                 expected, (end-start)), (end - start) > expected);
1261
1262     }
1263
1264     private void completeOperation(TransactionProxyOperation operation){
1265         completeOperation(operation, true);
1266     }
1267
1268     private void completeOperation(TransactionProxyOperation operation, boolean shardFound){
1269         ActorSystem actorSystem = getSystem();
1270         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1271
1272         doReturn(1).when(mockActorContext).getTransactionOutstandingOperationLimit();
1273
1274         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1275                 when(mockActorContext).actorSelection(shardActorRef.path().toString());
1276
1277         if(shardFound) {
1278             doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1279                     when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1280         } else {
1281             doReturn(Futures.failed(new Exception("not found")))
1282                     .when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1283         }
1284
1285         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1286         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder().
1287                 setTransactionId("txn-1").setTransactionActorPath(actorPath).
1288                 setMessageVersion(DataStoreVersions.CURRENT_VERSION).build();
1289
1290         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1291                 executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1292                         eqCreateTransaction(memberName, READ_WRITE));
1293
1294         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
1295
1296         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1297
1298         long start = System.nanoTime();
1299
1300         operation.run(transactionProxy);
1301
1302         long end = System.nanoTime();
1303
1304         long expected = TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds());
1305         Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
1306                 expected, (end-start)), (end - start) <= expected);
1307     }
1308
1309     public void testWriteThrottling(boolean shardFound){
1310
1311         throttleOperation(new TransactionProxyOperation() {
1312             @Override
1313             public void run(TransactionProxy transactionProxy) {
1314                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1315
1316                 expectBatchedModifications(2);
1317
1318                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1319
1320                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1321             }
1322         }, 1, shardFound);
1323     }
1324
1325     @Test
1326     public void testWriteThrottlingWhenShardFound(){
1327         throttleOperation(new TransactionProxyOperation() {
1328             @Override
1329             public void run(TransactionProxy transactionProxy) {
1330                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1331
1332                 expectIncompleteBatchedModifications();
1333
1334                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1335
1336                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1337             }
1338         });
1339     }
1340
1341     @Test
1342     public void testWriteThrottlingWhenShardNotFound(){
1343         // Confirm that there is no throttling when the Shard is not found
1344         completeOperation(new TransactionProxyOperation() {
1345             @Override
1346             public void run(TransactionProxy transactionProxy) {
1347                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1348
1349                 expectBatchedModifications(2);
1350
1351                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1352
1353                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1354             }
1355         }, false);
1356
1357     }
1358
1359
1360     @Test
1361     public void testWriteCompletion(){
1362         completeOperation(new TransactionProxyOperation() {
1363             @Override
1364             public void run(TransactionProxy transactionProxy) {
1365                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1366
1367                 expectBatchedModifications(2);
1368
1369                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1370
1371                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1372             }
1373         });
1374     }
1375
1376     @Test
1377     public void testMergeThrottlingWhenShardFound(){
1378
1379         throttleOperation(new TransactionProxyOperation() {
1380             @Override
1381             public void run(TransactionProxy transactionProxy) {
1382                 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1383
1384                 expectIncompleteBatchedModifications();
1385
1386                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1387
1388                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1389             }
1390         });
1391     }
1392
1393     @Test
1394     public void testMergeThrottlingWhenShardNotFound(){
1395
1396         completeOperation(new TransactionProxyOperation() {
1397             @Override
1398             public void run(TransactionProxy transactionProxy) {
1399                 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1400
1401                 expectBatchedModifications(2);
1402
1403                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1404
1405                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1406             }
1407         }, false);
1408     }
1409
1410     @Test
1411     public void testMergeCompletion(){
1412         completeOperation(new TransactionProxyOperation() {
1413             @Override
1414             public void run(TransactionProxy transactionProxy) {
1415                 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1416
1417                 expectBatchedModifications(2);
1418
1419                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1420
1421                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1422             }
1423         });
1424
1425     }
1426
1427     @Test
1428     public void testDeleteThrottlingWhenShardFound(){
1429
1430         throttleOperation(new TransactionProxyOperation() {
1431             @Override
1432             public void run(TransactionProxy transactionProxy) {
1433                 expectIncompleteBatchedModifications();
1434
1435                 transactionProxy.delete(TestModel.TEST_PATH);
1436
1437                 transactionProxy.delete(TestModel.TEST_PATH);
1438             }
1439         });
1440     }
1441
1442
1443     @Test
1444     public void testDeleteThrottlingWhenShardNotFound(){
1445
1446         completeOperation(new TransactionProxyOperation() {
1447             @Override
1448             public void run(TransactionProxy transactionProxy) {
1449                 expectBatchedModifications(2);
1450
1451                 transactionProxy.delete(TestModel.TEST_PATH);
1452
1453                 transactionProxy.delete(TestModel.TEST_PATH);
1454             }
1455         }, false);
1456     }
1457
1458     @Test
1459     public void testDeleteCompletion(){
1460         completeOperation(new TransactionProxyOperation() {
1461             @Override
1462             public void run(TransactionProxy transactionProxy) {
1463                 expectBatchedModifications(2);
1464
1465                 transactionProxy.delete(TestModel.TEST_PATH);
1466
1467                 transactionProxy.delete(TestModel.TEST_PATH);
1468             }
1469         });
1470
1471     }
1472
1473     @Test
1474     public void testReadThrottlingWhenShardFound(){
1475
1476         throttleOperation(new TransactionProxyOperation() {
1477             @Override
1478             public void run(TransactionProxy transactionProxy) {
1479                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1480                         any(ActorSelection.class), eqReadData());
1481
1482                 transactionProxy.read(TestModel.TEST_PATH);
1483
1484                 transactionProxy.read(TestModel.TEST_PATH);
1485             }
1486         });
1487     }
1488
1489     @Test
1490     public void testReadThrottlingWhenShardNotFound(){
1491
1492         completeOperation(new TransactionProxyOperation() {
1493             @Override
1494             public void run(TransactionProxy transactionProxy) {
1495                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1496                         any(ActorSelection.class), eqReadData());
1497
1498                 transactionProxy.read(TestModel.TEST_PATH);
1499
1500                 transactionProxy.read(TestModel.TEST_PATH);
1501             }
1502         }, false);
1503     }
1504
1505
1506     @Test
1507     public void testReadCompletion(){
1508         completeOperation(new TransactionProxyOperation() {
1509             @Override
1510             public void run(TransactionProxy transactionProxy) {
1511                 NormalizedNode<?, ?> nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1512
1513                 doReturn(readDataReply(nodeToRead)).when(mockActorContext).executeOperationAsync(
1514                         any(ActorSelection.class), eqReadData());
1515
1516                 transactionProxy.read(TestModel.TEST_PATH);
1517
1518                 transactionProxy.read(TestModel.TEST_PATH);
1519             }
1520         });
1521
1522     }
1523
1524     @Test
1525     public void testExistsThrottlingWhenShardFound(){
1526
1527         throttleOperation(new TransactionProxyOperation() {
1528             @Override
1529             public void run(TransactionProxy transactionProxy) {
1530                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1531                         any(ActorSelection.class), eqDataExists());
1532
1533                 transactionProxy.exists(TestModel.TEST_PATH);
1534
1535                 transactionProxy.exists(TestModel.TEST_PATH);
1536             }
1537         });
1538     }
1539
1540     @Test
1541     public void testExistsThrottlingWhenShardNotFound(){
1542
1543         completeOperation(new TransactionProxyOperation() {
1544             @Override
1545             public void run(TransactionProxy transactionProxy) {
1546                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1547                         any(ActorSelection.class), eqDataExists());
1548
1549                 transactionProxy.exists(TestModel.TEST_PATH);
1550
1551                 transactionProxy.exists(TestModel.TEST_PATH);
1552             }
1553         }, false);
1554     }
1555
1556
1557     @Test
1558     public void testExistsCompletion(){
1559         completeOperation(new TransactionProxyOperation() {
1560             @Override
1561             public void run(TransactionProxy transactionProxy) {
1562                 doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1563                         any(ActorSelection.class), eqDataExists());
1564
1565                 transactionProxy.exists(TestModel.TEST_PATH);
1566
1567                 transactionProxy.exists(TestModel.TEST_PATH);
1568             }
1569         });
1570
1571     }
1572
1573     @Test
1574     public void testReadyThrottling(){
1575
1576         throttleOperation(new TransactionProxyOperation() {
1577             @Override
1578             public void run(TransactionProxy transactionProxy) {
1579                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1580
1581                 expectBatchedModifications(1);
1582
1583                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1584                         any(ActorSelection.class), any(ReadyTransaction.class));
1585
1586                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1587
1588                 transactionProxy.ready();
1589             }
1590         });
1591     }
1592
1593     @Test
1594     public void testReadyThrottlingWithTwoTransactionContexts(){
1595
1596         throttleOperation(new TransactionProxyOperation() {
1597             @Override
1598             public void run(TransactionProxy transactionProxy) {
1599                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1600                 NormalizedNode<?, ?> carsNode = ImmutableNodes.containerNode(CarsModel.BASE_QNAME);
1601
1602                 expectBatchedModifications(2);
1603
1604                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1605                         any(ActorSelection.class), any(ReadyTransaction.class));
1606
1607                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1608
1609                 transactionProxy.write(TestModel.TEST_PATH, carsNode);
1610
1611                 transactionProxy.ready();
1612             }
1613         }, 2, true);
1614     }
1615
1616     @Test
1617     public void testModificationOperationBatching() throws Throwable {
1618         int shardBatchedModificationCount = 3;
1619         doReturn(dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount).build()).
1620                 when(mockActorContext).getDatastoreContext();
1621
1622         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
1623
1624         expectBatchedModifications(actorRef, shardBatchedModificationCount);
1625
1626         expectReadyTransaction(actorRef);
1627
1628         YangInstanceIdentifier writePath1 = TestModel.TEST_PATH;
1629         NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1630
1631         YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH;
1632         NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1633
1634         YangInstanceIdentifier writePath3 = TestModel.INNER_LIST_PATH;
1635         NormalizedNode<?, ?> writeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1636
1637         YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH;
1638         NormalizedNode<?, ?> mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1639
1640         YangInstanceIdentifier mergePath2 = TestModel.OUTER_LIST_PATH;
1641         NormalizedNode<?, ?> mergeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1642
1643         YangInstanceIdentifier mergePath3 = TestModel.INNER_LIST_PATH;
1644         NormalizedNode<?, ?> mergeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1645
1646         YangInstanceIdentifier deletePath1 = TestModel.TEST_PATH;
1647         YangInstanceIdentifier deletePath2 = TestModel.OUTER_LIST_PATH;
1648
1649         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1650
1651         transactionProxy.write(writePath1, writeNode1);
1652         transactionProxy.write(writePath2, writeNode2);
1653         transactionProxy.delete(deletePath1);
1654         transactionProxy.merge(mergePath1, mergeNode1);
1655         transactionProxy.merge(mergePath2, mergeNode2);
1656         transactionProxy.write(writePath3, writeNode3);
1657         transactionProxy.merge(mergePath3, mergeNode3);
1658         transactionProxy.delete(deletePath2);
1659
1660         // This sends the last batch.
1661         transactionProxy.ready();
1662
1663         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1664         assertEquals("Captured BatchedModifications count", 3, batchedModifications.size());
1665
1666         verifyBatchedModifications(batchedModifications.get(0), new WriteModification(writePath1, writeNode1),
1667                 new WriteModification(writePath2, writeNode2), new DeleteModification(deletePath1));
1668
1669         verifyBatchedModifications(batchedModifications.get(1), new MergeModification(mergePath1, mergeNode1),
1670                 new MergeModification(mergePath2, mergeNode2), new WriteModification(writePath3, writeNode3));
1671
1672         verifyBatchedModifications(batchedModifications.get(2), new MergeModification(mergePath3, mergeNode3),
1673                 new DeleteModification(deletePath2));
1674
1675         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
1676                 BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class);
1677     }
1678
1679     @Test
1680     public void testModificationOperationBatchingWithInterleavedReads() throws Throwable {
1681         int shardBatchedModificationCount = 10;
1682         doReturn(dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount).build()).
1683                 when(mockActorContext).getDatastoreContext();
1684
1685         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
1686
1687         expectBatchedModifications(actorRef, shardBatchedModificationCount);
1688
1689         YangInstanceIdentifier writePath1 = TestModel.TEST_PATH;
1690         NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1691
1692         YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH;
1693         NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1694
1695         YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH;
1696         NormalizedNode<?, ?> mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1697
1698         YangInstanceIdentifier mergePath2 = TestModel.INNER_LIST_PATH;
1699         NormalizedNode<?, ?> mergeNode2 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1700
1701         YangInstanceIdentifier deletePath = TestModel.OUTER_LIST_PATH;
1702
1703         doReturn(readSerializedDataReply(writeNode2)).when(mockActorContext).executeOperationAsync(
1704                 eq(actorSelection(actorRef)), eqSerializedReadData(writePath2));
1705
1706         doReturn(readSerializedDataReply(mergeNode2)).when(mockActorContext).executeOperationAsync(
1707                 eq(actorSelection(actorRef)), eqSerializedReadData(mergePath2));
1708
1709         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
1710                 eq(actorSelection(actorRef)), eqSerializedDataExists());
1711
1712         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1713
1714         transactionProxy.write(writePath1, writeNode1);
1715         transactionProxy.write(writePath2, writeNode2);
1716
1717         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(writePath2).
1718                 get(5, TimeUnit.SECONDS);
1719
1720         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1721         assertEquals("Response NormalizedNode", writeNode2, readOptional.get());
1722
1723         transactionProxy.merge(mergePath1, mergeNode1);
1724         transactionProxy.merge(mergePath2, mergeNode2);
1725
1726         readOptional = transactionProxy.read(mergePath2).get(5, TimeUnit.SECONDS);
1727
1728         transactionProxy.delete(deletePath);
1729
1730         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
1731         assertEquals("Exists response", true, exists);
1732
1733         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1734         assertEquals("Response NormalizedNode", mergeNode2, readOptional.get());
1735
1736         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1737         assertEquals("Captured BatchedModifications count", 3, batchedModifications.size());
1738
1739         verifyBatchedModifications(batchedModifications.get(0), new WriteModification(writePath1, writeNode1),
1740                 new WriteModification(writePath2, writeNode2));
1741
1742         verifyBatchedModifications(batchedModifications.get(1), new MergeModification(mergePath1, mergeNode1),
1743                 new MergeModification(mergePath2, mergeNode2));
1744
1745         verifyBatchedModifications(batchedModifications.get(2), new DeleteModification(deletePath));
1746
1747         InOrder inOrder = Mockito.inOrder(mockActorContext);
1748         inOrder.verify(mockActorContext).executeOperationAsync(
1749                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
1750
1751         inOrder.verify(mockActorContext).executeOperationAsync(
1752                 eq(actorSelection(actorRef)), eqSerializedReadData(writePath2));
1753
1754         inOrder.verify(mockActorContext).executeOperationAsync(
1755                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
1756
1757         inOrder.verify(mockActorContext).executeOperationAsync(
1758                 eq(actorSelection(actorRef)), eqSerializedReadData(mergePath2));
1759
1760         inOrder.verify(mockActorContext).executeOperationAsync(
1761                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
1762
1763         inOrder.verify(mockActorContext).executeOperationAsync(
1764                 eq(actorSelection(actorRef)), eqSerializedDataExists());
1765
1766         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
1767                 BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class);
1768     }
1769
1770     private List<BatchedModifications> captureBatchedModifications(ActorRef actorRef) {
1771         ArgumentCaptor<BatchedModifications> batchedModificationsCaptor =
1772                 ArgumentCaptor.forClass(BatchedModifications.class);
1773         verify(mockActorContext, Mockito.atLeastOnce()).executeOperationAsync(
1774                 eq(actorSelection(actorRef)), batchedModificationsCaptor.capture());
1775
1776         List<BatchedModifications> batchedModifications = filterCaptured(
1777                 batchedModificationsCaptor, BatchedModifications.class);
1778         return batchedModifications;
1779     }
1780
1781     private <T> List<T> filterCaptured(ArgumentCaptor<T> captor, Class<T> type) {
1782         List<T> captured = new ArrayList<>();
1783         for(T c: captor.getAllValues()) {
1784             if(type.isInstance(c)) {
1785                 captured.add(c);
1786             }
1787         }
1788
1789         return captured;
1790     }
1791
1792     private void verifyOneBatchedModification(ActorRef actorRef, Modification expected) {
1793         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1794         assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
1795
1796         verifyBatchedModifications(batchedModifications.get(0), expected);
1797     }
1798
1799     private void verifyBatchedModifications(Object message, Modification... expected) {
1800         assertEquals("Message type", BatchedModifications.class, message.getClass());
1801         BatchedModifications batchedModifications = (BatchedModifications)message;
1802         assertEquals("BatchedModifications size", expected.length, batchedModifications.getModifications().size());
1803         for(int i = 0; i < batchedModifications.getModifications().size(); i++) {
1804             Modification actual = batchedModifications.getModifications().get(i);
1805             assertEquals("Modification type", expected[i].getClass(), actual.getClass());
1806             assertEquals("getPath", ((AbstractModification)expected[i]).getPath(),
1807                     ((AbstractModification)actual).getPath());
1808             if(actual instanceof WriteModification) {
1809                 assertEquals("getData", ((WriteModification)expected[i]).getData(),
1810                         ((WriteModification)actual).getData());
1811             }
1812         }
1813     }
1814 }