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 2010/06/17 13:04:35 UTC
svn commit: r955553 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/processor/ test/java/org/apache/camel/management/
test/java/org/apache/camel/processor/async/
Author: davsclaus
Date: Thu Jun 17 11:04:34 2010
New Revision: 955553
URL: http://svn.apache.org/viewvc?rev=955553&view=rev
Log:
CAMEL-2723: RedeliveryErrorHandler now supports async routing engine (work in progress). Fixed test and wiretap.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncFailureProcessorWithRedeliveryTest.java (contents, props changed)
- copied, changed from r955500, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointFailedAfterTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncOnExceptionFailureProcessorWithRedeliveryTest.java (with props)
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierFailureHandledEventsTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncEndpoint.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncProducer.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java?rev=955553&r1=955552&r2=955553&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java Thu Jun 17 11:04:34 2010
@@ -16,6 +16,8 @@
*/
package org.apache.camel.processor;
+import java.util.concurrent.RejectedExecutionException;
+
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
import org.apache.camel.Exchange;
@@ -104,6 +106,15 @@ public abstract class RedeliveryErrorHan
protected boolean processErrorHandler(final Exchange exchange, final AsyncCallback callback, final RedeliveryData data) {
while (true) {
+ // can we still run
+ if (!isRunAllowed()) {
+ if (exchange.getException() == null) {
+ exchange.setException(new RejectedExecutionException());
+ }
+ callback.done(data.sync);
+ return data.sync;
+ }
+
// did previous processing cause an exception?
boolean handle = shouldHandleException(exchange);
if (handle) {
@@ -154,13 +165,12 @@ public abstract class RedeliveryErrorHan
data.sync = false;
// only process if the exchange hasn't failed
// and it has not been handled by the error processor
- if (!isDone(exchange)) {
- // TODO: async process redelivery (eg duplicate the error handler logic)
- // And have a timer task scheduled when redelivery should occur to avoid blocking thread
- log.debug("Not done continuing error handling asynchronously: " + exchange);
- } else {
+ if (isDone(exchange)) {
callback.done(sync);
+ return;
}
+ // error occurred so loop back around which we do by invoking the processAsyncErrorHandler
+ processAsyncErrorHandler(exchange, callback, data);
}
});
if (!sync) {
@@ -168,6 +178,7 @@ public abstract class RedeliveryErrorHan
return false;
}
+ // we route synchronously
boolean done = isDone(exchange);
if (done) {
callback.done(true);
@@ -178,6 +189,78 @@ public abstract class RedeliveryErrorHan
}
+ protected void processAsyncErrorHandler(final Exchange exchange, final AsyncCallback callback, final RedeliveryData data) {
+ // can we still run
+ if (!isRunAllowed()) {
+ if (exchange.getException() == null) {
+ exchange.setException(new RejectedExecutionException());
+ }
+ callback.done(data.sync);
+ return;
+ }
+
+ // did previous processing cause an exception?
+ boolean handle = shouldHandleException(exchange);
+ if (handle) {
+ handleException(exchange, data);
+ }
+
+ // compute if we should redeliver or not
+ boolean shouldRedeliver = shouldRedeliver(exchange, data);
+ if (!shouldRedeliver) {
+ // no we should not redeliver to the same output so either try an onException (if any given)
+ // or the dead letter queue
+ Processor target = data.failureProcessor != null ? data.failureProcessor : data.deadLetterProcessor;
+ // deliver to the failure processor (either an on exception or dead letter queue
+ deliverToFailureProcessor(target, exchange, data, callback);
+ // we are breaking out
+ return;
+ }
+
+ // TODO: Use a scheduler to schedule the redelivery delay
+ // which contains the outputAsync task being executed
+ // we can optimize and only use the scheduler if there is a delay
+ if (shouldRedeliver && data.redeliveryCounter > 0) {
+ // prepare for redelivery
+ prepareExchangeForRedelivery(exchange);
+
+ // if we are redelivering then sleep before trying again
+ // wait until we should redeliver
+ try {
+ data.redeliveryDelay = data.currentRedeliveryPolicy.sleep(data.redeliveryDelay, data.redeliveryCounter);
+ } catch (InterruptedException e) {
+ if (log.isDebugEnabled()) {
+ log.debug("Sleep interrupted, are we stopping? " + (isStopping() || isStopped()));
+ }
+ return;
+ }
+
+ // letting onRedeliver be executed
+ deliverToRedeliveryProcessor(exchange, data);
+ }
+
+ // process the exchange (also redelivery)
+ outputAsync.process(exchange, new AsyncCallback() {
+ public void done(boolean sync) {
+ // this callback should only handle the async case
+ if (sync) {
+ return;
+ }
+
+ // mark we are in async mode now
+ data.sync = false;
+ // only process if the exchange hasn't failed
+ // and it has not been handled by the error processor
+ if (isDone(exchange)) {
+ callback.done(sync);
+ return;
+ }
+ // error occurred so loop back around which we do by invoking the processAsyncErrorHandler
+ processAsyncErrorHandler(exchange, callback, data);
+ }
+ });
+ }
+
/**
* Strategy whether the exchange has an exception that we should try to handle.
* <p/>
@@ -392,8 +475,7 @@ public abstract class RedeliveryErrorHan
// no processor but we need to prepare after failure as well
prepareExchangeAfterFailure(exchange, data);
} finally {
- // indicate we are done synchronously
- data.sync = true;
+ // callback we are done
callback.done(data.sync);
}
}
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=955553&r1=955552&r2=955553&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 Jun 17 11:04:34 2010
@@ -67,17 +67,19 @@ public class WireTapProcessor extends Se
return "wireTap(" + destination.getEndpointUri() + ")";
}
- public void process(final Exchange exchange) throws Exception {
+ 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, exchange, pattern, new ProducerCallback<Exchange>() {
+ return producerCache.doInProducer(destination, wireTapExchange, pattern, new ProducerCallback<Exchange>() {
public Exchange doInProducer(Producer producer, Exchange exchange, ExchangePattern pattern) throws Exception {
- exchange = configureExchange(exchange, pattern);
if (log.isDebugEnabled()) {
log.debug(">>>> (wiretap) " + destination + " " + exchange);
}
@@ -89,17 +91,19 @@ public class WireTapProcessor extends Se
});
}
- public boolean process(final Exchange exchange, final AsyncCallback callback) {
+ public boolean process(Exchange exchange, final AsyncCallback callback) {
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, exchange, pattern, new ProducerCallback<Exchange>() {
+ return producerCache.doInProducer(destination, wireTapExchange, pattern, new ProducerCallback<Exchange>() {
public Exchange doInProducer(Producer producer, Exchange exchange, ExchangePattern pattern) throws Exception {
- exchange = configureExchange(exchange, pattern);
if (log.isDebugEnabled()) {
log.debug(">>>> (wiretap) " + destination + " " + exchange);
}
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierFailureHandledEventsTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierFailureHandledEventsTest.java?rev=955553&r1=955552&r2=955553&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierFailureHandledEventsTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierFailureHandledEventsTest.java Thu Jun 17 11:04:34 2010
@@ -99,15 +99,16 @@ public class EventNotifierFailureHandled
assertIsInstanceOf(RouteStartedEvent.class, events.get(1));
assertIsInstanceOf(CamelContextStartedEvent.class, events.get(2));
assertIsInstanceOf(ExchangeCreatedEvent.class, events.get(3));
- assertIsInstanceOf(ExchangeSentEvent.class, events.get(4));
- ExchangeFailureHandledEvent e = assertIsInstanceOf(ExchangeFailureHandledEvent.class, events.get(5));
+ ExchangeFailureHandledEvent e = assertIsInstanceOf(ExchangeFailureHandledEvent.class, events.get(4));
assertEquals("should be DLC", true, e.isDeadLetterChannel());
SendProcessor send = assertIsInstanceOf(SendProcessor.class, e.getFailureHandler());
assertEquals("mock://dead", send.getDestination().getEndpointUri());
// dead letter channel will mark the exchange as completed
- assertIsInstanceOf(ExchangeCompletedEvent.class, events.get(6));
+ assertIsInstanceOf(ExchangeCompletedEvent.class, events.get(5));
+ // and the sent will be logged after they are complete sending as it record the time taken as well
+ assertIsInstanceOf(ExchangeSentEvent.class, events.get(6));
assertIsInstanceOf(ExchangeSentEvent.class, events.get(7));
}
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncFailureProcessorWithRedeliveryTest.java (from r955500, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointFailedAfterTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncFailureProcessorWithRedeliveryTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncFailureProcessorWithRedeliveryTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointFailedAfterTest.java&r1=955500&r2=955553&rev=955553&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointFailedAfterTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncFailureProcessorWithRedeliveryTest.java Thu Jun 17 11:04:34 2010
@@ -16,7 +16,6 @@
*/
package org.apache.camel.processor.async;
-import org.apache.camel.CamelExecutionException;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -25,7 +24,7 @@ import org.apache.camel.builder.RouteBui
/**
* @version $Revision$
*/
-public class AsyncEndpointFailedAfterTest extends ContextTestSupport {
+public class AsyncFailureProcessorWithRedeliveryTest extends ContextTestSupport {
private static String beforeThreadName;
private static String afterThreadName;
@@ -33,15 +32,10 @@ public class AsyncEndpointFailedAfterTes
public void testAsyncEndpoint() throws Exception {
getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel");
- getMockEndpoint("mock:result").expectedMessageCount(0);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye Camel");
- try {
- template.requestBody("direct:start", "Hello Camel", String.class);
- fail("Should have thrown an exception");
- } catch (CamelExecutionException e) {
- assertIsInstanceOf(IllegalArgumentException.class, e.getCause());
- assertEquals("Damn", e.getCause().getMessage());
- }
+ String reply = template.requestBody("direct:start", "Hello Camel", String.class);
+ assertEquals("Bye Camel", reply);
assertMockEndpointsSatisfied();
@@ -55,6 +49,9 @@ public class AsyncEndpointFailedAfterTes
public void configure() throws Exception {
context.addComponent("async", new MyAsyncComponent());
+ // use redelivery up till 5 times
+ errorHandler(defaultErrorHandler().maximumRedeliveries(5));
+
from("direct:start")
.to("mock:before")
.to("log:before")
@@ -63,7 +60,9 @@ public class AsyncEndpointFailedAfterTes
beforeThreadName = Thread.currentThread().getName();
}
})
- .to("async:Bye Camel")
+ // invoking the async endpoint could also cause a failure so
+ // test that we can do redelivery
+ .to("async:Bye Camel?failFirstAttempts=2")
.process(new Processor() {
public void process(Exchange exchange) throws Exception {
afterThreadName = Thread.currentThread().getName();
@@ -71,7 +70,6 @@ public class AsyncEndpointFailedAfterTes
})
.to("log:after")
.to("mock:after")
- .throwException(new IllegalArgumentException("Damn"))
.to("mock:result");
}
};
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncFailureProcessorWithRedeliveryTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncFailureProcessorWithRedeliveryTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncOnExceptionFailureProcessorWithRedeliveryTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncOnExceptionFailureProcessorWithRedeliveryTest.java?rev=955553&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncOnExceptionFailureProcessorWithRedeliveryTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncOnExceptionFailureProcessorWithRedeliveryTest.java Thu Jun 17 11:04:34 2010
@@ -0,0 +1,77 @@
+/**
+ * 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.async;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+
+/**
+ * @version $Revision$
+ */
+public class AsyncOnExceptionFailureProcessorWithRedeliveryTest extends ContextTestSupport {
+
+ private static String beforeThreadName;
+ private static String afterThreadName;
+
+ public void testAsyncEndpoint() throws Exception {
+ getMockEndpoint("mock:error").expectedBodiesReceived("Bye Camel");
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+
+ String reply = template.requestBody("direct:start", "Hello Camel", String.class);
+ assertEquals("Bye Camel", reply);
+
+ assertMockEndpointsSatisfied();
+
+ assertFalse("Should use different threads", beforeThreadName.equalsIgnoreCase(afterThreadName));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ // use redelivery up till 5 times
+ errorHandler(defaultErrorHandler().maximumRedeliveries(5));
+
+ onException(Exception.class).handled(true)
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ beforeThreadName = Thread.currentThread().getName();
+ }
+ })
+ // invoking the async endpoint could also cause a failure so
+ // test that we can do redelivery
+ .to("async:Bye Camel?failFirstAttempts=2")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ afterThreadName = Thread.currentThread().getName();
+ }
+ })
+ .to("mock:error");
+
+ from("direct:start")
+ .throwException(new IllegalArgumentException("Damn"))
+ .to("mock:result");
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncOnExceptionFailureProcessorWithRedeliveryTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncOnExceptionFailureProcessorWithRedeliveryTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncEndpoint.java?rev=955553&r1=955552&r2=955553&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncEndpoint.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncEndpoint.java Thu Jun 17 11:04:34 2010
@@ -29,6 +29,7 @@ public class MyAsyncEndpoint extends Def
private String reply;
private long delay = 500;
+ private int failFirstAttempts;
public MyAsyncEndpoint(String endpointUri, Component component) {
super(endpointUri, component);
@@ -61,4 +62,12 @@ public class MyAsyncEndpoint extends Def
public void setDelay(long delay) {
this.delay = delay;
}
+
+ public int getFailFirstAttempts() {
+ return failFirstAttempts;
+ }
+
+ public void setFailFirstAttempts(int failFirstAttempts) {
+ this.failFirstAttempts = failFirstAttempts;
+ }
}
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncProducer.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncProducer.java?rev=955553&r1=955552&r2=955553&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncProducer.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/MyAsyncProducer.java Thu Jun 17 11:04:34 2010
@@ -19,9 +19,11 @@ package org.apache.camel.processor.async
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
+import org.apache.camel.CamelExchangeException;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Producer;
@@ -38,6 +40,7 @@ public class MyAsyncProducer implements
private static final Log LOG = LogFactory.getLog(MyAsyncProducer.class);
private final ExecutorService executor = Executors.newSingleThreadExecutor();
private final MyAsyncEndpoint endpoint;
+ private final AtomicInteger counter = new AtomicInteger();
public MyAsyncProducer(MyAsyncEndpoint endpoint) {
this.endpoint = endpoint;
@@ -51,10 +54,17 @@ public class MyAsyncProducer implements
executor.submit(new Callable<Object>() {
public Object call() throws Exception {
LOG.info("Simulating a task which takes " + endpoint.getDelay() + " millis to reply");
-
Thread.sleep(endpoint.getDelay());
- String reply = endpoint.getReply();
- exchange.getOut().setBody(reply);
+
+ int count = counter.incrementAndGet();
+ if (endpoint.getFailFirstAttempts() >= count) {
+ LOG.info("Simulating a failure at attempt " + count);
+ exchange.setException(new CamelExchangeException("Simulated error at attempt " + count, exchange));
+ } else {
+ String reply = endpoint.getReply();
+ exchange.getOut().setBody(reply);
+ LOG.info("Setting reply " + reply);
+ }
LOG.info("Callback done(false)");
callback.done(false);