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 2012/03/15 18:40:19 UTC
svn commit: r1301115 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/model/WireTapDefinition.java
main/java/org/apache/camel/processor/WireTapProcessor.java
test/java/org/apache/camel/processor/WireTapOnExceptionTest.java
Author: davsclaus
Date: Thu Mar 15 17:40:18 2012
New Revision: 1301115
URL: http://svn.apache.org/viewvc?rev=1301115&view=rev
Log:
CAMEL-5090: WireTap now runs in UoW and error handling when tapping to destination.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/WireTapOnExceptionTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java?rev=1301115&r1=1301114&r2=1301115&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java Thu Mar 15 17:40:18 2012
@@ -32,6 +32,8 @@ 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.UnitOfWorkProcessor;
import org.apache.camel.processor.WireTapProcessor;
import org.apache.camel.spi.RouteContext;
import org.apache.camel.util.CamelContextHelper;
@@ -80,13 +82,19 @@ public class WireTapDefinition<Type exte
@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
- Endpoint endpoint = resolveEndpoint(routeContext);
-
// executor service is mandatory for wire tap
boolean shutdownThreadPool = ProcessorDefinitionHelper.willCreateNewThreadPool(routeContext, this, true);
ExecutorService threadPool = ProcessorDefinitionHelper.getConfiguredExecutorService(routeContext, "WireTap", this, true);
- WireTapProcessor answer = new WireTapProcessor(endpoint, getPattern(), threadPool, shutdownThreadPool);
+ // create the producer to send to the wire tapped endpoint
+ Endpoint endpoint = resolveEndpoint(routeContext);
+ Producer producer = endpoint.createProducer();
+ // create error handler we need to use for processing the wire tapped
+ Processor target = wrapInErrorHandler(routeContext, producer);
+ // and wrap in UoW, which is needed for error handler as well
+ target = new UnitOfWorkProcessor(routeContext, target);
+
+ WireTapProcessor answer = new WireTapProcessor(endpoint, target, getPattern(), threadPool, shutdownThreadPool);
answer.setCopy(isCopy());
if (newExchangeProcessorRef != null) {
newExchangeProcessor = routeContext.lookup(newExchangeProcessorRef, Processor.class);
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java?rev=1301115&r1=1301114&r2=1301115&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java Thu Mar 15 17:40:18 2012
@@ -22,23 +22,32 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Expression;
import org.apache.camel.Processor;
-import org.apache.camel.Producer;
-import org.apache.camel.ProducerCallback;
+import org.apache.camel.Traceable;
import org.apache.camel.impl.DefaultExchange;
+import org.apache.camel.support.ServiceSupport;
+import org.apache.camel.util.AsyncProcessorHelper;
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.ServiceHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Processor for wire tapping exchanges to an endpoint destination.
*
* @version
*/
-public class WireTapProcessor extends SendProcessor {
+public class WireTapProcessor extends ServiceSupport implements AsyncProcessor, Traceable {
+ private final static transient Logger LOG = LoggerFactory.getLogger(WireTapProcessor.class);
+ private final Endpoint destination;
+ private final Processor processor;
+ private final ExchangePattern exchangePattern;
private final ExecutorService executorService;
private volatile boolean shutdownExecutorService;
@@ -49,15 +58,11 @@ public class WireTapProcessor extends Se
private boolean copy;
private Processor onPrepare;
- public WireTapProcessor(Endpoint destination, ExecutorService executorService, boolean shutdownExecutorService) {
- super(destination);
- ObjectHelper.notNull(executorService, "executorService");
- this.executorService = executorService;
- this.shutdownExecutorService = shutdownExecutorService;
- }
-
- public WireTapProcessor(Endpoint destination, ExchangePattern pattern, ExecutorService executorService, boolean shutdownExecutorService) {
- super(destination, pattern);
+ public WireTapProcessor(Endpoint destination, Processor processor, ExchangePattern exchangePattern,
+ ExecutorService executorService, boolean shutdownExecutorService) {
+ this.destination = destination;
+ this.processor = processor;
+ this.exchangePattern = exchangePattern;
ObjectHelper.notNull(executorService, "executorService");
this.executorService = executorService;
this.shutdownExecutorService = shutdownExecutorService;
@@ -74,25 +79,7 @@ public class WireTapProcessor extends Se
}
public void process(Exchange exchange) throws Exception {
- if (!isStarted()) {
- throw new IllegalStateException("WireTapProcessor has not been started: " + this);
- }
-
- // must configure the wire tap beforehand
- final Exchange wireTapExchange = configureExchange(exchange, pattern);
-
- // send the exchange to the destination using an executor service
- executorService.submit(new Callable<Exchange>() {
- public Exchange call() throws Exception {
- return producerCache.doInProducer(destination, wireTapExchange, pattern, new ProducerCallback<Exchange>() {
- public Exchange doInProducer(Producer producer, Exchange exchange, ExchangePattern pattern) throws Exception {
- log.debug(">>>> (wiretap) {} {}", destination, exchange);
- producer.process(exchange);
- return exchange;
- }
- });
- };
- });
+ AsyncProcessorHelper.process(this, exchange);
}
public boolean process(Exchange exchange, final AsyncCallback callback) {
@@ -101,18 +88,18 @@ public class WireTapProcessor extends Se
}
// must configure the wire tap beforehand
- final Exchange wireTapExchange = configureExchange(exchange, pattern);
+ final Exchange wireTapExchange = configureExchange(exchange, exchangePattern);
// send the exchange to the destination using an executor service
executorService.submit(new Callable<Exchange>() {
public Exchange call() throws Exception {
- return producerCache.doInProducer(destination, wireTapExchange, pattern, new ProducerCallback<Exchange>() {
- public Exchange doInProducer(Producer producer, Exchange exchange, ExchangePattern pattern) throws Exception {
- log.debug(">>>> (wiretap) {} {}", destination, exchange);
- producer.process(exchange);
- return exchange;
- }
- });
+ try {
+ LOG.debug(">>>> (wiretap) {} {}", destination, wireTapExchange);
+ processor.process(wireTapExchange);
+ } catch (Throwable e) {
+ LOG.warn("Error occurred during processing " + wireTapExchange + " wiretap to " + destination + ". This exception will be ignored.", e);
+ }
+ return wireTapExchange;
};
});
@@ -122,7 +109,6 @@ public class WireTapProcessor extends Se
}
- @Override
protected Exchange configureExchange(Exchange exchange, ExchangePattern pattern) {
Exchange answer;
if (copy) {
@@ -178,13 +164,6 @@ public class WireTapProcessor extends Se
return new DefaultExchange(exchange.getFromEndpoint(), ExchangePattern.InOnly);
}
- protected void doShutdown() throws Exception {
- if (shutdownExecutorService) {
- getDestination().getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
- }
- super.doShutdown();
- }
-
public List<Processor> getNewExchangeProcessors() {
return newExchangeProcessors;
}
@@ -224,4 +203,21 @@ public class WireTapProcessor extends Se
this.onPrepare = onPrepare;
}
+ @Override
+ protected void doStart() throws Exception {
+ ServiceHelper.startService(processor);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ ServiceHelper.stopService(processor);
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ ServiceHelper.stopAndShutdownService(processor);
+ if (shutdownExecutorService) {
+ destination.getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
+ }
+ }
}
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/WireTapOnExceptionTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/WireTapOnExceptionTest.java?rev=1301115&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/WireTapOnExceptionTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/WireTapOnExceptionTest.java Thu Mar 15 17:40:18 2012
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.impl.DefaultProducer;
+
+/**
+ *
+ */
+public class WireTapOnExceptionTest extends ContextTestSupport {
+
+ public void testWireTapOnException() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+ getMockEndpoint("mock:error").expectedMessageCount(1);
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ MyProducerFailEndpoint my = new MyProducerFailEndpoint("myEndpoint://foo", getContext());
+ context.addEndpoint("myEndpoint://foo", my);
+
+ onException(IllegalArgumentException.class).to("mock:error");
+
+ from("direct:start").wireTap("myEndpoint:foo").to("mock:result");
+ }
+ };
+ }
+
+ public static class MyProducerFailEndpoint extends DefaultEndpoint {
+
+ private MyProducerFailEndpoint(String endpointUri, CamelContext camelContext) {
+ super(endpointUri, camelContext);
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new DefaultProducer(this) {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ throw new IllegalArgumentException("Forced");
+ }
+ };
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return null;
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+
+ }
+}