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 2011/08/19 18:53:52 UTC
svn commit: r1159682 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/impl/ test/java/org/apache/camel/management/
test/java/org/apache/camel/processor/async/
Author: davsclaus
Date: Fri Aug 19 16:53:51 2011
New Revision: 1159682
URL: http://svn.apache.org/viewvc?rev=1159682&view=rev
Log:
CAMEL-4354: Fixed issue with ExchangeSentEvent being emitted to soon when using async routing engine.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEventNotifierTest.java (contents, props changed)
- copied, changed from r1159650, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEnricherTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java
camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierFailureHandledEventsTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierRedeliveryEventsTest.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java?rev=1159682&r1=1159681&r2=1159682&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java Fri Aug 19 16:53:51 2011
@@ -259,11 +259,12 @@ public class ProducerCache extends Servi
* @param producerCallback the producer template callback to be executed
* @return (doneSync) <tt>true</tt> to continue execute synchronously, <tt>false</tt> to continue being executed asynchronously
*/
- public boolean doInAsyncProducer(Endpoint endpoint, Exchange exchange, ExchangePattern pattern, AsyncCallback callback, AsyncProducerCallback producerCallback) {
+ public boolean doInAsyncProducer(final Endpoint endpoint, final Exchange exchange, final ExchangePattern pattern,
+ final AsyncCallback callback, final AsyncProducerCallback producerCallback) {
boolean sync = true;
// get the producer and we do not mind if its pooled as we can handle returning it back to the pool
- Producer producer = doGetProducer(endpoint, true);
+ final Producer producer = doGetProducer(endpoint, true);
if (producer == null) {
if (isStopped()) {
@@ -274,39 +275,44 @@ public class ProducerCache extends Servi
}
}
- StopWatch watch = null;
- if (exchange != null) {
- // record timing for sending the exchange using the producer
- watch = new StopWatch();
- }
+ // record timing for sending the exchange using the producer
+ final StopWatch watch = exchange != null ? new StopWatch() : null;
try {
// invoke the callback
AsyncProcessor asyncProcessor = AsyncProcessorTypeConverter.convert(producer);
- sync = producerCallback.doInAsyncProducer(producer, asyncProcessor, exchange, pattern, callback);
+ sync = producerCallback.doInAsyncProducer(producer, asyncProcessor, exchange, pattern, new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ try {
+ if (watch != null) {
+ long timeTaken = watch.stop();
+ // emit event that the exchange was sent to the endpoint
+ EventHelper.notifyExchangeSent(exchange.getContext(), exchange, endpoint, timeTaken);
+ }
+
+ if (producer instanceof ServicePoolAware) {
+ // release back to the pool
+ pool.release(endpoint, producer);
+ } else if (!producer.isSingleton()) {
+ // stop non singleton producers as we should not leak resources
+ try {
+ ServiceHelper.stopService(producer);
+ } catch (Exception e) {
+ // ignore and continue
+ LOG.warn("Error stopping producer: " + producer, e);
+ }
+ }
+ } finally {
+ callback.done(doneSync);
+ }
+ }
+ });
} catch (Throwable e) {
// ensure exceptions is caught and set on the exchange
if (exchange != null) {
exchange.setException(e);
}
- } finally {
- if (exchange != null && exchange.getException() == null) {
- long timeTaken = watch.stop();
- // emit event that the exchange was sent to the endpoint
- EventHelper.notifyExchangeSent(exchange.getContext(), exchange, endpoint, timeTaken);
- }
- if (producer instanceof ServicePoolAware) {
- // release back to the pool
- pool.release(endpoint, producer);
- } else if (!producer.isSingleton()) {
- // stop non singleton producers as we should not leak resources
- try {
- ServiceHelper.stopService(producer);
- } catch (Exception e) {
- // ignore and continue
- LOG.warn("Error stopping producer: " + producer, e);
- }
- }
}
return sync;
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=1159682&r1=1159681&r2=1159682&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 Fri Aug 19 16:53:51 2011
@@ -94,17 +94,19 @@ 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(4));
+ ExchangeFailureHandledEvent e = assertIsInstanceOf(ExchangeFailureHandledEvent.class, events.get(5));
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(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(ExchangeCompletedEvent.class, events.get(6));
+ // and the last event should be the direct:start
assertIsInstanceOf(ExchangeSentEvent.class, events.get(7));
+ ExchangeSentEvent sent = (ExchangeSentEvent) events.get(7);
+ assertEquals("direct://start", sent.getEndpoint().getEndpointUri());
}
public void testExchangeOnException() throws Exception {
@@ -127,14 +129,17 @@ 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(4));
+ ExchangeFailureHandledEvent e = assertIsInstanceOf(ExchangeFailureHandledEvent.class, events.get(5));
assertEquals("should NOT be DLC", false, e.isDeadLetterChannel());
// onException will handle the exception
- assertIsInstanceOf(ExchangeCompletedEvent.class, events.get(5));
- assertIsInstanceOf(ExchangeSentEvent.class, events.get(6));
+ assertIsInstanceOf(ExchangeCompletedEvent.class, events.get(6));
+ // and the last event should be the direct:start
assertIsInstanceOf(ExchangeSentEvent.class, events.get(7));
+ ExchangeSentEvent sent = (ExchangeSentEvent) events.get(7);
+ assertEquals("direct://start", sent.getEndpoint().getEndpointUri());
}
}
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierRedeliveryEventsTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierRedeliveryEventsTest.java?rev=1159682&r1=1159681&r2=1159682&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierRedeliveryEventsTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierRedeliveryEventsTest.java Fri Aug 19 16:53:51 2011
@@ -100,9 +100,9 @@ public class EventNotifierRedeliveryEven
assertEquals(3, e.getAttempt());
e = assertIsInstanceOf(ExchangeRedeliveryEvent.class, events.get(4));
assertEquals(4, e.getAttempt());
- assertIsInstanceOf(ExchangeFailureHandledEvent.class, events.get(5));
- assertIsInstanceOf(ExchangeCompletedEvent.class, events.get(6));
- assertIsInstanceOf(ExchangeSentEvent.class, events.get(7));
+ assertIsInstanceOf(ExchangeSentEvent.class, events.get(5));
+ assertIsInstanceOf(ExchangeFailureHandledEvent.class, events.get(6));
+ assertIsInstanceOf(ExchangeCompletedEvent.class, events.get(7));
assertIsInstanceOf(ExchangeSentEvent.class, events.get(8));
}
@@ -134,7 +134,6 @@ public class EventNotifierRedeliveryEven
assertEquals(3, e.getAttempt());
e = assertIsInstanceOf(ExchangeRedeliveryEvent.class, events.get(4));
assertEquals(4, e.getAttempt());
- assertIsInstanceOf(ExchangeFailureHandledEvent.class, events.get(5));
// since its async the ordering of the rest can be different depending per OS and timing
}
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEventNotifierTest.java (from r1159650, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEnricherTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEventNotifierTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEventNotifierTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEnricherTest.java&r1=1159650&r2=1159682&rev=1159682&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEnricherTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEventNotifierTest.java Fri Aug 19 16:53:51 2011
@@ -16,22 +16,28 @@
*/
package org.apache.camel.processor.async;
+import java.util.EventObject;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.camel.CamelContext;
import org.apache.camel.ContextTestSupport;
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.management.EventNotifierSupport;
+import org.apache.camel.management.event.ExchangeSentEvent;
/**
* @version
*/
-public class AsyncEndpointEnricherTest extends ContextTestSupport {
+public class AsyncEndpointEventNotifierTest extends ContextTestSupport {
- private static String beforeThreadName;
- private static String afterThreadName;
+ private final CountDownLatch latch = new CountDownLatch(1);
+ private final AtomicLong time = new AtomicLong();
- public void testAsyncEndpoint() throws Exception {
+ public void testAsyncEndpointEventNotifer() throws Exception {
getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
- getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel");
getMockEndpoint("mock:result").expectedBodiesReceived("Bye Camel");
String reply = template.requestBody("direct:start", "Hello Camel", String.class);
@@ -39,7 +45,43 @@ public class AsyncEndpointEnricherTest e
assertMockEndpointsSatisfied();
- assertFalse("Should use different threads", beforeThreadName.equalsIgnoreCase(afterThreadName));
+ assertTrue("Should count down", latch.await(10, TimeUnit.SECONDS));
+
+ long delta = time.get();
+ assertTrue("Should take about 1000 millis sec, was: " + delta, delta > 800);
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ DefaultCamelContext context = new DefaultCamelContext(createRegistry());
+ context.getManagementStrategy().addEventNotifier(new EventNotifierSupport() {
+ public void notify(EventObject event) throws Exception {
+ try {
+ ExchangeSentEvent sent = (ExchangeSentEvent) event;
+ time.set(sent.getTimeTaken());
+ } finally {
+ latch.countDown();
+ }
+ }
+
+ public boolean isEnabled(EventObject event) {
+ // we only want the async endpoint
+ if (event instanceof ExchangeSentEvent) {
+ ExchangeSentEvent sent = (ExchangeSentEvent) event;
+ return sent.getEndpoint().getEndpointUri().startsWith("async");
+ }
+ return false;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ }
+ });
+ return context;
}
@Override
@@ -51,20 +93,7 @@ public class AsyncEndpointEnricherTest e
from("direct:start")
.to("mock:before")
- .to("log:before")
- .process(new Processor() {
- public void process(Exchange exchange) throws Exception {
- beforeThreadName = Thread.currentThread().getName();
- }
- })
- .enrich("async:Bye Camel")
- .process(new Processor() {
- public void process(Exchange exchange) throws Exception {
- afterThreadName = Thread.currentThread().getName();
- }
- })
- .to("log:after")
- .to("mock:after")
+ .to("async:Bye Camel?delay=1000")
.to("mock:result");
}
};
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEventNotifierTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointEventNotifierTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date