Added unit test based on user forum issue. Make sure bridgeErrorHandler runs in UoW.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/36d5544d Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/36d5544d Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/36d5544d Branch: refs/heads/camel-2.13.x Commit: 36d5544d1ff65c1dbea3c1f5087516ef879ccabf Parents: 398f949 Author: Claus Ibsen <davscl...@apache.org> Authored: Fri Apr 25 07:49:27 2014 +0200 Committer: Claus Ibsen <davscl...@apache.org> Committed: Fri Apr 25 08:23:59 2014 +0200 ---------------------------------------------------------------------- .../BridgeExceptionHandlerToErrorHandler.java | 10 +- ...nsumerBridgeErrorHandlerOnExceptionTest.java | 164 +++++++++++++++++++ 2 files changed, 172 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/36d5544d/camel-core/src/main/java/org/apache/camel/impl/BridgeExceptionHandlerToErrorHandler.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/impl/BridgeExceptionHandlerToErrorHandler.java b/camel-core/src/main/java/org/apache/camel/impl/BridgeExceptionHandlerToErrorHandler.java index 6292189..b6bbadc 100644 --- a/camel-core/src/main/java/org/apache/camel/impl/BridgeExceptionHandlerToErrorHandler.java +++ b/camel-core/src/main/java/org/apache/camel/impl/BridgeExceptionHandlerToErrorHandler.java @@ -16,10 +16,11 @@ */ package org.apache.camel.impl; -import org.apache.camel.Consumer; import org.apache.camel.Exchange; import org.apache.camel.Processor; import org.apache.camel.spi.ExceptionHandler; +import org.apache.camel.spi.UnitOfWork; +import org.apache.camel.util.UnitOfWorkHelper; /** * An {@link ExceptionHandler} that uses the {@link DefaultConsumer} to @@ -39,7 +40,7 @@ import org.apache.camel.spi.ExceptionHandler; public class BridgeExceptionHandlerToErrorHandler implements ExceptionHandler { private final LoggingExceptionHandler fallback; - private final Consumer consumer; + private final DefaultConsumer consumer; private final Processor bridge; public BridgeExceptionHandlerToErrorHandler(DefaultConsumer consumer) { @@ -71,10 +72,15 @@ public class BridgeExceptionHandlerToErrorHandler implements ExceptionHandler { // and mark as redelivery exhausted as we cannot do redeliveries exchange.setProperty(Exchange.REDELIVERY_EXHAUSTED, Boolean.TRUE); + // wrap in UoW + UnitOfWork uow = null; try { + uow = consumer.createUoW(exchange); bridge.process(exchange); } catch (Exception e) { fallback.handleException("Error handling exception " + exception.getMessage(), exchange, e); + } finally { + UnitOfWorkHelper.doneUow(uow, exchange); } } } http://git-wip-us.apache.org/repos/asf/camel/blob/36d5544d/camel-core/src/test/java/org/apache/camel/processor/DefaultConsumerBridgeErrorHandlerOnExceptionTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/processor/DefaultConsumerBridgeErrorHandlerOnExceptionTest.java b/camel-core/src/test/java/org/apache/camel/processor/DefaultConsumerBridgeErrorHandlerOnExceptionTest.java new file mode 100644 index 0000000..b2fc321 --- /dev/null +++ b/camel-core/src/test/java/org/apache/camel/processor/DefaultConsumerBridgeErrorHandlerOnExceptionTest.java @@ -0,0 +1,164 @@ +/** + * 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 java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.Component; +import org.apache.camel.Consumer; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Endpoint; +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.DefaultComponent; +import org.apache.camel.impl.DefaultConsumer; +import org.apache.camel.impl.DefaultEndpoint; + +/** + * + */ +public class DefaultConsumerBridgeErrorHandlerOnExceptionTest extends ContextTestSupport { + + protected final CountDownLatch latch = new CountDownLatch(1); + + public void testDefaultConsumerBridgeErrorHandler() throws Exception { + getMockEndpoint("mock:result").expectedBodiesReceived("Hello World", "Hello World"); + getMockEndpoint("mock:a").expectedBodiesReceived("Cannot process"); + getMockEndpoint("mock:b").expectedBodiesReceived("Cannot process"); + getMockEndpoint("mock:dead").expectedBodiesReceived("Cannot process"); + + latch.countDown(); + + assertMockEndpointsSatisfied(); + + Exception cause = getMockEndpoint("mock:dead").getReceivedExchanges().get(0).getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class); + assertNotNull(cause); + assertEquals("Simulated", cause.getMessage()); + } + + @Override + protected RouteBuilder createRouteBuilder() throws Exception { + // START SNIPPET: e1 + return new RouteBuilder() { + @Override + public void configure() throws Exception { + // register our custom component + getContext().addComponent("my", new MyComponent()); + + // configure on exception + onException(Exception.class) + .handled(true) + .to("mock:a") + .to("direct:error") + .to("mock:dead"); + + // configure the consumer to bridge with the Camel error handler, + // so the above error handler will trigger if exceptions also + // occurs inside the consumer + from("my:foo?consumer.bridgeErrorHandler=true") + .to("log:foo") + .to("mock:result"); + + from("direct:error") + .to("mock:b") + .log("Error happened due ${exception.message}"); + } + }; + // END SNIPPET: e1 + } + + public class MyComponent extends DefaultComponent { + + @Override + protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception { + return new MyEndpoint(uri, this); + } + } + + public class MyEndpoint extends DefaultEndpoint { + + public MyEndpoint(String endpointUri, Component component) { + super(endpointUri, component); + } + + @Override + public Producer createProducer() throws Exception { + return null; + } + + @Override + public Consumer createConsumer(Processor processor) throws Exception { + Consumer answer = new MyConsumer(this, processor); + configureConsumer(answer); + return answer; + } + + @Override + public boolean isSingleton() { + return true; + } + } + + public class MyConsumer extends DefaultConsumer { + + private int invoked; + + public MyConsumer(Endpoint endpoint, Processor processor) { + super(endpoint, processor); + } + + public void doSomething() throws Exception { + try { + if (invoked++ == 0) { + throw new IllegalArgumentException("Simulated"); + } + + Exchange exchange = getEndpoint().createExchange(); + exchange.getIn().setBody("Hello World"); + getProcessor().process(exchange); + + } catch (Exception e) { + getExceptionHandler().handleException("Cannot process", e); + } + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + + Thread thread = new Thread() { + @Override + public void run() { + try { + // do not start before the mocks has been setup and is ready + latch.await(5, TimeUnit.SECONDS); + doSomething(); + doSomething(); + doSomething(); + } catch (Exception e) { + // ignore + } + } + }; + thread.start(); + } + } +}
