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 2020/01/16 18:40:40 UTC
[camel] 02/04: CAMEL-14409: camel-core - ExtendedExchange for
advanced API
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 207e7dfdd4d3890ccf4f394853f4f6a2a64ef06d
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Thu Jan 16 14:59:38 2020 +0100
CAMEL-14409: camel-core - ExtendedExchange for advanced API
---
.../src/main/java/org/apache/camel/Exchange.java | 60 ----------------------
.../java/org/apache/camel/ExtendedExchange.java | 58 +++++++++++++++++++++
.../camel/impl/engine/DefaultConsumerTemplate.java | 3 +-
.../camel/impl/engine/DefaultUnitOfWork.java | 3 +-
.../camel/processor/CamelInternalProcessor.java | 6 +--
.../java/org/apache/camel/processor/Enricher.java | 3 +-
.../org/apache/camel/processor/PollEnricher.java | 3 +-
.../java/org/apache/camel/processor/Splitter.java | 3 +-
.../processor/aggregate/AggregateProcessor.java | 3 +-
.../processor/idempotent/IdempotentConsumer.java | 3 +-
.../apache/camel/builder/xml/XsltBuilderTest.java | 3 +-
.../seda/SedaDiscardIfNoConsumerTest.java | 3 +-
.../seda/SedaInOutChainedWithOnCompletionTest.java | 3 +-
.../SedaWaitForTaskCompleteOnCompletionTest.java | 3 +-
.../seda/SedaWaitForTaskNewerOnCompletionTest.java | 3 +-
.../converter/stream/CachedOutputStreamTest.java | 3 +-
.../impl/engine/CamelPostProcessorHelperTest.java | 5 +-
.../org/apache/camel/issues/GertJBIIssueTest.java | 3 +-
.../apache/camel/language/simple/SimpleTest.java | 3 +-
.../processor/MDCOnCompletionOnCompletionTest.java | 3 +-
.../camel/processor/OnCompletionContainsTest.java | 15 +++---
.../processor/OnCompletionShouldBeLastTest.java | 7 +--
.../processor/RouteAwareSynchronizationTest.java | 3 +-
.../processor/async/AsyncEndpointDelayUoWTest.java | 3 +-
.../async/AsyncEndpointUoWFailedTest.java | 3 +-
.../processor/async/AsyncEndpointUoWTest.java | 3 +-
.../converter/stream/FileInputStreamCache.java | 3 +-
.../org/apache/camel/support/DefaultConsumer.java | 5 +-
.../org/apache/camel/support/DefaultExchange.java | 3 +-
.../org/apache/camel/support/ExchangeHelper.java | 5 +-
.../org/apache/camel/support/UnitOfWorkHelper.java | 3 +-
31 files changed, 127 insertions(+), 103 deletions(-)
diff --git a/core/camel-api/src/main/java/org/apache/camel/Exchange.java b/core/camel-api/src/main/java/org/apache/camel/Exchange.java
index 6a8fa26..49ab34e 100644
--- a/core/camel-api/src/main/java/org/apache/camel/Exchange.java
+++ b/core/camel-api/src/main/java/org/apache/camel/Exchange.java
@@ -554,43 +554,18 @@ public interface Exchange {
Endpoint getFromEndpoint();
/**
- * Sets the endpoint which originated this message exchange. This method
- * should typically only be called by {@link org.apache.camel.Endpoint} implementations
- *
- * @param fromEndpoint the endpoint which is originating this message exchange
- */
- // TODO: Move to ExtendedExchange
- void setFromEndpoint(Endpoint fromEndpoint);
-
- /**
* Returns the route id which originated this message exchange if a route consumer on an endpoint
* created the message exchange, otherwise this property will be <tt>null</tt>
*/
String getFromRouteId();
/**
- * Sets the route id which originated this message exchange. This method
- * should typically only be called by the internal framework.
- *
- * @param fromRouteId the from route id
- */
- // TODO: Move to ExtendedExchange
- void setFromRouteId(String fromRouteId);
-
- /**
* Returns the unit of work that this exchange belongs to; which may map to
* zero, one or more physical transactions
*/
UnitOfWork getUnitOfWork();
/**
- * Sets the unit of work that this exchange belongs to; which may map to
- * zero, one or more physical transactions
- */
- // TODO: Move to ExtendedExchange
- void setUnitOfWork(UnitOfWork unitOfWork);
-
- /**
* Returns the exchange id (unique)
*/
String getExchangeId();
@@ -601,41 +576,6 @@ public interface Exchange {
void setExchangeId(String id);
/**
- * Adds a {@link org.apache.camel.spi.Synchronization} to be invoked as callback when
- * this exchange is completed.
- *
- * @param onCompletion the callback to invoke on completion of this exchange
- */
- // TODO: Move to ExtendedExchange
- void addOnCompletion(Synchronization onCompletion);
-
- /**
- * Checks if the passed {@link org.apache.camel.spi.Synchronization} instance is
- * already contained on this exchange.
- *
- * @param onCompletion the callback instance that is being checked for
- * @return <tt>true</tt>, if callback instance is already contained on this exchange, else <tt>false</tt>
- */
- // TODO: Move to ExtendedExchange
- boolean containsOnCompletion(Synchronization onCompletion);
-
- /**
- * Handover all the on completions from this exchange to the target exchange.
- *
- * @param target the target exchange
- */
- // TODO: Move to ExtendedExchange
- void handoverCompletions(Exchange target);
-
- /**
- * Handover all the on completions from this exchange
- *
- * @return the on completions
- */
- // TODO: Move to ExtendedExchange
- List<Synchronization> handoverCompletions();
-
- /**
* Gets the timestamp in millis when this exchange was created.
*/
long getCreated();
diff --git a/core/camel-api/src/main/java/org/apache/camel/ExtendedExchange.java b/core/camel-api/src/main/java/org/apache/camel/ExtendedExchange.java
index 718bdb0..2b07540 100644
--- a/core/camel-api/src/main/java/org/apache/camel/ExtendedExchange.java
+++ b/core/camel-api/src/main/java/org/apache/camel/ExtendedExchange.java
@@ -1,5 +1,10 @@
package org.apache.camel;
+import java.util.List;
+
+import org.apache.camel.spi.Synchronization;
+import org.apache.camel.spi.UnitOfWork;
+
/**
* Extended {@link Exchange} which contains the methods and APIs that are not primary intended for Camel end users
* but for SPI, custom components, or more advanced used-cases with Camel.
@@ -7,6 +12,59 @@ package org.apache.camel;
public interface ExtendedExchange extends Exchange {
/**
+ * Sets the endpoint which originated this message exchange. This method
+ * should typically only be called by {@link Endpoint} implementations
+ *
+ * @param fromEndpoint the endpoint which is originating this message exchange
+ */
+ void setFromEndpoint(Endpoint fromEndpoint);
+
+ /**
+ * Sets the route id which originated this message exchange. This method
+ * should typically only be called by the internal framework.
+ *
+ * @param fromRouteId the from route id
+ */
+ void setFromRouteId(String fromRouteId);
+
+ /**
+ * Sets the unit of work that this exchange belongs to; which may map to
+ * zero, one or more physical transactions
+ */
+ void setUnitOfWork(UnitOfWork unitOfWork);
+
+ /**
+ * Adds a {@link org.apache.camel.spi.Synchronization} to be invoked as callback when
+ * this exchange is completed.
+ *
+ * @param onCompletion the callback to invoke on completion of this exchange
+ */
+ void addOnCompletion(Synchronization onCompletion);
+
+ /**
+ * Checks if the passed {@link Synchronization} instance is
+ * already contained on this exchange.
+ *
+ * @param onCompletion the callback instance that is being checked for
+ * @return <tt>true</tt>, if callback instance is already contained on this exchange, else <tt>false</tt>
+ */
+ boolean containsOnCompletion(Synchronization onCompletion);
+
+ /**
+ * Handover all the on completions from this exchange to the target exchange.
+ *
+ * @param target the target exchange
+ */
+ void handoverCompletions(Exchange target);
+
+ /**
+ * Handover all the on completions from this exchange
+ *
+ * @return the on completions
+ */
+ List<Synchronization> handoverCompletions();
+
+ /**
* Sets the history node id (the current processor that will process the exchange)
*/
void setHistoryNodeId(String historyNodeId);
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultConsumerTemplate.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultConsumerTemplate.java
index 3acec66..b5d6652 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultConsumerTemplate.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultConsumerTemplate.java
@@ -22,6 +22,7 @@ import org.apache.camel.CamelContext;
import org.apache.camel.ConsumerTemplate;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.spi.ConsumerCache;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.support.CamelContextHelper;
@@ -224,7 +225,7 @@ public class DefaultConsumerTemplate extends ServiceSupport implements ConsumerT
}
if (exchange.getUnitOfWork() == null) {
// handover completions and done them manually to ensure they are being executed
- List<Synchronization> synchronizations = exchange.handoverCompletions();
+ List<Synchronization> synchronizations = exchange.adapt(ExtendedExchange.class).handoverCompletions();
UnitOfWorkHelper.doneSynchronizations(exchange, synchronizations, log);
} else {
// done the unit of work
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java
index 4f08567..a96df0b 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java
@@ -28,6 +28,7 @@ import java.util.function.Predicate;
import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.Route;
@@ -185,7 +186,7 @@ public class DefaultUnitOfWork implements UnitOfWork, Service {
if (handover && (filter == null || filter.test(synchronization))) {
log.trace("Handover synchronization {} to: {}", synchronization, target);
- target.addOnCompletion(synchronization);
+ target.adapt(ExtendedExchange.class).addOnCompletion(synchronization);
// remove it if its handed over
it.remove();
} else {
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java b/core/camel-base/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
index 5cfd5e0..9c9164f 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
@@ -594,7 +594,7 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor {
if (routeId == null) {
this.routeId = routeContext.getRouteId();
}
- exchange.setFromRouteId(routeId);
+ exchange.adapt(ExtendedExchange.class).setFromRouteId(routeId);
}
// only return UnitOfWork if we created a new as then its us that handle the lifecycle to done the created UoW
@@ -604,7 +604,7 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor {
// If there is no existing UoW, then we should start one and
// terminate it once processing is completed for the exchange.
created = createUnitOfWork(exchange);
- exchange.setUnitOfWork(created);
+ exchange.adapt(ExtendedExchange.class).setUnitOfWork(created);
created.start();
}
@@ -856,7 +856,7 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor {
if (!contains) {
added = true;
tracer.traceBeforeRoute(routeDefinition, exchange);
- exchange.addOnCompletion(tracingAfterRoute);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(tracingAfterRoute);
}
}
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/Enricher.java b/core/camel-base/src/main/java/org/apache/camel/processor/Enricher.java
index efd8b67..411b36a 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/Enricher.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/Enricher.java
@@ -27,6 +27,7 @@ import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Expression;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.impl.engine.DefaultProducerCache;
import org.apache.camel.spi.EndpointUtilizationStatistics;
import org.apache.camel.spi.IdAware;
@@ -326,7 +327,7 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA
if (isShareUnitOfWork()) {
target.setProperty(Exchange.PARENT_UNIT_OF_WORK, source.getUnitOfWork());
// and then share the unit of work
- target.setUnitOfWork(source.getUnitOfWork());
+ target.adapt(ExtendedExchange.class).setUnitOfWork(source.getUnitOfWork());
}
return target;
}
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/PollEnricher.java b/core/camel-base/src/main/java/org/apache/camel/processor/PollEnricher.java
index c441146..828e3ef 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/PollEnricher.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/PollEnricher.java
@@ -25,6 +25,7 @@ import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Expression;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.PollingConsumer;
import org.apache.camel.impl.engine.DefaultConsumerCache;
import org.apache.camel.spi.ConsumerCache;
@@ -290,7 +291,7 @@ public class PollEnricher extends AsyncProcessorSupport implements IdAware, Rout
copyResultsPreservePattern(exchange, aggregatedExchange);
// handover any synchronization
if (resourceExchange != null) {
- resourceExchange.handoverCompletions(exchange);
+ resourceExchange.adapt(ExtendedExchange.class).handoverCompletions(exchange);
}
}
}
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/Splitter.java b/core/camel-base/src/main/java/org/apache/camel/processor/Splitter.java
index 7250144..0f09d12 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/Splitter.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/Splitter.java
@@ -31,6 +31,7 @@ import org.apache.camel.AsyncProcessor;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Expression;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.RuntimeCamelException;
@@ -239,7 +240,7 @@ public class Splitter extends MulticastProcessor implements AsyncProcessor, Trac
@Override
protected void updateNewExchange(Exchange exchange, int index, Iterable<ProcessorExchangePair> allPairs, boolean hasNext) {
// do not share unit of work
- exchange.setUnitOfWork(null);
+ exchange.adapt(ExtendedExchange.class).setUnitOfWork(null);
exchange.setProperty(Exchange.SPLIT_INDEX, index);
if (allPairs instanceof Collection) {
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java b/core/camel-base/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
index f894765..4924359 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
@@ -42,6 +42,7 @@ import org.apache.camel.CamelExchangeException;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Expression;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Navigate;
import org.apache.camel.NoSuchEndpointException;
import org.apache.camel.Predicate;
@@ -823,7 +824,7 @@ public class AggregateProcessor extends AsyncProcessorSupport implements Navigat
log.debug("Processing aggregated exchange: {}", exchange);
// add on completion task so we remember to update the inProgressCompleteExchanges
- exchange.addOnCompletion(new AggregateOnCompletion(exchange.getExchangeId()));
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new AggregateOnCompletion(exchange.getExchangeId()));
// send this exchange
// the call to schedule last if needed to ensure in-order processing of the aggregates
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java b/core/camel-base/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java
index a9de078..b126382 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java
@@ -26,6 +26,7 @@ import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
import org.apache.camel.Exchange;
import org.apache.camel.Expression;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Navigate;
import org.apache.camel.Processor;
import org.apache.camel.spi.IdAware;
@@ -149,7 +150,7 @@ public class IdempotentConsumer extends AsyncProcessorSupport implements CamelCo
target = new IdempotentConsumerCallback(exchange, onCompletion, callback, completionEager);
if (!completionEager) {
// the scope is to do the idempotent completion work as an unit of work on the exchange when its done being routed
- exchange.addOnCompletion(onCompletion);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(onCompletion);
}
} catch (Exception e) {
exchange.setException(e);
diff --git a/core/camel-core/src/test/java/org/apache/camel/builder/xml/XsltBuilderTest.java b/core/camel-core/src/test/java/org/apache/camel/builder/xml/XsltBuilderTest.java
index ba16332..1f56420 100644
--- a/core/camel-core/src/test/java/org/apache/camel/builder/xml/XsltBuilderTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/builder/xml/XsltBuilderTest.java
@@ -26,6 +26,7 @@ import javax.xml.transform.Source;
import javax.xml.transform.Templates;
import javax.xml.transform.sax.SAXSource;
+import org.apache.camel.ExtendedExchange;
import org.w3c.dom.Document;
import org.xml.sax.InputSource;
@@ -257,7 +258,7 @@ public class XsltBuilderTest extends ContextTestSupport {
assertTrue(body.endsWith("<goodbye>world!</goodbye>"));
// now done the exchange
- List<Synchronization> onCompletions = exchange.handoverCompletions();
+ List<Synchronization> onCompletions = exchange.adapt(ExtendedExchange.class).handoverCompletions();
UnitOfWorkHelper.doneSynchronizations(exchange, onCompletions, log);
// the file should be deleted
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaDiscardIfNoConsumerTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaDiscardIfNoConsumerTest.java
index 95a39cd..893ca0d 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaDiscardIfNoConsumerTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaDiscardIfNoConsumerTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.seda;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.support.SynchronizationAdapter;
@@ -52,7 +53,7 @@ public class SedaDiscardIfNoConsumerTest extends ContextTestSupport {
@Override
public void process(Exchange exchange) throws Exception {
exchange.getIn().setBody("Hello World");
- exchange.addOnCompletion(myCompletion);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(myCompletion);
}
});
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaInOutChainedWithOnCompletionTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaInOutChainedWithOnCompletionTest.java
index c2bc872..2ac0802 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaInOutChainedWithOnCompletionTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaInOutChainedWithOnCompletionTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.seda;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.support.SynchronizationAdapter;
@@ -47,7 +48,7 @@ public class SedaInOutChainedWithOnCompletionTest extends ContextTestSupport {
from("seda:a").process(new Processor() {
public void process(Exchange exchange) throws Exception {
// should come in last
- exchange.addOnCompletion(new SynchronizationAdapter() {
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new SynchronizationAdapter() {
@Override
public void onDone(Exchange exchange) {
template.sendBody("mock:c", "onCustomCompletion");
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaWaitForTaskCompleteOnCompletionTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaWaitForTaskCompleteOnCompletionTest.java
index 42e55b6..c987a0d 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaWaitForTaskCompleteOnCompletionTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaWaitForTaskCompleteOnCompletionTest.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.seda;
import org.apache.camel.CamelExecutionException;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.support.SynchronizationAdapter;
@@ -56,7 +57,7 @@ public class SedaWaitForTaskCompleteOnCompletionTest extends ContextTestSupport
from("direct:start").process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
- exchange.addOnCompletion(new SynchronizationAdapter() {
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new SynchronizationAdapter() {
@Override
public void onDone(Exchange exchange) {
done = done + "A";
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaWaitForTaskNewerOnCompletionTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaWaitForTaskNewerOnCompletionTest.java
index 412fb90..535d7b2 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaWaitForTaskNewerOnCompletionTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaWaitForTaskNewerOnCompletionTest.java
@@ -21,6 +21,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.support.SynchronizationAdapter;
@@ -55,7 +56,7 @@ public class SedaWaitForTaskNewerOnCompletionTest extends ContextTestSupport {
from("direct:start").process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
- exchange.addOnCompletion(new SynchronizationAdapter() {
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new SynchronizationAdapter() {
@Override
public void onDone(Exchange exchange) {
done = done + "A";
diff --git a/core/camel-core/src/test/java/org/apache/camel/converter/stream/CachedOutputStreamTest.java b/core/camel-core/src/test/java/org/apache/camel/converter/stream/CachedOutputStreamTest.java
index 6ff67e0..6713c77 100644
--- a/core/camel-core/src/test/java/org/apache/camel/converter/stream/CachedOutputStreamTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/converter/stream/CachedOutputStreamTest.java
@@ -26,6 +26,7 @@ import java.io.InputStreamReader;
import org.apache.camel.CamelContext;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.StreamCache;
import org.apache.camel.converter.IOConverter;
import org.apache.camel.impl.engine.DefaultUnitOfWork;
@@ -60,7 +61,7 @@ public class CachedOutputStreamTest extends ContextTestSupport {
exchange = new DefaultExchange(context);
UnitOfWork uow = new DefaultUnitOfWork(exchange);
- exchange.setUnitOfWork(uow);
+ exchange.adapt(ExtendedExchange.class).setUnitOfWork(uow);
}
@Override
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/engine/CamelPostProcessorHelperTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/engine/CamelPostProcessorHelperTest.java
index cdbf5c0..ebe5477 100644
--- a/core/camel-core/src/test/java/org/apache/camel/impl/engine/CamelPostProcessorHelperTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/engine/CamelPostProcessorHelperTest.java
@@ -27,6 +27,7 @@ import org.apache.camel.Consume;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.FluentProducerTemplate;
import org.apache.camel.NoSuchBeanException;
import org.apache.camel.PollingConsumer;
@@ -506,7 +507,7 @@ public class CamelPostProcessorHelperTest extends ContextTestSupport {
@Consume("seda:foo")
public void consumeSomething(String body, Exchange exchange) {
- exchange.addOnCompletion(mySynchronization);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(mySynchronization);
assertEquals("Hello World", body);
template.sendBody("mock:result", body);
}
@@ -521,7 +522,7 @@ public class CamelPostProcessorHelperTest extends ContextTestSupport {
assertEquals("Hello World", body);
Exchange exchange = producer.getEndpoint().createExchange();
- exchange.addOnCompletion(mySynchronization);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(mySynchronization);
exchange.getIn().setBody(body);
producer.process(exchange);
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/issues/GertJBIIssueTest.java b/core/camel-core/src/test/java/org/apache/camel/issues/GertJBIIssueTest.java
index 697950b..01a7309 100644
--- a/core/camel-core/src/test/java/org/apache/camel/issues/GertJBIIssueTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/issues/GertJBIIssueTest.java
@@ -21,6 +21,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.support.SynchronizationAdapter;
@@ -74,7 +75,7 @@ public class GertJBIIssueTest extends ContextTestSupport {
template.send("direct:start", new Processor() {
public void process(Exchange exchange) throws Exception {
- exchange.addOnCompletion(new SynchronizationAdapter() {
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new SynchronizationAdapter() {
@Override
public void onDone(Exchange exchange) {
cause = exchange.getException();
diff --git a/core/camel-core/src/test/java/org/apache/camel/language/simple/SimpleTest.java b/core/camel-core/src/test/java/org/apache/camel/language/simple/SimpleTest.java
index 311f1a8..a2a15e6 100644
--- a/core/camel-core/src/test/java/org/apache/camel/language/simple/SimpleTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/language/simple/SimpleTest.java
@@ -32,6 +32,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Expression;
import org.apache.camel.ExpressionIllegalSyntaxException;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.InvalidPayloadException;
import org.apache.camel.LanguageTestSupport;
import org.apache.camel.Predicate;
@@ -217,7 +218,7 @@ public class SimpleTest extends LanguageTestSupport {
assertExpression("${header.foo}", "abc");
assertExpression("${headers.foo}", "abc");
assertExpression("${routeId}", exchange.getFromRouteId());
- exchange.setFromRouteId("myRouteId");
+ exchange.adapt(ExtendedExchange.class).setFromRouteId("myRouteId");
assertExpression("${routeId}", "myRouteId");
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/MDCOnCompletionOnCompletionTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/MDCOnCompletionOnCompletionTest.java
index e2ef8d5..05a9cd1 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/MDCOnCompletionOnCompletionTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/MDCOnCompletionOnCompletionTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.processor;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
@@ -50,7 +51,7 @@ public class MDCOnCompletionOnCompletionTest extends ContextTestSupport {
from("timer:foo?period=5000").routeId("route-a").setBody().constant("Hello World").onCompletion().process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
- exchange.addOnCompletion(new MyOnCompletion());
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new MyOnCompletion());
}
}).end().to("log:foo").to("direct:b");
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionContainsTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionContainsTest.java
index a084bf3..0a45e35 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionContainsTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionContainsTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.processor;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.support.SynchronizationAdapter;
@@ -65,26 +66,26 @@ public class OnCompletionContainsTest extends ContextTestSupport {
from("direct:start").process(new Processor() {
public void process(Exchange exchange) throws Exception {
SynchronizationAdapter adapter = new SimpleSynchronizationAdapter("mock:sync", "A");
- exchange.addOnCompletion(adapter);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(adapter);
// should not add the adapter again as we already have
// it
- if (!exchange.containsOnCompletion(adapter)) {
- exchange.addOnCompletion(adapter);
+ if (!exchange.adapt(ExtendedExchange.class).containsOnCompletion(adapter)) {
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(adapter);
}
adapter = new SimpleSynchronizationAdapter("mock:sync", "B");
- exchange.addOnCompletion(adapter);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(adapter);
// now add the B again as we want to test that this also
// work
- if (exchange.containsOnCompletion(adapter)) {
- exchange.addOnCompletion(adapter);
+ if (exchange.adapt(ExtendedExchange.class).containsOnCompletion(adapter)) {
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(adapter);
}
// add a C that is no a SimpleSynchronizationAdapter
// class
- exchange.addOnCompletion(new SynchronizationAdapter() {
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new SynchronizationAdapter() {
@Override
public void onDone(Exchange exchange) {
template.sendBody("mock:sync", "C");
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionShouldBeLastTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionShouldBeLastTest.java
index d360527..1ea974b 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionShouldBeLastTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionShouldBeLastTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.processor;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.support.SynchronizationAdapter;
@@ -44,7 +45,7 @@ public class OnCompletionShouldBeLastTest extends ContextTestSupport {
from("direct:start").process(new Processor() {
public void process(Exchange exchange) throws Exception {
- exchange.addOnCompletion(new SynchronizationAdapter() {
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new SynchronizationAdapter() {
@Override
public void onDone(Exchange exchange) {
template.sendBody("mock:sync", "A");
@@ -56,7 +57,7 @@ public class OnCompletionShouldBeLastTest extends ContextTestSupport {
}
});
- exchange.addOnCompletion(new SynchronizationAdapter() {
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new SynchronizationAdapter() {
@Override
public void onDone(Exchange exchange) {
template.sendBody("mock:sync", "B");
@@ -68,7 +69,7 @@ public class OnCompletionShouldBeLastTest extends ContextTestSupport {
}
});
- exchange.addOnCompletion(new SynchronizationAdapter() {
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new SynchronizationAdapter() {
@Override
public void onDone(Exchange exchange) {
template.sendBody("mock:sync", "C");
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/RouteAwareSynchronizationTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/RouteAwareSynchronizationTest.java
index 5f1c81c..bb0338e 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/RouteAwareSynchronizationTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/RouteAwareSynchronizationTest.java
@@ -21,6 +21,7 @@ import java.util.List;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.Route;
import org.apache.camel.builder.RouteBuilder;
@@ -43,7 +44,7 @@ public class RouteAwareSynchronizationTest extends ContextTestSupport {
template.send("direct:start", new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
- exchange.addOnCompletion(new MyRouteAware());
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new MyRouteAware());
exchange.getIn().setBody("Hello World");
}
});
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointDelayUoWTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointDelayUoWTest.java
index 1f5c6f2..b0f82df 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointDelayUoWTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointDelayUoWTest.java
@@ -20,6 +20,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.support.SynchronizationAdapter;
@@ -58,7 +59,7 @@ public class AsyncEndpointDelayUoWTest extends ContextTestSupport {
from("direct:start").process(new Processor() {
public void process(Exchange exchange) throws Exception {
beforeThreadName = Thread.currentThread().getName();
- exchange.addOnCompletion(sync);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(sync);
}
}).to("mock:before").to("log:before").delay(500).asyncDelayed().process(new Processor() {
public void process(Exchange exchange) throws Exception {
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointUoWFailedTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointUoWFailedTest.java
index 957ef70..ce263b0 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointUoWFailedTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointUoWFailedTest.java
@@ -21,6 +21,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.CamelExecutionException;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.support.SynchronizationAdapter;
@@ -66,7 +67,7 @@ public class AsyncEndpointUoWFailedTest extends ContextTestSupport {
from("direct:start").process(new Processor() {
public void process(Exchange exchange) throws Exception {
beforeThreadName = Thread.currentThread().getName();
- exchange.addOnCompletion(sync);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(sync);
}
}).to("mock:before").to("log:before").to("async:bye:camel").process(new Processor() {
public void process(Exchange exchange) throws Exception {
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointUoWTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointUoWTest.java
index 8ec5f58..45a68d0 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointUoWTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointUoWTest.java
@@ -20,6 +20,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.support.SynchronizationAdapter;
@@ -60,7 +61,7 @@ public class AsyncEndpointUoWTest extends ContextTestSupport {
from("direct:start").process(new Processor() {
public void process(Exchange exchange) throws Exception {
beforeThreadName = Thread.currentThread().getName();
- exchange.addOnCompletion(sync);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(sync);
}
}).to("mock:before").to("log:before").to("async:bye:camel").process(new Processor() {
public void process(Exchange exchange) throws Exception {
diff --git a/core/camel-support/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java b/core/camel-support/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
index f5c2567..25280a3 100644
--- a/core/camel-support/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
+++ b/core/camel-support/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
@@ -34,6 +34,7 @@ import javax.crypto.CipherInputStream;
import javax.crypto.CipherOutputStream;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.StreamCache;
import org.apache.camel.spi.StreamCachingStrategy;
@@ -231,7 +232,7 @@ public final class FileInputStreamCache extends InputStream implements StreamCac
streamCacheUnitOfWork.addSynchronization(onCompletion);
} else {
// add on completion so we can cleanup after the exchange is done such as deleting temporary files
- exchange.addOnCompletion(onCompletion);
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(onCompletion);
}
}
}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/DefaultConsumer.java b/core/camel-support/src/main/java/org/apache/camel/support/DefaultConsumer.java
index e263c9a..1248651 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/DefaultConsumer.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/DefaultConsumer.java
@@ -21,6 +21,7 @@ import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.Route;
import org.apache.camel.RouteAware;
@@ -93,11 +94,11 @@ public class DefaultConsumer extends ServiceSupport implements Consumer, RouteAw
// if the exchange doesn't have from route id set, then set it if it originated
// from this unit of work
if (route != null && exchange.getFromRouteId() == null) {
- exchange.setFromRouteId(route.getId());
+ exchange.adapt(ExtendedExchange.class).setFromRouteId(route.getId());
}
UnitOfWork uow = endpoint.getCamelContext().adapt(ExtendedCamelContext.class).getUnitOfWorkFactory().createUnitOfWork(exchange);
- exchange.setUnitOfWork(uow);
+ exchange.adapt(ExtendedExchange.class).setUnitOfWork(uow);
uow.start();
return uow;
}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/DefaultExchange.java b/core/camel-support/src/main/java/org/apache/camel/support/DefaultExchange.java
index b88ecdd..2533b0c 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/DefaultExchange.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/DefaultExchange.java
@@ -17,7 +17,6 @@
package org.apache.camel.support;
import java.util.ArrayList;
-import java.util.Date;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
@@ -539,7 +538,7 @@ public final class DefaultExchange implements ExtendedExchange {
public void handoverCompletions(Exchange target) {
if (onCompletions != null) {
for (Synchronization onCompletion : onCompletions) {
- target.addOnCompletion(onCompletion);
+ target.adapt(ExtendedExchange.class).addOnCompletion(onCompletion);
}
// cleanup the temporary on completion list as they have been handed over
onCompletions.clear();
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java b/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
index 52568cb..0696b72 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
@@ -35,6 +35,7 @@ import org.apache.camel.CamelExecutionException;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Message;
import org.apache.camel.MessageHistory;
import org.apache.camel.NoSuchBeanException;
@@ -254,7 +255,7 @@ public final class ExchangeHelper {
copy.getIn().setMessageId(null);
}
// do not share the unit of work
- copy.setUnitOfWork(null);
+ copy.adapt(ExtendedExchange.class).setUnitOfWork(null);
// do not reuse the message id
// hand over on completion to the copy if we got any
UnitOfWork uow = exchange.getUnitOfWork();
@@ -814,7 +815,7 @@ public final class ExchangeHelper {
}
if (handover) {
// Need to hand over the completion for async invocation
- exchange.handoverCompletions(answer);
+ exchange.adapt(ExtendedExchange.class).handoverCompletions(answer);
}
answer.setIn(exchange.getIn().copy());
if (exchange.hasOut()) {
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/UnitOfWorkHelper.java b/core/camel-support/src/main/java/org/apache/camel/support/UnitOfWorkHelper.java
index afdc6d2..95bb88a 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/UnitOfWorkHelper.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/UnitOfWorkHelper.java
@@ -21,6 +21,7 @@ import java.util.Collections;
import java.util.List;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Route;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.spi.SynchronizationRouteAware;
@@ -64,7 +65,7 @@ public final class UnitOfWorkHelper {
}
// remove uow from exchange as its done
- exchange.setUnitOfWork(null);
+ exchange.adapt(ExtendedExchange.class).setUnitOfWork(null);
}
public static void doneSynchronizations(Exchange exchange, List<Synchronization> synchronizations, Logger log) {