- private static ActorSystem system;
-
- private final Configuration configuration = new MockConfiguration();
-
- @Mock
- private ActorContext mockActorContext;
-
- private SchemaContext schemaContext;
-
- @Mock
- private ClusterWrapper mockClusterWrapper;
-
- String memberName = "mock-member";
-
- @BeforeClass
- public static void setUpClass() throws IOException {
-
- Config config = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder().
- put("akka.actor.default-dispatcher.type",
- "akka.testkit.CallingThreadDispatcherConfigurator").build()).
- withFallback(ConfigFactory.load());
- system = ActorSystem.create("test", config);
- }
-
- @AfterClass
- public static void tearDownClass() throws IOException {
- JavaTestKit.shutdownActorSystem(system);
- system = null;
- }
-
- @Before
- public void setUp(){
- MockitoAnnotations.initMocks(this);
-
- schemaContext = TestModel.createTestContext();
-
- DatastoreContext dataStoreContext = DatastoreContext.newBuilder().operationTimeoutInSeconds(2).build();
-
- doReturn(getSystem()).when(mockActorContext).getActorSystem();
- doReturn(memberName).when(mockActorContext).getCurrentMemberName();
- doReturn(schemaContext).when(mockActorContext).getSchemaContext();
- doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper();
- doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper();
- doReturn(dataStoreContext).when(mockActorContext).getDatastoreContext();
- doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit();
-
- ShardStrategyFactory.setConfiguration(configuration);
- }
-
- private ActorSystem getSystem() {
- return system;
- }
-
- private CreateTransaction eqCreateTransaction(final String memberName,
- final TransactionType type) {
- ArgumentMatcher<CreateTransaction> matcher = new ArgumentMatcher<CreateTransaction>() {
- @Override
- public boolean matches(Object argument) {
- if(CreateTransaction.SERIALIZABLE_CLASS.equals(argument.getClass())) {
- CreateTransaction obj = CreateTransaction.fromSerializable(argument);
- return obj.getTransactionId().startsWith(memberName) &&
- obj.getTransactionType() == type.ordinal();
- }
-
- return false;
- }
- };
-
- return argThat(matcher);
- }
-
- private DataExists eqSerializedDataExists() {
- ArgumentMatcher<DataExists> matcher = new ArgumentMatcher<DataExists>() {
- @Override
- public boolean matches(Object argument) {
- return DataExists.SERIALIZABLE_CLASS.equals(argument.getClass()) &&
- DataExists.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH);
- }
- };
-
- return argThat(matcher);
- }
-
- private DataExists eqDataExists() {
- ArgumentMatcher<DataExists> matcher = new ArgumentMatcher<DataExists>() {
- @Override
- public boolean matches(Object argument) {
- return (argument instanceof DataExists) &&
- ((DataExists)argument).getPath().equals(TestModel.TEST_PATH);
- }
- };
-
- return argThat(matcher);
- }
-
- private ReadData eqSerializedReadData() {
- ArgumentMatcher<ReadData> matcher = new ArgumentMatcher<ReadData>() {
- @Override
- public boolean matches(Object argument) {
- return ReadData.SERIALIZABLE_CLASS.equals(argument.getClass()) &&
- ReadData.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH);
- }
- };
-
- return argThat(matcher);
- }
-
- private ReadData eqReadData() {
- ArgumentMatcher<ReadData> matcher = new ArgumentMatcher<ReadData>() {
- @Override
- public boolean matches(Object argument) {
- return (argument instanceof ReadData) &&
- ((ReadData)argument).getPath().equals(TestModel.TEST_PATH);
- }
- };
-
- return argThat(matcher);
- }
-
- private WriteData eqSerializedWriteData(final NormalizedNode<?, ?> nodeToWrite) {
- return eqSerializedWriteData(nodeToWrite, DataStoreVersions.CURRENT_VERSION);
- }
-
- private WriteData eqSerializedWriteData(final NormalizedNode<?, ?> nodeToWrite,
- final int transactionVersion) {
- ArgumentMatcher<WriteData> matcher = new ArgumentMatcher<WriteData>() {
- @Override
- public boolean matches(Object argument) {
- if((transactionVersion >= DataStoreVersions.LITHIUM_VERSION &&
- WriteData.SERIALIZABLE_CLASS.equals(argument.getClass())) ||
- (transactionVersion < DataStoreVersions.LITHIUM_VERSION &&
- ShardTransactionMessages.WriteData.class.equals(argument.getClass()))) {
-
- WriteData obj = WriteData.fromSerializable(argument);
- return obj.getPath().equals(TestModel.TEST_PATH) &&
- obj.getData().equals(nodeToWrite);
- }
-
- return false;
- }
- };
-
- return argThat(matcher);
- }
-
- private WriteData eqWriteData(final NormalizedNode<?, ?> nodeToWrite) {
- ArgumentMatcher<WriteData> matcher = new ArgumentMatcher<WriteData>() {
- @Override
- public boolean matches(Object argument) {
- if(argument instanceof WriteData) {
- WriteData obj = (WriteData) argument;
- return obj.getPath().equals(TestModel.TEST_PATH) &&
- obj.getData().equals(nodeToWrite);
- }
- return false;
- }
- };
-
- return argThat(matcher);
- }
-
- private MergeData eqSerializedMergeData(final NormalizedNode<?, ?> nodeToWrite) {
- return eqSerializedMergeData(nodeToWrite, DataStoreVersions.CURRENT_VERSION);
- }
-
- private MergeData eqSerializedMergeData(final NormalizedNode<?, ?> nodeToWrite,
- final int transactionVersion) {
- ArgumentMatcher<MergeData> matcher = new ArgumentMatcher<MergeData>() {
- @Override
- public boolean matches(Object argument) {
- if((transactionVersion >= DataStoreVersions.LITHIUM_VERSION &&
- MergeData.SERIALIZABLE_CLASS.equals(argument.getClass())) ||
- (transactionVersion < DataStoreVersions.LITHIUM_VERSION &&
- ShardTransactionMessages.MergeData.class.equals(argument.getClass()))) {
-
- MergeData obj = MergeData.fromSerializable(argument);
- return obj.getPath().equals(TestModel.TEST_PATH) &&
- obj.getData().equals(nodeToWrite);
- }
-
- return false;
- }
- };
-
- return argThat(matcher);
- }
-
- private MergeData eqMergeData(final NormalizedNode<?, ?> nodeToWrite) {
- ArgumentMatcher<MergeData> matcher = new ArgumentMatcher<MergeData>() {
- @Override
- public boolean matches(Object argument) {
- if(argument instanceof MergeData) {
- MergeData obj = ((MergeData) argument);
- return obj.getPath().equals(TestModel.TEST_PATH) &&
- obj.getData().equals(nodeToWrite);
- }
-
- return false;
- }
- };
-
- return argThat(matcher);
- }
-
- private DeleteData eqSerializedDeleteData() {
- ArgumentMatcher<DeleteData> matcher = new ArgumentMatcher<DeleteData>() {
- @Override
- public boolean matches(Object argument) {
- return DeleteData.SERIALIZABLE_CLASS.equals(argument.getClass()) &&
- DeleteData.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH);
- }
- };
-
- return argThat(matcher);
- }
-
- private DeleteData eqDeleteData() {
- ArgumentMatcher<DeleteData> matcher = new ArgumentMatcher<DeleteData>() {
- @Override
- public boolean matches(Object argument) {
- return argument instanceof DeleteData &&
- ((DeleteData)argument).getPath().equals(TestModel.TEST_PATH);
- }
- };
-
- return argThat(matcher);
- }
-
- private Future<Object> readySerializedTxReply(String path) {
- return Futures.successful((Object)new ReadyTransactionReply(path).toSerializable());
- }
-
- private Future<Object> readyTxReply(String path) {
- return Futures.successful((Object)new ReadyTransactionReply(path));
- }
-
- private Future<Object> readSerializedDataReply(NormalizedNode<?, ?> data,
- short transactionVersion) {
- return Futures.successful(new ReadDataReply(data).toSerializable(transactionVersion));
- }
-
- private Future<Object> readSerializedDataReply(NormalizedNode<?, ?> data) {
- return readSerializedDataReply(data, DataStoreVersions.CURRENT_VERSION);
- }
-
- private Future<ReadDataReply> readDataReply(NormalizedNode<?, ?> data) {
- return Futures.successful(new ReadDataReply(data));
- }
-
- private Future<Object> dataExistsSerializedReply(boolean exists) {
- return Futures.successful(new DataExistsReply(exists).toSerializable());
- }
-
- private Future<DataExistsReply> dataExistsReply(boolean exists) {
- return Futures.successful(new DataExistsReply(exists));
- }
-
- private Future<Object> writeSerializedDataReply(short version) {
- return Futures.successful(new WriteDataReply().toSerializable(version));
- }
-
- private Future<Object> writeSerializedDataReply() {
- return writeSerializedDataReply(DataStoreVersions.CURRENT_VERSION);
- }
-
- private Future<WriteDataReply> writeDataReply() {
- return Futures.successful(new WriteDataReply());
- }
-
- private Future<Object> mergeSerializedDataReply(short version) {
- return Futures.successful(new MergeDataReply().toSerializable(version));
- }
-
- private Future<Object> mergeSerializedDataReply() {
- return mergeSerializedDataReply(DataStoreVersions.CURRENT_VERSION);
- }
-
- private Future<Object> incompleteFuture(){
- return mock(Future.class);
- }
-
- private Future<MergeDataReply> mergeDataReply() {
- return Futures.successful(new MergeDataReply());
- }
-
- private Future<Object> deleteSerializedDataReply() {
- return Futures.successful(new DeleteDataReply().toSerializable());
- }
-
- private Future<DeleteDataReply> deleteDataReply() {
- return Futures.successful(new DeleteDataReply());
- }
-
- private ActorSelection actorSelection(ActorRef actorRef) {
- return getSystem().actorSelection(actorRef.path());
- }
-
- private CreateTransactionReply createTransactionReply(ActorRef actorRef, int transactionVersion){
- return CreateTransactionReply.newBuilder()
- .setTransactionActorPath(actorRef.path().toString())
- .setTransactionId("txn-1")
- .setMessageVersion(transactionVersion)
- .build();
- }
-
- private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem,
- TransactionType type, int transactionVersion) {
- ActorRef actorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
- doReturn(actorSystem.actorSelection(actorRef.path())).
- when(mockActorContext).actorSelection(actorRef.path().toString());
-
- doReturn(Futures.successful(actorSystem.actorSelection(actorRef.path()))).
- when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
-
- doReturn(Futures.successful(createTransactionReply(actorRef, transactionVersion))).when(mockActorContext).
- executeOperationAsync(eq(actorSystem.actorSelection(actorRef.path())),
- eqCreateTransaction(memberName, type));
-
- doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString());
-
- doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit();
-
- return actorRef;
- }
-
- private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) {
- return setupActorContextWithInitialCreateTransaction(actorSystem, type, DataStoreVersions.CURRENT_VERSION);
- }
-
-
- private void propagateReadFailedExceptionCause(CheckedFuture<?, ReadFailedException> future)
- throws Throwable {
-
- try {
- future.checkedGet(5, TimeUnit.SECONDS);
- fail("Expected ReadFailedException");
- } catch(ReadFailedException e) {
- throw e.getCause();
- }
- }
-