- private ListenableFuture<Void> voidOperation(final Object message, final Class expectedResponseClass){
- Callable<Void> call = new Callable<Void>() {
-
- @Override public Void call() throws Exception {
- for(ActorPath actorPath : cohortPaths){
- ActorSelection cohort = actorContext.actorSelection(actorPath);
-
- try {
- Object response =
- actorContext.executeRemoteOperation(cohort,
- message,
- ActorContext.ASK_DURATION);
-
- if (response != null && !response.getClass()
- .equals(expectedResponseClass)) {
- throw new RuntimeException(
- String.format(
- "did not get the expected response \n\t\t expected : %s \n\t\t actual : %s",
- expectedResponseClass.toString(),
- response.getClass().toString())
- );
+ private ListenableFuture<Void> voidOperation(final String operationName,
+ final MessageSupplier messageSupplier, final Class<?> expectedResponseClass,
+ final boolean propagateException, final OperationCallback callback) {
+ LOG.debug("Tx {} {}", transactionId, operationName);
+
+ final SettableFuture<Void> returnFuture = SettableFuture.create();
+
+ // The cohort actor list should already be built at this point by the canCommit phase but,
+ // if not for some reason, we'll try to build it here.
+
+ ListenableFuture<Void> future = resolveCohorts();
+ if(successfulFuture(future)) {
+ finishVoidOperation(operationName, messageSupplier, expectedResponseClass, propagateException,
+ returnFuture, callback);
+ } else {
+ Futures.addCallback(future, new FutureCallback<Void>() {
+ @Override
+ public void onSuccess(Void notUsed) {
+ finishVoidOperation(operationName, messageSupplier, expectedResponseClass,
+ propagateException, returnFuture, callback);
+ }
+
+ @Override
+ public void onFailure(Throwable failure) {
+ LOG.debug("Tx {}: a {} cohort path Future failed: {}", transactionId, operationName, failure);
+
+ if(propagateException) {
+ returnFuture.setException(failure);
+ } else {
+ returnFuture.set(null);
+ }
+ }
+ });
+ }
+
+ return returnFuture;
+ }
+
+ private void finishVoidOperation(final String operationName, MessageSupplier messageSupplier,
+ final Class<?> expectedResponseClass, final boolean propagateException,
+ final SettableFuture<Void> returnFuture, final OperationCallback callback) {
+ LOG.debug("Tx {} finish {}", transactionId, operationName);
+
+ callback.resume();
+
+ Future<Iterable<Object>> combinedFuture = invokeCohorts(messageSupplier);
+
+ combinedFuture.onComplete(new OnComplete<Iterable<Object>>() {
+ @Override
+ public void onComplete(Throwable failure, Iterable<Object> responses) throws Throwable {
+ Throwable exceptionToPropagate = failure;
+ if(exceptionToPropagate == null) {
+ for(Object response: responses) {
+ if(!response.getClass().equals(expectedResponseClass)) {
+ exceptionToPropagate = new IllegalArgumentException(
+ String.format("Unexpected response type %s", response.getClass()));
+ break;