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 testUnusedTransaction() throws Exception {
1076         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1077
1078         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
1079
1080         assertEquals("canCommit", true, ready.canCommit().get());
1081         ready.preCommit().get();
1082         ready.commit().get();
1083     }
1084
1085     @Test
1086     public void testGetIdentifier() {
1087         setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
1088         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,
1089                 TransactionProxy.TransactionType.READ_ONLY);
1090
1091         Object id = transactionProxy.getIdentifier();
1092         assertNotNull("getIdentifier returned null", id);
1093         assertTrue("Invalid identifier: " + id, id.toString().startsWith(memberName));
1094     }
1095
1096     @Test
1097     public void testClose() throws Exception{
1098         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
1099
1100         doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync(
1101                 eq(actorSelection(actorRef)), eqSerializedReadData());
1102
1103         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1104
1105         transactionProxy.read(TestModel.TEST_PATH);
1106
1107         transactionProxy.close();
1108
1109         verify(mockActorContext).sendOperationAsync(
1110                 eq(actorSelection(actorRef)), isA(CloseTransaction.SERIALIZABLE_CLASS));
1111     }
1112
1113
1114     /**
1115      * Method to test a local Tx actor. The Tx paths are matched to decide if the
1116      * Tx actor is local or not. This is done by mocking the Tx actor path
1117      * and the caller paths and ensuring that the paths have the remote-address format
1118      *
1119      * Note: Since the default akka provider for test is not a RemoteActorRefProvider,
1120      * the paths returned for the actors for all the tests are not qualified remote paths.
1121      * Hence are treated as non-local/remote actors. In short, all tests except
1122      * few below run for remote actors
1123      *
1124      * @throws Exception
1125      */
1126     @Test
1127     public void testLocalTxActorRead() throws Exception {
1128         ActorSystem actorSystem = getSystem();
1129         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1130
1131         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1132             when(mockActorContext).actorSelection(shardActorRef.path().toString());
1133
1134         doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1135             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1136
1137         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1138         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder()
1139             .setTransactionId("txn-1").setTransactionActorPath(actorPath).build();
1140
1141         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1142             executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1143                 eqCreateTransaction(memberName, READ_ONLY));
1144
1145         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
1146
1147         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,READ_ONLY);
1148
1149         // negative test case with null as the reply
1150         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
1151             any(ActorSelection.class), eqReadData());
1152
1153         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
1154             TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
1155
1156         assertEquals("NormalizedNode isPresent", false, readOptional.isPresent());
1157
1158         // test case with node as read data reply
1159         NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1160
1161         doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
1162             any(ActorSelection.class), eqReadData());
1163
1164         readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
1165
1166         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1167
1168         assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
1169
1170         // test for local data exists
1171         doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1172             any(ActorSelection.class), eqDataExists());
1173
1174         boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
1175
1176         assertEquals("Exists response", true, exists);
1177     }
1178
1179     @Test
1180     public void testLocalTxActorReady() throws Exception {
1181         ActorSystem actorSystem = getSystem();
1182         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1183
1184         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1185             when(mockActorContext).actorSelection(shardActorRef.path().toString());
1186
1187         doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1188             when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1189
1190         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1191         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder().
1192             setTransactionId("txn-1").setTransactionActorPath(actorPath).
1193             setMessageVersion(DataStoreVersions.CURRENT_VERSION).build();
1194
1195         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1196             executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1197                 eqCreateTransaction(memberName, WRITE_ONLY));
1198
1199         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
1200
1201         doReturn(batchedModificationsReply(1)).when(mockActorContext).executeOperationAsync(
1202                 any(ActorSelection.class), isA(BatchedModifications.class));
1203
1204         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY);
1205
1206         NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1207         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1208
1209         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
1210                 BatchedModificationsReply.class);
1211
1212         // testing ready
1213         doReturn(readyTxReply(shardActorRef.path().toString())).when(mockActorContext).executeOperationAsync(
1214             any(ActorSelection.class), isA(ReadyTransaction.class));
1215
1216         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
1217
1218         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
1219
1220         ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready;
1221
1222         verifyCohortFutures(proxy, getSystem().actorSelection(shardActorRef.path()));
1223     }
1224
1225     private static interface TransactionProxyOperation {
1226         void run(TransactionProxy transactionProxy);
1227     }
1228
1229     private void throttleOperation(TransactionProxyOperation operation) {
1230         throttleOperation(operation, 1, true);
1231     }
1232
1233     private void throttleOperation(TransactionProxyOperation operation, int outstandingOpsLimit, boolean shardFound){
1234         ActorSystem actorSystem = getSystem();
1235         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1236
1237         doReturn(outstandingOpsLimit).when(mockActorContext).getTransactionOutstandingOperationLimit();
1238
1239         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1240                 when(mockActorContext).actorSelection(shardActorRef.path().toString());
1241
1242         if(shardFound) {
1243             doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1244                     when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1245         } else {
1246             doReturn(Futures.failed(new Exception("not found")))
1247                     .when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1248         }
1249
1250         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1251         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder().
1252                 setTransactionId("txn-1").setTransactionActorPath(actorPath).
1253                 setMessageVersion(DataStoreVersions.CURRENT_VERSION).build();
1254
1255         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1256                 executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1257                         eqCreateTransaction(memberName, READ_WRITE));
1258
1259         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
1260
1261         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1262
1263         long start = System.nanoTime();
1264
1265         operation.run(transactionProxy);
1266
1267         long end = System.nanoTime();
1268
1269         long expected = TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds());
1270         Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
1271                 expected, (end-start)), (end - start) > expected);
1272
1273     }
1274
1275     private void completeOperation(TransactionProxyOperation operation){
1276         completeOperation(operation, true);
1277     }
1278
1279     private void completeOperation(TransactionProxyOperation operation, boolean shardFound){
1280         ActorSystem actorSystem = getSystem();
1281         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1282
1283         doReturn(1).when(mockActorContext).getTransactionOutstandingOperationLimit();
1284
1285         doReturn(actorSystem.actorSelection(shardActorRef.path())).
1286                 when(mockActorContext).actorSelection(shardActorRef.path().toString());
1287
1288         if(shardFound) {
1289             doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
1290                     when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1291         } else {
1292             doReturn(Futures.failed(new Exception("not found")))
1293                     .when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
1294         }
1295
1296         String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor";
1297         CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder().
1298                 setTransactionId("txn-1").setTransactionActorPath(actorPath).
1299                 setMessageVersion(DataStoreVersions.CURRENT_VERSION).build();
1300
1301         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).
1302                 executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1303                         eqCreateTransaction(memberName, READ_WRITE));
1304
1305         doReturn(true).when(mockActorContext).isPathLocal(actorPath);
1306
1307         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1308
1309         long start = System.nanoTime();
1310
1311         operation.run(transactionProxy);
1312
1313         long end = System.nanoTime();
1314
1315         long expected = TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds());
1316         Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
1317                 expected, (end-start)), (end - start) <= expected);
1318     }
1319
1320     public void testWriteThrottling(boolean shardFound){
1321
1322         throttleOperation(new TransactionProxyOperation() {
1323             @Override
1324             public void run(TransactionProxy transactionProxy) {
1325                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1326
1327                 expectBatchedModifications(2);
1328
1329                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1330
1331                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1332             }
1333         }, 1, shardFound);
1334     }
1335
1336     @Test
1337     public void testWriteThrottlingWhenShardFound(){
1338         throttleOperation(new TransactionProxyOperation() {
1339             @Override
1340             public void run(TransactionProxy transactionProxy) {
1341                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1342
1343                 expectIncompleteBatchedModifications();
1344
1345                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1346
1347                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1348             }
1349         });
1350     }
1351
1352     @Test
1353     public void testWriteThrottlingWhenShardNotFound(){
1354         // Confirm that there is no throttling when the Shard is not found
1355         completeOperation(new TransactionProxyOperation() {
1356             @Override
1357             public void run(TransactionProxy transactionProxy) {
1358                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1359
1360                 expectBatchedModifications(2);
1361
1362                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1363
1364                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1365             }
1366         }, false);
1367
1368     }
1369
1370
1371     @Test
1372     public void testWriteCompletion(){
1373         completeOperation(new TransactionProxyOperation() {
1374             @Override
1375             public void run(TransactionProxy transactionProxy) {
1376                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1377
1378                 expectBatchedModifications(2);
1379
1380                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1381
1382                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1383             }
1384         });
1385     }
1386
1387     @Test
1388     public void testMergeThrottlingWhenShardFound(){
1389
1390         throttleOperation(new TransactionProxyOperation() {
1391             @Override
1392             public void run(TransactionProxy transactionProxy) {
1393                 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1394
1395                 expectIncompleteBatchedModifications();
1396
1397                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1398
1399                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1400             }
1401         });
1402     }
1403
1404     @Test
1405     public void testMergeThrottlingWhenShardNotFound(){
1406
1407         completeOperation(new TransactionProxyOperation() {
1408             @Override
1409             public void run(TransactionProxy transactionProxy) {
1410                 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1411
1412                 expectBatchedModifications(2);
1413
1414                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1415
1416                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1417             }
1418         }, false);
1419     }
1420
1421     @Test
1422     public void testMergeCompletion(){
1423         completeOperation(new TransactionProxyOperation() {
1424             @Override
1425             public void run(TransactionProxy transactionProxy) {
1426                 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1427
1428                 expectBatchedModifications(2);
1429
1430                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1431
1432                 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1433             }
1434         });
1435
1436     }
1437
1438     @Test
1439     public void testDeleteThrottlingWhenShardFound(){
1440
1441         throttleOperation(new TransactionProxyOperation() {
1442             @Override
1443             public void run(TransactionProxy transactionProxy) {
1444                 expectIncompleteBatchedModifications();
1445
1446                 transactionProxy.delete(TestModel.TEST_PATH);
1447
1448                 transactionProxy.delete(TestModel.TEST_PATH);
1449             }
1450         });
1451     }
1452
1453
1454     @Test
1455     public void testDeleteThrottlingWhenShardNotFound(){
1456
1457         completeOperation(new TransactionProxyOperation() {
1458             @Override
1459             public void run(TransactionProxy transactionProxy) {
1460                 expectBatchedModifications(2);
1461
1462                 transactionProxy.delete(TestModel.TEST_PATH);
1463
1464                 transactionProxy.delete(TestModel.TEST_PATH);
1465             }
1466         }, false);
1467     }
1468
1469     @Test
1470     public void testDeleteCompletion(){
1471         completeOperation(new TransactionProxyOperation() {
1472             @Override
1473             public void run(TransactionProxy transactionProxy) {
1474                 expectBatchedModifications(2);
1475
1476                 transactionProxy.delete(TestModel.TEST_PATH);
1477
1478                 transactionProxy.delete(TestModel.TEST_PATH);
1479             }
1480         });
1481
1482     }
1483
1484     @Test
1485     public void testReadThrottlingWhenShardFound(){
1486
1487         throttleOperation(new TransactionProxyOperation() {
1488             @Override
1489             public void run(TransactionProxy transactionProxy) {
1490                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1491                         any(ActorSelection.class), eqReadData());
1492
1493                 transactionProxy.read(TestModel.TEST_PATH);
1494
1495                 transactionProxy.read(TestModel.TEST_PATH);
1496             }
1497         });
1498     }
1499
1500     @Test
1501     public void testReadThrottlingWhenShardNotFound(){
1502
1503         completeOperation(new TransactionProxyOperation() {
1504             @Override
1505             public void run(TransactionProxy transactionProxy) {
1506                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1507                         any(ActorSelection.class), eqReadData());
1508
1509                 transactionProxy.read(TestModel.TEST_PATH);
1510
1511                 transactionProxy.read(TestModel.TEST_PATH);
1512             }
1513         }, false);
1514     }
1515
1516
1517     @Test
1518     public void testReadCompletion(){
1519         completeOperation(new TransactionProxyOperation() {
1520             @Override
1521             public void run(TransactionProxy transactionProxy) {
1522                 NormalizedNode<?, ?> nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1523
1524                 doReturn(readDataReply(nodeToRead)).when(mockActorContext).executeOperationAsync(
1525                         any(ActorSelection.class), eqReadData());
1526
1527                 transactionProxy.read(TestModel.TEST_PATH);
1528
1529                 transactionProxy.read(TestModel.TEST_PATH);
1530             }
1531         });
1532
1533     }
1534
1535     @Test
1536     public void testExistsThrottlingWhenShardFound(){
1537
1538         throttleOperation(new TransactionProxyOperation() {
1539             @Override
1540             public void run(TransactionProxy transactionProxy) {
1541                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1542                         any(ActorSelection.class), eqDataExists());
1543
1544                 transactionProxy.exists(TestModel.TEST_PATH);
1545
1546                 transactionProxy.exists(TestModel.TEST_PATH);
1547             }
1548         });
1549     }
1550
1551     @Test
1552     public void testExistsThrottlingWhenShardNotFound(){
1553
1554         completeOperation(new TransactionProxyOperation() {
1555             @Override
1556             public void run(TransactionProxy transactionProxy) {
1557                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1558                         any(ActorSelection.class), eqDataExists());
1559
1560                 transactionProxy.exists(TestModel.TEST_PATH);
1561
1562                 transactionProxy.exists(TestModel.TEST_PATH);
1563             }
1564         }, false);
1565     }
1566
1567
1568     @Test
1569     public void testExistsCompletion(){
1570         completeOperation(new TransactionProxyOperation() {
1571             @Override
1572             public void run(TransactionProxy transactionProxy) {
1573                 doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1574                         any(ActorSelection.class), eqDataExists());
1575
1576                 transactionProxy.exists(TestModel.TEST_PATH);
1577
1578                 transactionProxy.exists(TestModel.TEST_PATH);
1579             }
1580         });
1581
1582     }
1583
1584     @Test
1585     public void testReadyThrottling(){
1586
1587         throttleOperation(new TransactionProxyOperation() {
1588             @Override
1589             public void run(TransactionProxy transactionProxy) {
1590                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1591
1592                 expectBatchedModifications(1);
1593
1594                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1595                         any(ActorSelection.class), any(ReadyTransaction.class));
1596
1597                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1598
1599                 transactionProxy.ready();
1600             }
1601         });
1602     }
1603
1604     @Test
1605     public void testReadyThrottlingWithTwoTransactionContexts(){
1606
1607         throttleOperation(new TransactionProxyOperation() {
1608             @Override
1609             public void run(TransactionProxy transactionProxy) {
1610                 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1611                 NormalizedNode<?, ?> carsNode = ImmutableNodes.containerNode(CarsModel.BASE_QNAME);
1612
1613                 expectBatchedModifications(2);
1614
1615                 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1616                         any(ActorSelection.class), any(ReadyTransaction.class));
1617
1618                 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1619
1620                 transactionProxy.write(TestModel.TEST_PATH, carsNode);
1621
1622                 transactionProxy.ready();
1623             }
1624         }, 2, true);
1625     }
1626
1627     @Test
1628     public void testModificationOperationBatching() throws Throwable {
1629         int shardBatchedModificationCount = 3;
1630         doReturn(dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount).build()).
1631                 when(mockActorContext).getDatastoreContext();
1632
1633         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
1634
1635         expectBatchedModifications(actorRef, shardBatchedModificationCount);
1636
1637         expectReadyTransaction(actorRef);
1638
1639         YangInstanceIdentifier writePath1 = TestModel.TEST_PATH;
1640         NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1641
1642         YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH;
1643         NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1644
1645         YangInstanceIdentifier writePath3 = TestModel.INNER_LIST_PATH;
1646         NormalizedNode<?, ?> writeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1647
1648         YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH;
1649         NormalizedNode<?, ?> mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1650
1651         YangInstanceIdentifier mergePath2 = TestModel.OUTER_LIST_PATH;
1652         NormalizedNode<?, ?> mergeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1653
1654         YangInstanceIdentifier mergePath3 = TestModel.INNER_LIST_PATH;
1655         NormalizedNode<?, ?> mergeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1656
1657         YangInstanceIdentifier deletePath1 = TestModel.TEST_PATH;
1658         YangInstanceIdentifier deletePath2 = TestModel.OUTER_LIST_PATH;
1659
1660         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1661
1662         transactionProxy.write(writePath1, writeNode1);
1663         transactionProxy.write(writePath2, writeNode2);
1664         transactionProxy.delete(deletePath1);
1665         transactionProxy.merge(mergePath1, mergeNode1);
1666         transactionProxy.merge(mergePath2, mergeNode2);
1667         transactionProxy.write(writePath3, writeNode3);
1668         transactionProxy.merge(mergePath3, mergeNode3);
1669         transactionProxy.delete(deletePath2);
1670
1671         // This sends the last batch.
1672         transactionProxy.ready();
1673
1674         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1675         assertEquals("Captured BatchedModifications count", 3, batchedModifications.size());
1676
1677         verifyBatchedModifications(batchedModifications.get(0), new WriteModification(writePath1, writeNode1),
1678                 new WriteModification(writePath2, writeNode2), new DeleteModification(deletePath1));
1679
1680         verifyBatchedModifications(batchedModifications.get(1), new MergeModification(mergePath1, mergeNode1),
1681                 new MergeModification(mergePath2, mergeNode2), new WriteModification(writePath3, writeNode3));
1682
1683         verifyBatchedModifications(batchedModifications.get(2), new MergeModification(mergePath3, mergeNode3),
1684                 new DeleteModification(deletePath2));
1685
1686         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
1687                 BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class);
1688     }
1689
1690     @Test
1691     public void testModificationOperationBatchingWithInterleavedReads() throws Throwable {
1692         int shardBatchedModificationCount = 10;
1693         doReturn(dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount).build()).
1694                 when(mockActorContext).getDatastoreContext();
1695
1696         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
1697
1698         expectBatchedModifications(actorRef, shardBatchedModificationCount);
1699
1700         YangInstanceIdentifier writePath1 = TestModel.TEST_PATH;
1701         NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1702
1703         YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH;
1704         NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1705
1706         YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH;
1707         NormalizedNode<?, ?> mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1708
1709         YangInstanceIdentifier mergePath2 = TestModel.INNER_LIST_PATH;
1710         NormalizedNode<?, ?> mergeNode2 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1711
1712         YangInstanceIdentifier deletePath = TestModel.OUTER_LIST_PATH;
1713
1714         doReturn(readSerializedDataReply(writeNode2)).when(mockActorContext).executeOperationAsync(
1715                 eq(actorSelection(actorRef)), eqSerializedReadData(writePath2));
1716
1717         doReturn(readSerializedDataReply(mergeNode2)).when(mockActorContext).executeOperationAsync(
1718                 eq(actorSelection(actorRef)), eqSerializedReadData(mergePath2));
1719
1720         doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync(
1721                 eq(actorSelection(actorRef)), eqSerializedDataExists());
1722
1723         TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE);
1724
1725         transactionProxy.write(writePath1, writeNode1);
1726         transactionProxy.write(writePath2, writeNode2);
1727
1728         Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(writePath2).
1729                 get(5, TimeUnit.SECONDS);
1730
1731         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1732         assertEquals("Response NormalizedNode", writeNode2, readOptional.get());
1733
1734         transactionProxy.merge(mergePath1, mergeNode1);
1735         transactionProxy.merge(mergePath2, mergeNode2);
1736
1737         readOptional = transactionProxy.read(mergePath2).get(5, TimeUnit.SECONDS);
1738
1739         transactionProxy.delete(deletePath);
1740
1741         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
1742         assertEquals("Exists response", true, exists);
1743
1744         assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1745         assertEquals("Response NormalizedNode", mergeNode2, readOptional.get());
1746
1747         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1748         assertEquals("Captured BatchedModifications count", 3, batchedModifications.size());
1749
1750         verifyBatchedModifications(batchedModifications.get(0), new WriteModification(writePath1, writeNode1),
1751                 new WriteModification(writePath2, writeNode2));
1752
1753         verifyBatchedModifications(batchedModifications.get(1), new MergeModification(mergePath1, mergeNode1),
1754                 new MergeModification(mergePath2, mergeNode2));
1755
1756         verifyBatchedModifications(batchedModifications.get(2), new DeleteModification(deletePath));
1757
1758         InOrder inOrder = Mockito.inOrder(mockActorContext);
1759         inOrder.verify(mockActorContext).executeOperationAsync(
1760                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
1761
1762         inOrder.verify(mockActorContext).executeOperationAsync(
1763                 eq(actorSelection(actorRef)), eqSerializedReadData(writePath2));
1764
1765         inOrder.verify(mockActorContext).executeOperationAsync(
1766                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
1767
1768         inOrder.verify(mockActorContext).executeOperationAsync(
1769                 eq(actorSelection(actorRef)), eqSerializedReadData(mergePath2));
1770
1771         inOrder.verify(mockActorContext).executeOperationAsync(
1772                 eq(actorSelection(actorRef)), isA(BatchedModifications.class));
1773
1774         inOrder.verify(mockActorContext).executeOperationAsync(
1775                 eq(actorSelection(actorRef)), eqSerializedDataExists());
1776
1777         verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(),
1778                 BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class);
1779     }
1780
1781     private List<BatchedModifications> captureBatchedModifications(ActorRef actorRef) {
1782         ArgumentCaptor<BatchedModifications> batchedModificationsCaptor =
1783                 ArgumentCaptor.forClass(BatchedModifications.class);
1784         verify(mockActorContext, Mockito.atLeastOnce()).executeOperationAsync(
1785                 eq(actorSelection(actorRef)), batchedModificationsCaptor.capture());
1786
1787         List<BatchedModifications> batchedModifications = filterCaptured(
1788                 batchedModificationsCaptor, BatchedModifications.class);
1789         return batchedModifications;
1790     }
1791
1792     private <T> List<T> filterCaptured(ArgumentCaptor<T> captor, Class<T> type) {
1793         List<T> captured = new ArrayList<>();
1794         for(T c: captor.getAllValues()) {
1795             if(type.isInstance(c)) {
1796                 captured.add(c);
1797             }
1798         }
1799
1800         return captured;
1801     }
1802
1803     private void verifyOneBatchedModification(ActorRef actorRef, Modification expected) {
1804         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1805         assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
1806
1807         verifyBatchedModifications(batchedModifications.get(0), expected);
1808     }
1809
1810     private void verifyBatchedModifications(Object message, Modification... expected) {
1811         assertEquals("Message type", BatchedModifications.class, message.getClass());
1812         BatchedModifications batchedModifications = (BatchedModifications)message;
1813         assertEquals("BatchedModifications size", expected.length, batchedModifications.getModifications().size());
1814         for(int i = 0; i < batchedModifications.getModifications().size(); i++) {
1815             Modification actual = batchedModifications.getModifications().get(i);
1816             assertEquals("Modification type", expected[i].getClass(), actual.getClass());
1817             assertEquals("getPath", ((AbstractModification)expected[i]).getPath(),
1818                     ((AbstractModification)actual).getPath());
1819             if(actual instanceof WriteModification) {
1820                 assertEquals("getData", ((WriteModification)expected[i]).getData(),
1821                         ((WriteModification)actual).getData());
1822             }
1823         }
1824     }
1825 }