camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [1/7] camel git commit: CAMEL-8965: WireTap supports dynamic uris like enrich/pollEnrich now does.
Date Mon, 20 Jul 2015 20:47:22 GMT
Repository: camel
Updated Branches:
  refs/heads/master d14fa8c45 -> 01dead92e


CAMEL-8965: WireTap supports dynamic uris like enrich/pollEnrich now does.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/2696b57b
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/2696b57b
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/2696b57b

Branch: refs/heads/master
Commit: 2696b57b831712548d10be1651e053685d6e25da
Parents: d14fa8c
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Mon Jul 20 16:18:35 2015 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Mon Jul 20 22:54:07 2015 +0200

----------------------------------------------------------------------
 .../apache/camel/model/ProcessorDefinition.java |  88 ++++++++-----
 .../apache/camel/model/WireTapDefinition.java   | 125 +++----------------
 .../camel/processor/WireTapProcessor.java       |  42 +++----
 ...outeRemoveWireTapExplicitThreadPoolTest.java |   2 +-
 .../camel/processor/CBRWithWireTapTest.java     |   4 +-
 .../camel/processor/WireTapCustomPool2Test.java |   2 +-
 .../camel/processor/WireTapNewExchangeTest.java |   2 +-
 .../processor/WireTapOnPrepareRefTest.java      |   2 +-
 .../camel/processor/WireTapOnPrepareTest.java   |   2 +-
 ...eTapUsingFireAndForgetCopyAsDefaultTest.java |   4 +-
 .../WireTapUsingFireAndForgetCopyTest.java      |   4 +-
 .../FileWireTapWithXMLPayloadIssueTest.xml      |   4 +-
 .../SpringAggregateFromWireTapTest.xml          |   4 +-
 .../processor/SpringComplexBlockWithEndTest.xml |   4 +-
 .../processor/SpringWireTapNewExchangeTest.xml  |   3 +-
 .../spring/processor/SpringWireTapTest.xml      |  10 +-
 .../SpringWireTapUsingFireAndForgetCopyTest.xml |   7 +-
 .../SpringWireTapUsingFireAndForgetTest.xml     |   7 +-
 .../spring/processor/WireTapOnPrepareTest.xml   |   4 +-
 19 files changed, 133 insertions(+), 187 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java b/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
index 17ccb0e..91d0e6f 100644
--- a/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
@@ -1471,6 +1471,7 @@ public abstract class ProcessorDefinition<Type extends ProcessorDefinition<Type>
         addOutput(answer);
         return ExpressionClause.createAndSetExpression(answer);
     }
+
     /**
      * <a href="http://camel.apache.org/load-balancer.html">Load Balancer EIP:</a>
      * Creates a loadbalance
@@ -2255,14 +2256,28 @@ public abstract class ProcessorDefinition<Type extends ProcessorDefinition<Type>
      * destination gets a copy of the original message to avoid the processors
      * interfering with each other using {@link ExchangePattern#InOnly}.
      *
+     * @return the builder
+     */
+    public ExpressionClause<WireTapDefinition> wireTap() {
+        WireTapDefinition answer = new WireTapDefinition();
+        addOutput(answer);
+        return ExpressionClause.createAndSetExpression(answer);
+    }
+
+    /**
+     * <a href="http://camel.apache.org/wiretap.html">WireTap EIP:</a>
+     * Sends messages to all its child outputs; so that each processor and
+     * destination gets a copy of the original message to avoid the processors
+     * interfering with each other using {@link ExchangePattern#InOnly}.
+     *
      * @param uri  the destination
      * @return the builder
      */
-    public WireTapDefinition<Type> wireTap(String uri) {
-        WireTapDefinition<Type> answer = new WireTapDefinition<Type>();
-        answer.setUri(uri);
+    public Type wireTap(String uri) {
+        WireTapDefinition answer = new WireTapDefinition();
+        answer.setExpression(new SimpleExpression(uri));
         addOutput(answer);
-        return answer;
+        return (Type) this;
     }
 
     /**
@@ -2275,15 +2290,13 @@ public abstract class ProcessorDefinition<Type extends ProcessorDefinition<Type>
      * @param      executorService a custom {@link ExecutorService} to use as thread pool
      *             for sending tapped exchanges
      * @return the builder
-     * @deprecated use the fluent builder from {@link WireTapDefinition}, will be removed
in Camel 3.0
      */
-    @Deprecated
-    public WireTapDefinition<Type> wireTap(String uri, ExecutorService executorService)
{
-        WireTapDefinition<Type> answer = new WireTapDefinition<Type>();
-        answer.setUri(uri);
+    public Type wireTap(String uri, ExecutorService executorService) {
+        WireTapDefinition answer = new WireTapDefinition();
+        answer.setExpression(new SimpleExpression(uri));
         answer.setExecutorService(executorService);
         addOutput(answer);
-        return answer;
+        return (Type) this;
     }
 
     /**
@@ -2296,15 +2309,13 @@ public abstract class ProcessorDefinition<Type extends ProcessorDefinition<Type>
      * @param      executorServiceRef reference to lookup a custom {@link ExecutorService}
      *             to use as thread pool for sending tapped exchanges
      * @return the builder
-     * @deprecated use the fluent builder from {@link WireTapDefinition}, will be removed
in Camel 3.0
      */
-    @Deprecated
-    public WireTapDefinition<Type> wireTap(String uri, String executorServiceRef) {
-        WireTapDefinition<Type> answer = new WireTapDefinition<Type>();
-        answer.setUri(uri);
+    public Type wireTap(String uri, String executorServiceRef) {
+        WireTapDefinition answer = new WireTapDefinition();
+        answer.setExpression(new SimpleExpression(uri));
         answer.setExecutorServiceRef(executorServiceRef);
         addOutput(answer);
-        return answer;
+        return (Type) this;
     }
 
     /**
@@ -2318,10 +2329,8 @@ public abstract class ProcessorDefinition<Type extends ProcessorDefinition<Type>
      * @param uri  the destination
      * @param body expression that creates the body to send
      * @return the builder
-     * @deprecated use the fluent builder from {@link WireTapDefinition}, will be removed
in Camel 3.0
      */
-    @Deprecated
-    public WireTapDefinition<Type> wireTap(String uri, Expression body) {
+    public Type wireTap(String uri, Expression body) {
         return wireTap(uri, true, body);
     }
 
@@ -2332,18 +2341,33 @@ public abstract class ProcessorDefinition<Type extends ProcessorDefinition<Type>
      *
      * @param uri  the destination
      * @param copy whether or not use a copy of the original exchange or a new empty exchange
+     * @return the builder
+     */
+    public Type wireTap(String uri, boolean copy) {
+        WireTapDefinition answer = new WireTapDefinition();
+        answer.setExpression(new SimpleExpression(uri));
+        answer.setCopy(copy);
+        addOutput(answer);
+        return (Type) this;
+    }
+
+    /**
+     * <a href="http://camel.apache.org/wiretap.html">WireTap EIP:</a>
+     * Sends a new {@link org.apache.camel.Exchange} to the destination
+     * using {@link ExchangePattern#InOnly}.
+     *
+     * @param uri  the destination
+     * @param copy whether or not use a copy of the original exchange or a new empty exchange
      * @param body expression that creates the body to send
      * @return the builder
-     * @deprecated use the fluent builder from {@link WireTapDefinition}, will be removed
in Camel 3.0
      */
-    @Deprecated
-    public WireTapDefinition<Type> wireTap(String uri, boolean copy, Expression body)
{
-        WireTapDefinition<Type> answer = new WireTapDefinition<Type>();
-        answer.setUri(uri);
+    public Type wireTap(String uri, boolean copy, Expression body) {
+        WireTapDefinition answer = new WireTapDefinition();
+        answer.setExpression(new SimpleExpression(uri));
         answer.setCopy(copy);
         answer.setNewExchangeExpression(body);
         addOutput(answer);
-        return answer;
+        return (Type) this;
     }
 
     /**
@@ -2357,10 +2381,8 @@ public abstract class ProcessorDefinition<Type extends ProcessorDefinition<Type>
      * @param uri  the destination
      * @param processor  processor preparing the new exchange to send
      * @return the builder
-     * @deprecated use the fluent builder from {@link WireTapDefinition}, will be removed
in Camel 3.0
      */
-    @Deprecated
-    public WireTapDefinition<Type> wireTap(String uri, Processor processor) {
+    public Type wireTap(String uri, Processor processor) {
         return wireTap(uri, true, processor);
     }
 
@@ -2373,16 +2395,14 @@ public abstract class ProcessorDefinition<Type extends ProcessorDefinition<Type>
      * @param copy whether or not use a copy of the original exchange or a new empty exchange
      * @param processor  processor preparing the new exchange to send
      * @return the builder
-     * @deprecated use the fluent builder from {@link WireTapDefinition}, will be removed
in Camel 3.0
      */
-    @Deprecated
-    public WireTapDefinition<Type> wireTap(String uri, boolean copy, Processor processor)
{
-        WireTapDefinition<Type> answer = new WireTapDefinition<Type>();
-        answer.setUri(uri);
+    public Type wireTap(String uri, boolean copy, Processor processor) {
+        WireTapDefinition answer = new WireTapDefinition();
+        answer.setExpression(new SimpleExpression(uri));
         answer.setCopy(copy);
         answer.setNewExchangeProcessor(processor);
         addOutput(answer);
-        return answer;
+        return (Type) this;
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java b/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
index 49631bb..ebe95b1 100644
--- a/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
@@ -27,12 +27,11 @@ import javax.xml.bind.annotation.XmlElementRef;
 import javax.xml.bind.annotation.XmlRootElement;
 import javax.xml.bind.annotation.XmlTransient;
 
-import org.apache.camel.Endpoint;
 import org.apache.camel.ExchangePattern;
 import org.apache.camel.Expression;
 import org.apache.camel.Processor;
-import org.apache.camel.Producer;
 import org.apache.camel.processor.CamelInternalProcessor;
+import org.apache.camel.processor.SendDynamicProcessor;
 import org.apache.camel.processor.WireTapProcessor;
 import org.apache.camel.spi.Metadata;
 import org.apache.camel.spi.RouteContext;
@@ -44,15 +43,7 @@ import org.apache.camel.util.CamelContextHelper;
 @Metadata(label = "eip,endpoint,routing")
 @XmlRootElement(name = "wireTap")
 @XmlAccessorType(XmlAccessType.FIELD)
-public class WireTapDefinition<Type extends ProcessorDefinition<Type>> extends
NoOutputDefinition<WireTapDefinition<Type>>
-        implements ExecutorServiceAwareDefinition<WireTapDefinition<Type>>, EndpointRequiredDefinition
{
-    @XmlAttribute
-    protected String uri;
-    @XmlAttribute
-    @Deprecated
-    protected String ref;
-    @XmlTransient
-    protected Endpoint endpoint;
+public class WireTapDefinition extends NoOutputExpressionNode implements ExecutorServiceAwareDefinition<WireTapDefinition>
{
     @XmlTransient
     private Processor newExchangeProcessor;
     @XmlAttribute(name = "processorRef")
@@ -75,37 +66,17 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
     public WireTapDefinition() {
     }
 
-    public WireTapDefinition(String uri) {
-        setUri(uri);
-    }
-
-    public WireTapDefinition(Endpoint endpoint) {
-        setEndpoint(endpoint);
-    }
-
-    @Override
-    public String getEndpointUri() {
-        if (uri != null) {
-            return uri;
-        } else if (endpoint != null) {
-            return endpoint.getEndpointUri();
-        } else {
-            return null;
-        }
-    }
-
     @Override
     public Processor createProcessor(RouteContext routeContext) throws Exception {
         // executor service is mandatory for wire tap
         boolean shutdownThreadPool = ProcessorDefinitionHelper.willCreateNewThreadPool(routeContext,
this, true);
         ExecutorService threadPool = ProcessorDefinitionHelper.getConfiguredExecutorService(routeContext,
"WireTap", this, true);
 
-        // create the producer to send to the wire tapped endpoint
-        Endpoint endpoint = resolveEndpoint(routeContext);
-        Producer producer = endpoint.createProducer();
+        // create the send dynamic producer to send to the wire tapped endpoint
+        Processor dynamicTo = new SendDynamicProcessor(null, getExpression());
 
         // create error handler we need to use for processing the wire tapped
-        Processor target = wrapInErrorHandler(routeContext, producer);
+        Processor target = wrapInErrorHandler(routeContext, dynamicTo);
 
         // and wrap in unit of work
         CamelInternalProcessor internal = new CamelInternalProcessor(target);
@@ -114,7 +85,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
         // is true bt default
         boolean isCopy = getCopy() == null || getCopy();
 
-        WireTapProcessor answer = new WireTapProcessor(endpoint, internal, getPattern(),
threadPool, shutdownThreadPool);
+        WireTapProcessor answer = new WireTapProcessor(getExpression(), internal, getPattern(),
threadPool, shutdownThreadPool);
         answer.setCopy(isCopy);
         if (newExchangeProcessorRef != null) {
             newExchangeProcessor = routeContext.mandatoryLookup(newExchangeProcessorRef,
Processor.class);
@@ -147,37 +118,12 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
 
     @Override
     public String toString() {
-        return "WireTap[" + description() + "]";
+        return "WireTap[" + getExpression() + "]";
     }
     
-    protected String description() {
-        return FromDefinition.description(getUri(), getRef(), getEndpoint());
-    }
-
     @Override
     public String getLabel() {
-        return "wireTap[" + description() + "]";
-    }
-
-    @Override
-    @SuppressWarnings("unchecked")
-    public Type end() {
-        // allow end() to return to previous type so you can continue in the DSL
-        return (Type) super.end();
-    }
-
-    @Override
-    public void addOutput(ProcessorDefinition<?> output) {
-        // add outputs on parent as this wiretap does not support outputs
-        getParent().addOutput(output);
-    }
-
-    public Endpoint resolveEndpoint(RouteContext context) {
-        if (endpoint == null) {
-            return context.resolveEndpoint(getUri(), getRef());
-        } else {
-            return endpoint;
-        }
+        return "wireTap[" + getExpression() + "]";
     }
 
     // Fluent API
@@ -190,7 +136,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      *                        for sending tapped exchanges
      * @return the builder
      */
-    public WireTapDefinition<Type> executorService(ExecutorService executorService)
{
+    public WireTapDefinition executorService(ExecutorService executorService) {
         setExecutorService(executorService);
         return this;
     }
@@ -202,7 +148,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      *                           to use as thread pool for sending tapped exchanges
      * @return the builder
      */
-    public WireTapDefinition<Type> executorServiceRef(String executorServiceRef) {
+    public WireTapDefinition executorServiceRef(String executorServiceRef) {
         setExecutorServiceRef(executorServiceRef);
         return this;
     }
@@ -212,7 +158,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      *
      * @return the builder
      */
-    public WireTapDefinition<Type> copy() {
+    public WireTapDefinition copy() {
         setCopy(true);
         return this;
     }
@@ -224,7 +170,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      *             if it is false camel will not copy the original exchange 
      * @return the builder
      */
-    public WireTapDefinition<Type> copy(boolean copy) {
+    public WireTapDefinition copy(boolean copy) {
         setCopy(copy);
         return this;
     }
@@ -233,7 +179,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      * @deprecated will be removed in Camel 3.0 Instead use {@link #newExchangeBody(org.apache.camel.Expression)}
      */
     @Deprecated
-    public WireTapDefinition<Type> newExchange(Expression expression) {
+    public WireTapDefinition newExchange(Expression expression) {
         return newExchangeBody(expression);
     }
 
@@ -244,7 +190,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      * @return the builder
      * @see #newExchangeHeader(String, org.apache.camel.Expression)
      */
-    public WireTapDefinition<Type> newExchangeBody(Expression expression) {
+    public WireTapDefinition newExchangeBody(Expression expression) {
         setNewExchangeExpression(expression);
         return this;
     }
@@ -256,7 +202,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      *            be used for preparing the new exchange to send
      * @return the builder
      */
-    public WireTapDefinition<Type> newExchangeRef(String ref) {
+    public WireTapDefinition newExchangeRef(String ref) {
         setNewExchangeProcessorRef(ref);
         return this;
     }
@@ -268,7 +214,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      * @return the builder
      * @see #newExchangeHeader(String, org.apache.camel.Expression)
      */
-    public WireTapDefinition<Type> newExchange(Processor processor) {
+    public WireTapDefinition newExchange(Processor processor) {
         setNewExchangeProcessor(processor);
         return this;
     }
@@ -283,7 +229,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      * @param expression  the expression setting the header value
      * @return the builder
      */
-    public WireTapDefinition<Type> newExchangeHeader(String headerName, Expression
expression) {
+    public WireTapDefinition newExchangeHeader(String headerName, Expression expression)
{
         headers.add(new SetHeaderDefinition(headerName, expression));
         return this;
     }
@@ -296,7 +242,7 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      * @param onPrepare the processor
      * @return the builder
      */
-    public WireTapDefinition<Type> onPrepare(Processor onPrepare) {
+    public WireTapDefinition onPrepare(Processor onPrepare) {
         setOnPrepare(onPrepare);
         return this;
     }
@@ -309,44 +255,11 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>>
extends N
      * @param onPrepareRef reference to the processor to lookup in the {@link org.apache.camel.spi.Registry}
      * @return the builder
      */
-    public WireTapDefinition<Type> onPrepareRef(String onPrepareRef) {
+    public WireTapDefinition onPrepareRef(String onPrepareRef) {
         setOnPrepareRef(onPrepareRef);
         return this;
     }
 
-    public String getUri() {
-        return uri;
-    }
-
-    /**
-     * Uri of the endpoint to use as wire tap
-     */
-    public void setUri(String uri) {
-        this.uri = uri;
-    }
-
-    public String getRef() {
-        return ref;
-    }
-
-    /**
-     * Reference of the endpoint to use as wire tap
-     *
-     * @deprecated use uri with ref:uri instead
-     */
-    @Deprecated
-    public void setRef(String ref) {
-        this.ref = ref;
-    }
-
-    public Endpoint getEndpoint() {
-        return endpoint;
-    }
-
-    public void setEndpoint(Endpoint endpoint) {
-        this.endpoint = endpoint;
-    }
-
     public Processor getNewExchangeProcessor() {
         return newExchangeProcessor;
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
index 1d6b835..87a3365 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
@@ -24,8 +24,8 @@ import java.util.concurrent.ExecutorService;
 
 import org.apache.camel.AsyncCallback;
 import org.apache.camel.AsyncProcessor;
-import org.apache.camel.Endpoint;
-import org.apache.camel.EndpointAware;
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
 import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePattern;
 import org.apache.camel.Expression;
@@ -37,11 +37,9 @@ import org.apache.camel.impl.DefaultExchange;
 import org.apache.camel.spi.IdAware;
 import org.apache.camel.support.ServiceSupport;
 import org.apache.camel.util.AsyncProcessorHelper;
-import org.apache.camel.util.EventHelper;
 import org.apache.camel.util.ExchangeHelper;
 import org.apache.camel.util.ObjectHelper;
 import org.apache.camel.util.ServiceHelper;
-import org.apache.camel.util.StopWatch;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -50,10 +48,11 @@ import org.slf4j.LoggerFactory;
  *
  * @version 
  */
-public class WireTapProcessor extends ServiceSupport implements AsyncProcessor, Traceable,
EndpointAware, IdAware {
+public class WireTapProcessor extends ServiceSupport implements AsyncProcessor, Traceable,
IdAware, CamelContextAware {
     private static final Logger LOG = LoggerFactory.getLogger(WireTapProcessor.class);
     private String id;
-    private final Endpoint destination;
+    private CamelContext camelContext;
+    private final Expression expression;
     private final Processor processor;
     private final ExchangePattern exchangePattern;
     private final ExecutorService executorService;
@@ -66,9 +65,9 @@ public class WireTapProcessor extends ServiceSupport implements AsyncProcessor,
     private boolean copy;
     private Processor onPrepare;
 
-    public WireTapProcessor(Endpoint destination, Processor processor, ExchangePattern exchangePattern,
+    public WireTapProcessor(Expression expression, Processor processor, ExchangePattern exchangePattern,
                             ExecutorService executorService, boolean shutdownExecutorService)
{
-        this.destination = destination;
+        this.expression = expression;
         this.processor = processor;
         this.exchangePattern = exchangePattern;
         ObjectHelper.notNull(executorService, "executorService");
@@ -78,12 +77,12 @@ public class WireTapProcessor extends ServiceSupport implements AsyncProcessor,
 
     @Override
     public String toString() {
-        return "WireTap[" + destination + "]";
+        return "WireTap[" + expression + "]";
     }
 
     @Override
     public String getTraceLabel() {
-        return "wireTap(" + destination + ")";
+        return "wireTap(" + expression + ")";
     }
 
     public String getId() {
@@ -94,8 +93,12 @@ public class WireTapProcessor extends ServiceSupport implements AsyncProcessor,
         this.id = id;
     }
 
-    public Endpoint getEndpoint() {
-        return destination;
+    public CamelContext getCamelContext() {
+        return camelContext;
+    }
+
+    public void setCamelContext(CamelContext camelContext) {
+        this.camelContext = camelContext;
     }
 
     public void process(Exchange exchange) throws Exception {
@@ -122,17 +125,11 @@ public class WireTapProcessor extends ServiceSupport implements AsyncProcessor,
         // send the exchange to the destination using an executor service
         executorService.submit(new Callable<Exchange>() {
             public Exchange call() throws Exception {
-                final StopWatch watch = new StopWatch();
                 try {
-                    EventHelper.notifyExchangeSending(wireTapExchange.getContext(), wireTapExchange,
destination);
-                    LOG.debug(">>>> (wiretap) {} {}", destination, wireTapExchange);
+                    LOG.debug(">>>> (wiretap) {} {}", expression, wireTapExchange);
                     processor.process(wireTapExchange);
                 } catch (Throwable e) {
-                    LOG.warn("Error occurred during processing " + wireTapExchange + " wiretap
to " + destination + ". This exception will be ignored.", e);
-                } finally {
-                    // emit event that the exchange was sent to the endpoint
-                    long timeTaken = watch.stop();
-                    EventHelper.notifyExchangeSent(wireTapExchange.getContext(), wireTapExchange,
destination, timeTaken);
+                    LOG.warn("Error occurred during processing " + wireTapExchange + " wiretap
to " + expression + ". This exception will be ignored.", e);
                 }
                 return wireTapExchange;
             }
@@ -154,9 +151,6 @@ public class WireTapProcessor extends ServiceSupport implements AsyncProcessor,
             answer = configureNewExchange(exchange);
         }
 
-        // set property which endpoint we send to
-        answer.setProperty(Exchange.TO_ENDPOINT, destination.getEndpointUri());
-
         // prepare the exchange
         if (newExchangeExpression != null) {
             Object body = newExchangeExpression.evaluate(answer, Object.class);
@@ -263,7 +257,7 @@ public class WireTapProcessor extends ServiceSupport implements AsyncProcessor,
     protected void doShutdown() throws Exception {
         ServiceHelper.stopAndShutdownService(processor);
         if (shutdownExecutorService) {
-            destination.getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
+            getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/test/java/org/apache/camel/management/ManagedRouteRemoveWireTapExplicitThreadPoolTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedRouteRemoveWireTapExplicitThreadPoolTest.java
b/camel-core/src/test/java/org/apache/camel/management/ManagedRouteRemoveWireTapExplicitThreadPoolTest.java
index 24e365f..e593349 100644
--- a/camel-core/src/test/java/org/apache/camel/management/ManagedRouteRemoveWireTapExplicitThreadPoolTest.java
+++ b/camel-core/src/test/java/org/apache/camel/management/ManagedRouteRemoveWireTapExplicitThreadPoolTest.java
@@ -101,7 +101,7 @@ public class ManagedRouteRemoveWireTapExplicitThreadPoolTest extends ManagementT
                 // create a new thread pool to use for wire tap
                 myThreadPool = Executors.newFixedThreadPool(1);
 
-                from("seda:foo").routeId("foo").wireTap("direct:tap").executorService(myThreadPool).to("mock:result");
+                from("seda:foo").routeId("foo").wireTap("direct:tap", myThreadPool).to("mock:result");
                 from("direct:tap").routeId("tap").to("mock:tap");
             }
         };

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/test/java/org/apache/camel/processor/CBRWithWireTapTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/CBRWithWireTapTest.java b/camel-core/src/test/java/org/apache/camel/processor/CBRWithWireTapTest.java
index 0dd7cad..578a8ed 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/CBRWithWireTapTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/CBRWithWireTapTest.java
@@ -62,9 +62,9 @@ public class CBRWithWireTapTest extends ContextTestSupport {
                 from("direct:start")
                     .choice()
                         .when(body().contains("Camel"))
-                            .wireTap("mock:camel").end()
+                            .wireTap("mock:camel")
                         .when(body().contains("Donkey"))
-                            .wireTap("mock:donkey").copy().end()
+                            .wireTap("mock:donkey", true)
                         .otherwise()
                             .to("mock:other");
             }

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/test/java/org/apache/camel/processor/WireTapCustomPool2Test.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/WireTapCustomPool2Test.java
b/camel-core/src/test/java/org/apache/camel/processor/WireTapCustomPool2Test.java
index a17836e..7f607e1 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/WireTapCustomPool2Test.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/WireTapCustomPool2Test.java
@@ -67,7 +67,7 @@ public class WireTapCustomPool2Test extends ContextTestSupport {
                 from("direct:start")
                     .to("log:foo")
                     // pass in the custom pool to the wireTap DSL
-                    .wireTap("direct:tap").executorService(pool)
+                    .wireTap().constant("direct:tap").executorService(pool)
                     .to("mock:result");
                 // END SNIPPET: e1
 

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/test/java/org/apache/camel/processor/WireTapNewExchangeTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/WireTapNewExchangeTest.java
b/camel-core/src/test/java/org/apache/camel/processor/WireTapNewExchangeTest.java
index 2ecff83..43232ee 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/WireTapNewExchangeTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/WireTapNewExchangeTest.java
@@ -52,7 +52,7 @@ public class WireTapNewExchangeTest extends ContextTestSupport {
                 from("direct:start")
                     // tap a new message and send it to direct:tap
                     // the new message should be Bye World with 2 headers
-                    .wireTap("direct:tap")
+                    .wireTap().constant("direct:tap")
                         // create the new tap message body and headers
                         .newExchangeBody(constant("Bye World"))
                         .newExchangeHeader("id", constant(123))

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/test/java/org/apache/camel/processor/WireTapOnPrepareRefTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/WireTapOnPrepareRefTest.java
b/camel-core/src/test/java/org/apache/camel/processor/WireTapOnPrepareRefTest.java
index ce4ae2d..2d3915f 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/WireTapOnPrepareRefTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/WireTapOnPrepareRefTest.java
@@ -37,7 +37,7 @@ public class WireTapOnPrepareRefTest extends WireTapOnPrepareTest {
             @Override
             public void configure() throws Exception {
                 from("direct:start")
-                    .wireTap("direct:a").onPrepareRef("deepClone")
+                    .wireTap().constant("direct:a").onPrepareRef("deepClone")
                     .to("direct:b");
 
                 from("direct:a").process(new ProcessorA()).to("mock:a");

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/test/java/org/apache/camel/processor/WireTapOnPrepareTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/WireTapOnPrepareTest.java
b/camel-core/src/test/java/org/apache/camel/processor/WireTapOnPrepareTest.java
index 64f04d7..68866c5 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/WireTapOnPrepareTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/WireTapOnPrepareTest.java
@@ -51,7 +51,7 @@ public class WireTapOnPrepareTest extends ContextTestSupport {
             public void configure() throws Exception {
                 // START SNIPPET: e1
                 from("direct:start")
-                    .wireTap("direct:a").onPrepare(new AnimalDeepClonePrepare())
+                    .wireTap("direct:a", new AnimalDeepClonePrepare())
                     .to("direct:b");
                 // END SNIPPET: e1
 

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetCopyAsDefaultTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetCopyAsDefaultTest.java
b/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetCopyAsDefaultTest.java
index 3ea4043..2ac11fed 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetCopyAsDefaultTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetCopyAsDefaultTest.java
@@ -80,7 +80,7 @@ public class WireTapUsingFireAndForgetCopyAsDefaultTest extends ContextTestSuppo
             @Override
             public void configure() throws Exception {
                 from("direct:start")
-                    .wireTap("direct:foo").newExchange(new Processor() {
+                    .wireTap("direct:foo", new Processor() {
                         public void process(Exchange exchange) throws Exception {
                             String body = exchange.getIn().getBody(String.class);
                             exchange.getIn().setBody("Bye " + body);
@@ -156,7 +156,7 @@ public class WireTapUsingFireAndForgetCopyAsDefaultTest extends ContextTestSuppo
             @Override
             public void configure() throws Exception {
                 from("direct:start")
-                    .wireTap("direct:foo").newExchangeBody(simple("Bye ${body}"))
+                    .wireTap("direct:foo", simple("Bye ${body}"))
                     .to("mock:result");
 
                 from("direct:foo").to("mock:foo");

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetCopyTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetCopyTest.java
b/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetCopyTest.java
index ffb3aee..d1c6a60 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetCopyTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/WireTapUsingFireAndForgetCopyTest.java
@@ -80,7 +80,7 @@ public class WireTapUsingFireAndForgetCopyTest extends ContextTestSupport
{
             @Override
             public void configure() throws Exception {
                 from("direct:start")
-                    .wireTap("direct:foo").copy().newExchange(new Processor() {
+                    .wireTap("direct:foo", true, new Processor() {
                         public void process(Exchange exchange) throws Exception {
                             String body = exchange.getIn().getBody(String.class);
                             exchange.getIn().setBody("Bye " + body);
@@ -156,7 +156,7 @@ public class WireTapUsingFireAndForgetCopyTest extends ContextTestSupport
{
             @Override
             public void configure() throws Exception {
                 from("direct:start")
-                    .wireTap("direct:foo").copy().newExchangeBody(simple("Bye ${body}"))
+                    .wireTap("direct:foo", true, simple("Bye ${body}"))
                     .to("mock:result");
 
                 from("direct:foo").to("mock:foo");

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/components/camel-spring/src/test/resources/org/apache/camel/spring/issues/FileWireTapWithXMLPayloadIssueTest.xml
----------------------------------------------------------------------
diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/issues/FileWireTapWithXMLPayloadIssueTest.xml
b/components/camel-spring/src/test/resources/org/apache/camel/spring/issues/FileWireTapWithXMLPayloadIssueTest.xml
index 386ffe1..f538243 100644
--- a/components/camel-spring/src/test/resources/org/apache/camel/spring/issues/FileWireTapWithXMLPayloadIssueTest.xml
+++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/issues/FileWireTapWithXMLPayloadIssueTest.xml
@@ -26,7 +26,9 @@
         <route>
             <from uri="file://target/xmldata"/>
             <convertBodyTo type="java.lang.String"/>
-            <wireTap uri="mock:wiretap"/>
+            <wireTap>
+              <constant>mock:wiretap</constant>
+            </wireTap>
             <to uri="mock:result"/>
         </route>
     </camelContext>

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringAggregateFromWireTapTest.xml
----------------------------------------------------------------------
diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringAggregateFromWireTapTest.xml
b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringAggregateFromWireTapTest.xml
index 39677af..c461c57 100644
--- a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringAggregateFromWireTapTest.xml
+++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringAggregateFromWireTapTest.xml
@@ -25,7 +25,9 @@
     <camelContext xmlns="http://camel.apache.org/schema/spring">
         <route>
             <from uri="direct:start"/>
-            <wireTap uri="direct:tap"/>
+            <wireTap>
+                <constant>direct:tap</constant>
+            </wireTap>
             <to uri="mock:end"/>
         </route>
 

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringComplexBlockWithEndTest.xml
----------------------------------------------------------------------
diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringComplexBlockWithEndTest.xml
b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringComplexBlockWithEndTest.xml
index 1bf114c..686e270 100644
--- a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringComplexBlockWithEndTest.xml
+++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringComplexBlockWithEndTest.xml
@@ -44,7 +44,9 @@
                     <to uri="log:otherwise"/>
                     <to uri="mock:otherwise"/>
                     <multicast streaming="true">
-                        <wireTap uri="mock:trapped"/>
+                        <wireTap>
+                          <constant>mock:trapped</constant>
+                        </wireTap>
                         <split strategyRef="splitAggregate">
                             <tokenize token=","/>
                             <filter>

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapNewExchangeTest.xml
----------------------------------------------------------------------
diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapNewExchangeTest.xml
b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapNewExchangeTest.xml
index 3b81716..4e17834 100644
--- a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapNewExchangeTest.xml
+++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapNewExchangeTest.xml
@@ -28,7 +28,8 @@
             <from uri="direct:start"/>
             <!-- tap a new message and send it to direct:tap -->
             <!-- the new message should be Bye World with 2 headers -->
-            <wireTap uri="direct:tap">
+            <wireTap>
+              <constant>direct:tap</constant>
                 <!-- create the new tap message body and headers -->
                 <body><constant>Bye World</constant></body>
                 <setHeader headerName="id"><constant>123</constant></setHeader>

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapTest.xml
----------------------------------------------------------------------
diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapTest.xml
b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapTest.xml
index 9abd708..42b1efb 100644
--- a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapTest.xml
+++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapTest.xml
@@ -28,8 +28,10 @@
         <route>
             <from uri="direct:start"/>
             <to uri="log:foo"/>
-            <wireTap uri="direct:tap"/>
-            <to uri="mock:result"/>
+            <wireTap>
+              <constant>direct:tap</constant>
+            </wireTap>
+          <to uri="mock:result"/>
         </route>
         <!-- END SNIPPET: e1 -->
 
@@ -43,7 +45,9 @@
         
         <route>
         	<from uri="direct:test"/>
-        	<wireTap uri="direct:a" id="wiretap_1" />
+        	<wireTap id="wiretap_1">
+            <constant>direct:a</constant>
+          </wireTap>
         	<to uri="mock:a"/>
         </route>
         

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetCopyTest.xml
----------------------------------------------------------------------
diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetCopyTest.xml
b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetCopyTest.xml
index d2d94c8..3a01f27 100644
--- a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetCopyTest.xml
+++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetCopyTest.xml
@@ -27,7 +27,8 @@
         <!-- START SNIPPET: e1 -->
         <route>
             <from uri="direct:start"/>
-            <wireTap uri="direct:foo">
+            <wireTap>
+                <constant>direct:foo</constant>
                 <body><simple>Bye ${body}</simple></body>
             </wireTap>
             <to uri="mock:result"/>
@@ -37,7 +38,9 @@
         <!-- START SNIPPET: e2 -->
         <route>
             <from uri="direct:start2"/>
-            <wireTap uri="direct:foo" processorRef="myProcessor"/>
+            <wireTap processorRef="myProcessor">
+                <constant>direct:foo</constant>
+            </wireTap>
             <to uri="mock:result"/>
         </route>
         <!-- END SNIPPET: e2 -->

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetTest.xml
----------------------------------------------------------------------
diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetTest.xml
b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetTest.xml
index 99becd0..ad39400 100644
--- a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetTest.xml
+++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringWireTapUsingFireAndForgetTest.xml
@@ -27,7 +27,8 @@
         <!-- START SNIPPET: e1 -->
         <route>
             <from uri="direct:start"/>
-            <wireTap uri="direct:foo">
+            <wireTap>
+                <constant>direct:foo</constant>
                 <body><constant>Bye World</constant></body>
             </wireTap>
             <to uri="mock:result"/>
@@ -37,7 +38,9 @@
         <!-- START SNIPPET: e2 -->
         <route>
             <from uri="direct:start2"/>
-            <wireTap uri="direct:foo" processorRef="myProcessor"/>
+            <wireTap processorRef="myProcessor">
+                <constant>direct:foo</constant>
+            </wireTap>
             <to uri="mock:result"/>
         </route>
         <!-- END SNIPPET: e2 -->

http://git-wip-us.apache.org/repos/asf/camel/blob/2696b57b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/WireTapOnPrepareTest.xml
----------------------------------------------------------------------
diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/WireTapOnPrepareTest.xml
b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/WireTapOnPrepareTest.xml
index 3ed30d1..c9ae5ad 100644
--- a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/WireTapOnPrepareTest.xml
+++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/WireTapOnPrepareTest.xml
@@ -27,7 +27,9 @@
         <route>
             <from uri="direct:start"/>
             <!-- use on prepare with wiretap -->
-            <wireTap uri="direct:a" onPrepareRef="animalDeepClonePrepare"/>
+            <wireTap onPrepareRef="animalDeepClonePrepare">
+                <constant>direct:a</constant>
+            </wireTap>
             <to uri="direct:b"/>
         </route>
 


Mime
View raw message