@Override public void onCompletion(Exchange exchange) { if (wrappedAggregationStrategy != null && wrappedAggregationStrategy instanceof CompletionAwareAggregationStrategy) { ((CompletionAwareAggregationStrategy) wrappedAggregationStrategy).onCompletion(exchange); } // Remove exception, fault and redelivery info from exchange exchange.setException(null); exchange.removeProperty(Exchange.FAILURE_HANDLED); exchange.removeProperty(Exchange.FAILURE_ENDPOINT); exchange.removeProperty(Exchange.FAILURE_ROUTE_ID); exchange.removeProperty(Exchange.ERRORHANDLER_CIRCUIT_DETECTED); exchange.removeProperty(Exchange.ERRORHANDLER_HANDLED); exchange.removeProperty(Exchange.EXCEPTION_HANDLED); exchange.removeProperty(Exchange.EXCEPTION_CAUGHT); Message message = exchange.hasOut() ? exchange.getOut() : exchange.getIn(); message.setFault(false); message.removeHeader(Exchange.REDELIVERED); message.removeHeader(Exchange.REDELIVERY_COUNTER); message.removeHeader(Exchange.REDELIVERY_DELAY); message.removeHeader(Exchange.REDELIVERY_EXHAUSTED); message.removeHeader(Exchange.REDELIVERY_MAX_COUNTER); }
/** * Common work which must be done when we are done multicasting. * <p/> * This logic applies for both running synchronous and asynchronous as there are multiple exist points * when using the asynchronous routing engine. And therefore we want the logic in one method instead * of being scattered. * * @param original the original exchange * @param subExchange the current sub exchange, can be <tt>null</tt> for the synchronous part * @param pairs the pairs with the exchanges to process * @param callback the callback * @param doneSync the <tt>doneSync</tt> parameter to call on callback * @param forceExhaust whether or not error handling is exhausted */ protected void doDone(Exchange original, Exchange subExchange, final Iterable<ProcessorExchangePair> pairs, AsyncCallback callback, boolean doneSync, boolean forceExhaust) { // we are done so close the pairs iterator if (pairs != null && pairs instanceof Closeable) { IOHelper.close((Closeable) pairs, "pairs", LOG); } AggregationStrategy strategy = getAggregationStrategy(subExchange); if (strategy instanceof DelegateAggregationStrategy) { strategy = ((DelegateAggregationStrategy) strategy).getDelegate(); } // invoke the on completion callback if (strategy instanceof CompletionAwareAggregationStrategy) { ((CompletionAwareAggregationStrategy) strategy).onCompletion(subExchange); } // cleanup any per exchange aggregation strategy removeAggregationStrategyFromExchange(original); // we need to know if there was an exception, and if the stopOnException option was enabled // also we would need to know if any error handler has attempted redelivery and exhausted boolean stoppedOnException = false; boolean exception = false; boolean exhaust = forceExhaust || subExchange != null && (subExchange.getException() != null || ExchangeHelper.isRedeliveryExhausted(subExchange)); if (original.getException() != null || subExchange != null && subExchange.getException() != null) { // there was an exception and we stopped stoppedOnException = isStopOnException(); exception = true; } // must copy results at this point if (subExchange != null) { if (stoppedOnException) { // if we stopped due an exception then only propagate the exception original.setException(subExchange.getException()); } else { // copy the current result to original so it will contain this result of this eip ExchangeHelper.copyResults(original, subExchange); } } // .. and then if there was an exception we need to configure the redelivery exhaust // for example the noErrorHandler will not cause redelivery exhaust so if this error // handled has been in use, then the exhaust would be false (if not forced) if (exception) { // multicast uses error handling on its output processors and they have tried to redeliver // so we shall signal back to the other error handlers that we are exhausted and they should not // also try to redeliver as we will then do that twice original.setProperty(Exchange.REDELIVERY_EXHAUSTED, exhaust); } callback.done(doneSync); }
/** * Common work which must be done when we are done multicasting. * <p/> * This logic applies for both running synchronous and asynchronous as there are multiple exist points * when using the asynchronous routing engine. And therefore we want the logic in one method instead * of being scattered. * * @param original the original exchange * @param subExchange the current sub exchange, can be <tt>null</tt> for the synchronous part * @param pairs the pairs with the exchanges to process * @param callback the callback * @param doneSync the <tt>doneSync</tt> parameter to call on callback * @param forceExhaust whether or not error handling is exhausted */ protected void doDone(Exchange original, Exchange subExchange, final Iterable<ProcessorExchangePair> pairs, AsyncCallback callback, boolean doneSync, boolean forceExhaust) { // we are done so close the pairs iterator if (pairs != null && pairs instanceof Closeable) { IOHelper.close((Closeable) pairs, "pairs", LOG); } AggregationStrategy strategy = getAggregationStrategy(subExchange); // invoke the on completion callback if (strategy instanceof CompletionAwareAggregationStrategy) { ((CompletionAwareAggregationStrategy) strategy).onCompletion(subExchange); } // cleanup any per exchange aggregation strategy removeAggregationStrategyFromExchange(original); // we need to know if there was an exception, and if the stopOnException option was enabled // also we would need to know if any error handler has attempted redelivery and exhausted boolean stoppedOnException = false; boolean exception = false; boolean exhaust = forceExhaust || subExchange != null && (subExchange.getException() != null || ExchangeHelper.isRedeliveryExhausted(subExchange)); if (original.getException() != null || subExchange != null && subExchange.getException() != null) { // there was an exception and we stopped stoppedOnException = isStopOnException(); exception = true; } // must copy results at this point if (subExchange != null) { if (stoppedOnException) { // if we stopped due an exception then only propagte the exception original.setException(subExchange.getException()); } else { // copy the current result to original so it will contain this result of this eip ExchangeHelper.copyResults(original, subExchange); } } // .. and then if there was an exception we need to configure the redelivery exhaust // for example the noErrorHandler will not cause redelivery exhaust so if this error // handled has been in use, then the exhaust would be false (if not forced) if (exception) { // multicast uses error handling on its output processors and they have tried to redeliver // so we shall signal back to the other error handlers that we are exhausted and they should not // also try to redeliver as we will then do that twice original.setProperty(Exchange.REDELIVERY_EXHAUSTED, exhaust); } callback.done(doneSync); }
