+ @Override
+ public ListenableFuture<Void> abort() {
+ // Note - we pass false for propagateException. In the front-end data broker, this method
+ // is called when one of the 3 phases fails with an exception. We'd rather have that
+ // original exception propagated to the client. If our abort fails and we propagate the
+ // exception then that exception will supersede and suppress the original exception. But
+ // it's the original exception that is the root cause and of more interest to the client.
+
+ return voidOperation("abort", new AbortTransaction(transactionId).toSerializable(),
+ AbortTransactionReply.SERIALIZABLE_CLASS, false);
+ }
+
+ @Override
+ public ListenableFuture<Void> commit() {
+ OperationCallback operationCallback = (cohortFutures.size() == 0) ? NO_OP_CALLBACK :
+ new TransactionRateLimitingCallback(actorContext);
+
+ return voidOperation("commit", new CommitTransaction(transactionId).toSerializable(),
+ CommitTransactionReply.SERIALIZABLE_CLASS, true, operationCallback);
+ }
+
+ private ListenableFuture<Void> voidOperation(final String operationName, final Object message,
+ final Class<?> expectedResponseClass, final boolean propagateException) {
+ return voidOperation(operationName, message, expectedResponseClass, propagateException, NO_OP_CALLBACK);
+ }
+
+ private ListenableFuture<Void> voidOperation(final String operationName, final Object message,
+ final Class<?> expectedResponseClass, final boolean propagateException, final OperationCallback callback) {
+
+ if(LOG.isDebugEnabled()) {
+ 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.
+
+ if(cohorts != null) {
+ finishVoidOperation(operationName, message, expectedResponseClass, propagateException,
+ returnFuture, callback);
+ } else {
+ buildCohortList().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,
+ operationName, failure);
+ }
+ if(propagateException) {
+ returnFuture.setException(failure);
+ } else {
+ returnFuture.set(null);
+ }
+ } else {
+ finishVoidOperation(operationName, message, expectedResponseClass,
+ propagateException, returnFuture, callback);
+ }
+ }
+ }, actorContext.getClientDispatcher());
+ }
+
+ return returnFuture;
+ }
+
+ private void finishVoidOperation(final String operationName, final Object message,
+ final Class<?> expectedResponseClass, final boolean propagateException,
+ final SettableFuture<Void> returnFuture, final OperationCallback callback) {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Tx {} finish {}", transactionId, operationName);
+ }
+
+ callback.run();
+
+ Future<Iterable<Object>> combinedFuture = invokeCohorts(message);
+
+ 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;
+ }
+ }
+ }
+
+ if(exceptionToPropagate != null) {
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Tx {}: a {} cohort Future failed: {}", transactionId,
+ operationName, exceptionToPropagate);
+ }
+ if(propagateException) {
+ // We don't log the exception here to avoid redundant logging since we're
+ // propagating to the caller in MD-SAL core who will log it.
+ returnFuture.setException(exceptionToPropagate);
+ } else {
+ // Since the caller doesn't want us to propagate the exception we'll also
+ // not log it normally. But it's usually not good to totally silence
+ // exceptions so we'll log it to debug level.
+ if(LOG.isDebugEnabled()) {
+ LOG.debug(String.format("%s failed", message.getClass().getSimpleName()),
+ exceptionToPropagate);
+ }
+ returnFuture.set(null);
+ }
+
+ callback.failure();
+ } else {
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Tx {}: {} succeeded", transactionId, operationName);
+ }
+ returnFuture.set(null);
+
+ callback.success();
+ }
+ }
+ }, actorContext.getClientDispatcher());