- 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);