- public Boolean call() throws Exception {
- for(ActorPath actorPath : cohortPaths){
- ActorSelection cohort = actorContext.actorSelection(actorPath);
-
- try {
- Object response =
- actorContext.executeRemoteOperation(cohort,
- new CanCommitTransaction().toSerializable(),
- ActorContext.ASK_DURATION);
-
- if (response.getClass().equals(CanCommitTransactionReply.SERIALIZABLE_CLASS)) {
- CanCommitTransactionReply reply =
- CanCommitTransactionReply.fromSerializable(response);
- if (!reply.getCanCommit()) {
- System.out.println("**TOM - failed: false");
- return false;
- }
+ public Void apply(Iterable<ActorPath> paths) {
+ cohortPaths = Lists.newArrayList(paths);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Tx {} successfully built cohort path list: {}",
+ transactionId, cohortPaths);
+ }
+ return null;
+ }
+ }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getActorSystem().dispatcher());
+ }
+
+ @Override
+ public ListenableFuture<Boolean> canCommit() {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Tx {} canCommit", transactionId);
+ }
+ final SettableFuture<Boolean> returnFuture = SettableFuture.create();
+
+ // The first phase of canCommit is to gather the list of cohort actor paths that will
+ // participate in the commit. buildCohortPathsList combines the cohort path Futures into
+ // one Future which we wait on asynchronously here. The cohort actor paths are
+ // extracted from ReadyTransactionReply messages by the Futures that were obtained earlier
+ // and passed to us from upstream processing. If any one fails then we'll fail canCommit.
+
+ buildCohortPathsList().onComplete(new OnComplete<Void>() {
+ @Override
+ public void onComplete(Throwable failure, Void notUsed) throws Throwable {
+ if(failure != null) {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Tx {}: a cohort path Future failed: {}", transactionId, failure);
+ }
+ returnFuture.setException(failure);
+ } else {
+ finishCanCommit(returnFuture);
+ }
+ }
+ }, actorContext.getActorSystem().dispatcher());
+
+ return returnFuture;
+ }
+
+ private void finishCanCommit(final SettableFuture<Boolean> returnFuture) {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Tx {} finishCanCommit", transactionId);
+ }
+ // The last phase of canCommit is to invoke all the cohort actors asynchronously to perform
+ // their canCommit processing. If any one fails then we'll fail canCommit.
+
+ Future<Iterable<Object>> combinedFuture =
+ invokeCohorts(new CanCommitTransaction().toSerializable());
+
+ combinedFuture.onComplete(new OnComplete<Iterable<Object>>() {
+ @Override
+ public void onComplete(Throwable failure, Iterable<Object> responses) throws Throwable {
+ if(failure != null) {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Tx {}: a canCommit cohort Future failed: {}", transactionId, failure);
+ }
+ returnFuture.setException(failure);
+ return;
+ }
+
+ boolean result = true;
+ for(Object response: responses) {
+ if (response.getClass().equals(CanCommitTransactionReply.SERIALIZABLE_CLASS)) {
+ CanCommitTransactionReply reply =
+ CanCommitTransactionReply.fromSerializable(response);
+ if (!reply.getCanCommit()) {
+ result = false;
+ break;