+ }
+
+
+ <T> Future<T> enqueueFutureOperation(final FutureOperation<T> op) {
+
+ Future<T> future;
+
+ if (transactionContext != null) {
+ future = op.invoke(transactionContext);
+ } else {
+ // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
+ // callback to be executed after the Tx is created.
+ final Promise<T> promise = akka.dispatch.Futures.promise();
+ addTxOperationOnComplete(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ promise.completeWith(op.invoke(transactionContext));
+ }
+ });
+
+ future = promise.future();
+ }
+
+ return future;
+ }
+
+ <T> CheckedFuture<T, ReadFailedException> enqueueReadOperation(final ReadOperation<T> op) {
+
+ CheckedFuture<T, ReadFailedException> future;
+
+ if (transactionContext != null) {
+ future = op.invoke(transactionContext);
+ } else {
+ // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
+ // callback to be executed after the Tx is created.
+ final SettableFuture<T> proxyFuture = SettableFuture.create();
+ addTxOperationOnComplete(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ Futures.addCallback(op.invoke(transactionContext), new FutureCallback<T>() {
+ @Override
+ public void onSuccess(T data) {
+ proxyFuture.set(data);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ proxyFuture.setException(t);
+ }
+ });
+ }
+ });
+
+ future = MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
+ }
+
+ return future;
+ }
+
+ void enqueueModifyOperation(final TransactionOperation op) {
+
+ if (transactionContext != null) {
+ op.invoke(transactionContext);
+ } else {
+ // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
+ // callback to be executed after the Tx is created.
+ addTxOperationOnComplete(op);
+ }
+ }
+
+ /**
+ * Performs a CreateTransaction try async.
+ */
+ private void tryCreateTransaction() {
+ Future<Object> createTxFuture = sendCreateTransaction(primaryShard,
+ new CreateTransaction(identifier.toString(),
+ TransactionProxy.this.transactionType.ordinal(),
+ getTransactionChainId()).toSerializable());
+
+ createTxFuture.onComplete(this, actorContext.getActorSystem().dispatcher());
+ }
+
+ @Override
+ public void onComplete(Throwable failure, Object response) {
+ if(failure instanceof NoShardLeaderException) {
+ // There's no leader for the shard yet - schedule and try again, unless we're out
+ // of retries. Note: createTxTries is volatile as it may be written by different
+ // threads however not concurrently, therefore decrementing it non-atomically here
+ // is ok.
+ if(--createTxTries > 0) {
+ LOG.debug("Tx {} Shard {} has no leader yet - scheduling create Tx retry",
+ identifier, shardName);
+
+ actorContext.getActorSystem().scheduler().scheduleOnce(CREATE_TX_TRY_INTERVAL,
+ new Runnable() {
+ @Override
+ public void run() {
+ tryCreateTransaction();
+ }
+ }, actorContext.getActorSystem().dispatcher());
+ return;
+ }
+ }
+
+ // Create the TransactionContext from the response or failure and execute delayed
+ // TransactionOperations. This entire section is done atomically (ie synchronized) with
+ // respect to #addTxOperationOnComplete to handle timing issues and ensure no
+ // TransactionOperation is missed and that they are processed in the order they occurred.
+ synchronized(txOperationsOnComplete) {
+ // Store the new TransactionContext locally until we've completed invoking the
+ // TransactionOperations. This avoids thread timing issues which could cause
+ // out-of-order TransactionOperations. Eg, on a modification operation, if the
+ // TransactionContext is non-null, then we directly call the TransactionContext.
+ // However, at the same time, the code may be executing the cached
+ // TransactionOperations. So to avoid thus timing, we don't publish the
+ // TransactionContext until after we've executed all cached TransactionOperations.
+ TransactionContext localTransactionContext;
+ if(failure != null) {
+ LOG.debug("Tx {} Creating NoOpTransaction because of error: {}", identifier,
+ failure.getMessage());
+
+ localTransactionContext = new NoOpTransactionContext(failure, identifier, operationLimiter);
+ } else if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
+ localTransactionContext = createValidTransactionContext(
+ CreateTransactionReply.fromSerializable(response));
+ } else {
+ IllegalArgumentException exception = new IllegalArgumentException(String.format(
+ "Invalid reply type %s for CreateTransaction", response.getClass()));
+
+ localTransactionContext = new NoOpTransactionContext(exception, identifier, operationLimiter);
+ }
+
+ for(TransactionOperation oper: txOperationsOnComplete) {
+ oper.invoke(localTransactionContext);
+ }
+
+ txOperationsOnComplete.clear();
+
+ // We're done invoking the TransactionOperations so we can now publish the
+ // TransactionContext.
+ transactionContext = localTransactionContext;
+ }
+ }
+
+ private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
+ String transactionPath = reply.getTransactionPath();
+
+ LOG.debug("Tx {} Received transaction actor path {}", identifier, transactionPath);
+
+ ActorSelection transactionActor = actorContext.actorSelection(transactionPath);
+
+ if (transactionType == TransactionType.READ_ONLY) {
+ // Add the actor to the remoteTransactionActors list for access by the
+ // cleanup PhantonReference.
+ remoteTransactionActors.add(transactionActor);
+
+ // Write to the memory barrier volatile to publish the above update to the
+ // remoteTransactionActors list for thread visibility.
+ remoteTransactionActorsMB.set(true);
+ }
+
+ // TxActor is always created where the leader of the shard is.
+ // Check if TxActor is created in the same node
+ boolean isTxActorLocal = actorContext.isPathLocal(transactionPath);
+
+ return new TransactionContextImpl(transactionPath, transactionActor, identifier,
+ actorContext, schemaContext, isTxActorLocal, reply.getVersion(), operationCompleter);