You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2016/08/03 16:14:25 UTC
camel git commit: CAMEL-10195: rest-dsl - automatic binding failure
with waitForTaskToComplete=Never
Repository: camel
Updated Branches:
refs/heads/master 0f5b58161 -> 8c18fecf7
CAMEL-10195: rest-dsl - automatic binding failure with waitForTaskToComplete=Never
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/8c18fecf
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/8c18fecf
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/8c18fecf
Branch: refs/heads/master
Commit: 8c18fecf7156dca6d2dbd87908914f465836d9eb
Parents: 0f5b581
Author: Claus Ibsen <da...@apache.org>
Authored: Wed Aug 3 17:18:30 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed Aug 3 17:18:30 2016 +0200
----------------------------------------------------------------------
.../camel/component/seda/SedaProducer.java | 20 +++++---------------
.../apache/camel/impl/DefaultUnitOfWork.java | 8 +++++++-
.../java/org/apache/camel/spi/UnitOfWork.java | 14 ++++++++++++++
.../org/apache/camel/util/ExchangeHelper.java | 19 ++++++++++++++++++-
4 files changed, 44 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/8c18fecf/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java b/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
index 3e34e8a..03ce9f3 100644
--- a/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
+++ b/camel-core/src/main/java/org/apache/camel/component/seda/SedaProducer.java
@@ -20,6 +20,7 @@ import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
@@ -27,6 +28,7 @@ import org.apache.camel.ExchangeTimedOutException;
import org.apache.camel.WaitForTaskToComplete;
import org.apache.camel.impl.DefaultAsyncProducer;
import org.apache.camel.spi.Synchronization;
+import org.apache.camel.spi.SynchronizationVetoable;
import org.apache.camel.support.SynchronizationAdapter;
import org.apache.camel.util.ExchangeHelper;
@@ -180,23 +182,11 @@ public class SedaProducer extends DefaultAsyncProducer {
protected Exchange prepareCopy(Exchange exchange, boolean handover) {
// use a new copy of the exchange to route async (and use same message id)
- Exchange copy = ExchangeHelper.createCorrelatedCopy(exchange, false, true);
+
// if handover we need to do special handover to avoid handing over
// RestBindingMarshalOnCompletion as it should not be handed over with SEDA
- if (handover) {
- List<Synchronization> completions = exchange.handoverCompletions();
- if (completions != null) {
- for (Synchronization sync : completions) {
- if (sync.getClass().getName().contains("RestBindingMarshalOnCompletion")) {
- // keep this one
- exchange.addOnCompletion(sync);
- } else {
- // handover
- copy.addOnCompletion(sync);
- }
- }
- }
- }
+ Exchange copy = ExchangeHelper.createCorrelatedCopy(exchange, handover, true,
+ synchronization -> !synchronization.getClass().getName().contains("RestBindingMarshalOnCompletion"));
// set a new from endpoint to be the seda queue
copy.setFromEndpoint(endpoint);
return copy;
http://git-wip-us.apache.org/repos/asf/camel/blob/8c18fecf/camel-core/src/main/java/org/apache/camel/impl/DefaultUnitOfWork.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultUnitOfWork.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultUnitOfWork.java
index 3fbd252..ae46435 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultUnitOfWork.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultUnitOfWork.java
@@ -23,6 +23,7 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.Stack;
+import java.util.function.Predicate;
import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
@@ -190,6 +191,11 @@ public class DefaultUnitOfWork implements UnitOfWork, Service {
}
public void handoverSynchronization(Exchange target) {
+ handoverSynchronization(target, null);
+ }
+
+ @Override
+ public void handoverSynchronization(Exchange target, Predicate<Synchronization> filter) {
if (synchronizations == null || synchronizations.isEmpty()) {
return;
}
@@ -204,7 +210,7 @@ public class DefaultUnitOfWork implements UnitOfWork, Service {
handover = veto.allowHandover();
}
- if (handover) {
+ if (handover && (filter == null || filter.test(synchronization))) {
log.trace("Handover synchronization {} to: {}", synchronization, target);
target.addOnCompletion(synchronization);
// remove it if its handed over
http://git-wip-us.apache.org/repos/asf/camel/blob/8c18fecf/camel-core/src/main/java/org/apache/camel/spi/UnitOfWork.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/spi/UnitOfWork.java b/camel-core/src/main/java/org/apache/camel/spi/UnitOfWork.java
index b165f43..99d1640 100644
--- a/camel-core/src/main/java/org/apache/camel/spi/UnitOfWork.java
+++ b/camel-core/src/main/java/org/apache/camel/spi/UnitOfWork.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.spi;
+import java.util.function.Predicate;
+
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
@@ -64,6 +66,18 @@ public interface UnitOfWork extends Service {
void handoverSynchronization(Exchange target);
/**
+ * Handover all the registered synchronizations to the target {@link org.apache.camel.Exchange}.
+ * <p/>
+ * This is used when a route turns into asynchronous and the {@link org.apache.camel.Exchange} that
+ * is continued and routed in the async thread should do the on completion callbacks instead of the
+ * original synchronous thread.
+ *
+ * @param target the target exchange
+ * @param filter optional filter to only handover if filter returns <tt>true</tt>
+ */
+ void handoverSynchronization(Exchange target, Predicate<Synchronization> filter);
+
+ /**
* Invoked when this unit of work has been completed, whether it has failed or completed
*
* @param exchange the current exchange
http://git-wip-us.apache.org/repos/asf/camel/blob/8c18fecf/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java b/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
index 030b78d..2c92cc5 100644
--- a/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
+++ b/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
@@ -25,6 +25,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelExchangeException;
@@ -44,6 +45,7 @@ import org.apache.camel.TypeConversionException;
import org.apache.camel.TypeConverter;
import org.apache.camel.impl.DefaultExchange;
import org.apache.camel.impl.MessageSupport;
+import org.apache.camel.spi.Synchronization;
import org.apache.camel.spi.UnitOfWork;
/**
@@ -229,6 +231,21 @@ public final class ExchangeHelper {
* @param useSameMessageId whether to use same message id on the copy message.
*/
public static Exchange createCorrelatedCopy(Exchange exchange, boolean handover, boolean useSameMessageId) {
+ return createCorrelatedCopy(exchange, handover, useSameMessageId, null);
+ }
+
+ /**
+ * Creates a new instance and copies from the current message exchange so that it can be
+ * forwarded to another destination as a new instance. Unlike regular copy this operation
+ * will not share the same {@link org.apache.camel.spi.UnitOfWork} so its should be used
+ * for async messaging, where the original and copied exchange are independent.
+ *
+ * @param exchange original copy of the exchange
+ * @param handover whether the on completion callbacks should be handed over to the new copy.
+ * @param useSameMessageId whether to use same message id on the copy message.
+ * @param filter whether to handover the on completion
+ */
+ public static Exchange createCorrelatedCopy(Exchange exchange, boolean handover, boolean useSameMessageId, Predicate<Synchronization> filter) {
String id = exchange.getExchangeId();
// make sure to do a safe copy as the correlated copy can be routed independently of the source.
@@ -246,7 +263,7 @@ public final class ExchangeHelper {
// hand over on completion to the copy if we got any
UnitOfWork uow = exchange.getUnitOfWork();
if (handover && uow != null) {
- uow.handoverSynchronization(copy);
+ uow.handoverSynchronization(copy, filter);
}
// set a correlation id so we can track back the original exchange
copy.setProperty(Exchange.CORRELATION_ID, id);