Updated Branches: refs/heads/master cfe5ddd56 -> 5761250c7
Fix for https://issues.apache.org/jira/browse/CAMEL-6918 - Error handler for SEDA producer doesn't work - added exception handling in the SendProducer + tests Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/5761250c Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/5761250c Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/5761250c Branch: refs/heads/master Commit: 5761250c74e02f02b51ad4c0a9a21a9d0ac4f511 Parents: cfe5ddd Author: Christian Posta <christian.po...@gmail.com> Authored: Mon Nov 4 15:25:28 2013 -0700 Committer: Christian Posta <christian.po...@gmail.com> Committed: Mon Nov 4 15:25:28 2013 -0700 ---------------------------------------------------------------------- .../apache/camel/processor/SendProcessor.java | 35 +++++++----- .../camel/component/seda/SedaErrorTest.java | 56 ++++++++++++++++++++ 2 files changed, 79 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/5761250c/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java index b7d3c3c..6ba6fc8 100644 --- a/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java +++ b/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java @@ -107,20 +107,31 @@ public class SendProcessor extends ServiceSupport implements AsyncProcessor, Tra EventHelper.notifyExchangeSending(exchange.getContext(), target, destination); LOG.debug(">>>> {} {}", destination, exchange); - return producer.process(exchange, new AsyncCallback() { - @Override - public void done(boolean doneSync) { - try { - // restore previous MEP - target.setPattern(existingPattern); - // emit event that the exchange was sent to the endpoint - long timeTaken = watch.stop(); - EventHelper.notifyExchangeSent(target.getContext(), target, destination, timeTaken); - } finally { - callback.done(doneSync); + + boolean sync = true; + try { + sync = producer.process(exchange, new AsyncCallback() { + @Override + public void done(boolean doneSync) { + try { + // restore previous MEP + target.setPattern(existingPattern); + // emit event that the exchange was sent to the endpoint + long timeTaken = watch.stop(); + EventHelper.notifyExchangeSent(target.getContext(), target, destination, timeTaken); + } finally { + callback.done(doneSync); + } } + }); + } catch (Throwable throwable) { + if (exchange != null) { + exchange.setException(throwable); } - }); + + } + + return sync; } // send the exchange to the destination using the producer cache for the non optimized producers http://git-wip-us.apache.org/repos/asf/camel/blob/5761250c/camel-core/src/test/java/org/apache/camel/component/seda/SedaErrorTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/component/seda/SedaErrorTest.java b/camel-core/src/test/java/org/apache/camel/component/seda/SedaErrorTest.java new file mode 100644 index 0000000..992d79c --- /dev/null +++ b/camel-core/src/test/java/org/apache/camel/component/seda/SedaErrorTest.java @@ -0,0 +1,56 @@ +/** + * 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.component.seda; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.builder.ExchangeBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.Test; + + +public class SedaErrorTest extends ContextTestSupport { + + @Test + public void testErrorHandle() throws InterruptedException { + MockEndpoint mockDLC = getMockEndpoint("mock:dlc"); + mockDLC.expectedMessageCount(1); + + try { + for (int i = 0; i < 3; i++) { + template.send("direct:start", ExchangeBuilder.anExchange(context).withBody("msg" + i).build()); + } + } catch (Exception ex) { + // noop + } + + assertMockEndpointsSatisfied(); + } + + + @Override + protected RouteBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + errorHandler(deadLetterChannel("mock:dlc")); + from("direct:start").log("start: ${body}").to("seda:seda1?size=2&blockWhenFull=false").log("after: ${body}"); + } + }; + + } +} \ No newline at end of file
