Author: gertv
Date: Sun Jun 27 11:50:24 2010
New Revision: 958366
URL: http://svn.apache.org/viewvc?rev=958366&view=rev
Log:
SMXCOMP-770: Improve code to leverage async camel routing
Added:
servicemix/components/engines/servicemix-camel/trunk/src/test/java/org/apache/servicemix/camel/AsyncJbiMessagingTest.java
Modified:
servicemix/components/engines/servicemix-camel/trunk/pom.xml
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelConsumerEndpoint.java
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelJbiComponent.java
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelProviderEndpoint.java
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiBinding.java
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiComponent.java
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiEndpoint.java
servicemix/components/engines/servicemix-camel/trunk/src/test/java/org/apache/servicemix/camel/JbiTestSupport.java
Modified: servicemix/components/engines/servicemix-camel/trunk/pom.xml
URL:
http://svn.apache.org/viewvc/servicemix/components/engines/servicemix-camel/trunk/pom.xml?rev=958366&r1=958365&r2=958366&view=diff
==============================================================================
--- servicemix/components/engines/servicemix-camel/trunk/pom.xml (original)
+++ servicemix/components/engines/servicemix-camel/trunk/pom.xml Sun Jun 27
11:50:24 2010
@@ -37,7 +37,7 @@
</scm>
<properties>
- <camel-version>2.3.0</camel-version>
+ <camel-version>2.4-SNAPSHOT</camel-version>
<servicemix.osgi.import>
!org.apache.servicemix.camel*,
org.apache.camel.converter,
Modified:
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelConsumerEndpoint.java
URL:
http://svn.apache.org/viewvc/servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelConsumerEndpoint.java?rev=958366&r1=958365&r2=958366&view=diff
==============================================================================
---
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelConsumerEndpoint.java
(original)
+++
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelConsumerEndpoint.java
Sun Jun 27 11:50:24 2010
@@ -22,24 +22,34 @@ import javax.jbi.messaging.MessageExchan
import javax.jbi.messaging.MessagingException;
import javax.xml.namespace.QName;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
import org.apache.camel.Exchange;
import org.apache.servicemix.common.endpoints.ConsumerEndpoint;
import org.apache.servicemix.common.util.URIResolver;
import org.apache.servicemix.id.IdGenerator;
import org.apache.servicemix.jbi.exception.FaultException;
+import java.net.URISyntaxException;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+
/**
* A consumer endpoint that will be used to send JBI exchanges
* originating from camel.
*/
-public class CamelConsumerEndpoint extends ConsumerEndpoint {
+public class CamelConsumerEndpoint extends ConsumerEndpoint implements
AsyncProcessor {
public static final QName SERVICE_NAME = new
QName("http://camel.apache.org/schema/jbi";, "provider");
private JbiBinding binding;
- private JbiEndpoint jbiEndpoint;
-
+ private JbiEndpoint jbiEndpoint;
+
+ private Map<String, ContinuationData> continuations = new
ConcurrentHashMap<String, ContinuationData>();
+
public CamelConsumerEndpoint(JbiBinding binding, JbiEndpoint jbiEndpoint) {
setService(SERVICE_NAME);
setEndpoint(new IdGenerator().generateId());
@@ -47,20 +57,32 @@ public class CamelConsumerEndpoint exten
this.jbiEndpoint = jbiEndpoint;
}
- public void process(MessageExchange exchange) throws Exception {
- // we don't expect any asynchronous MessageExchange callbacks because
we're using sendSync
- logger.error("Unexpected MessageExchange received: " + exchange);
+ /**
+ * Process the JBI MessageExchange that is being delivered asynchronously
as a response to what was sent by
+ * the {...@link #process(Exchange, AsyncCallback)} method
+ */
+ public void process(final MessageExchange messageExchange) throws
Exception {
+ final ContinuationData data =
continuations.remove(messageExchange.getExchangeId());
+
+ if (data == null) {
+ logger.error("Unexpected MessageExchange received: " +
messageExchange);
+ } else {
+ binding.runWithCamelContextClassLoader(new Callable<Object>() {
+ public Object call() throws Exception {
+ processReponse(messageExchange, data.exchange);
+ data.callback.done(false);
+ return null;
+ }
+ });
+ }
}
+ /**
+ * Process the Camel Exchange by sending/receiving a JBI MessageExchange
synchronously
+ */
public void process(Exchange exchange) throws Exception {
try {
- MessageExchange messageExchange =
binding.makeJbiMessageExchange(exchange, getExchangeFactory(),
jbiEndpoint.getMep());
-
- if (jbiEndpoint.getOperation() != null) {
- messageExchange.setOperation(jbiEndpoint.getOperation());
- }
-
- URIResolver.configureExchange(messageExchange, getContext(),
jbiEndpoint.getDestinationUri());
+ MessageExchange messageExchange = prepareMessageExchange(exchange);
sendSync(messageExchange);
@@ -69,9 +91,52 @@ public class CamelConsumerEndpoint exten
} catch (MessagingException e) {
exchange.setException(e);
throw new JbiException(e);
- }
+ }
}
-
+
+ /**
+ * Process the Camel Exchange by sending a JBI MessageExchange
asynchronously, where the response
+ * will be handled by the {...@link
#process(javax.jbi.messaging.MessageExchange)} method
+ */
+ public boolean process(Exchange exchange, AsyncCallback asyncCallback) {
+ MessageExchange messageExchange = null;
+ try {
+ messageExchange = prepareMessageExchange(exchange);
+
+ continuations.put(messageExchange.getExchangeId(),
+ new ContinuationData(exchange, asyncCallback));
+
+ send(messageExchange);
+
+ return false;
+ } catch (Exception e) {
+ if (messageExchange != null) {
+ continuations.remove(messageExchange.getExchangeId());
+ }
+
+ exchange.setException(e);
+ asyncCallback.done(true);
+ return true;
+ }
+ }
+
+ /*
+ * Create and configure a JBI MessageExchange for a given Camel Exchange
+ */
+ private MessageExchange prepareMessageExchange(Exchange exchange) throws
MessagingException, URISyntaxException {
+ MessageExchange messageExchange =
binding.makeJbiMessageExchange(exchange, getExchangeFactory(),
jbiEndpoint.getMep());
+
+ if (jbiEndpoint.getOperation() != null) {
+ messageExchange.setOperation(jbiEndpoint.getOperation());
+ }
+
+ URIResolver.configureExchange(messageExchange, getContext(),
jbiEndpoint.getDestinationUri());
+ return messageExchange;
+ }
+
+ /*
+ * Process a JBI response message by updating the corresponding Camel
exchange.
+ */
private void processReponse(MessageExchange messageExchange, Exchange
exchange) throws MessagingException {
if (messageExchange.getStatus() == ExchangeStatus.ERROR) {
exchange.setException(messageExchange.getError());
@@ -95,4 +160,33 @@ public class CamelConsumerEndpoint exten
public void validate() throws DeploymentException {
// No validation required
}
+
+ /**
+ * Provides read-only access to the underlying map of continuation data
+ */
+ protected Map<String, ContinuationData> getContinuationData() {
+ return Collections.unmodifiableMap(continuations);
+ }
+
+ /**
+ * Access the underlying Camel {...@link org.apache.camel.Endpoint}
+ */
+ protected JbiEndpoint getJbiEndpoint() {
+ return jbiEndpoint;
+ }
+
+ /**
+ * Encapsulates all the data necessary to continue processing the Camel
Exchange
+ */
+ private static final class ContinuationData {
+
+ private final Exchange exchange;
+ private final AsyncCallback callback;
+
+ private ContinuationData(Exchange exchange, AsyncCallback callback) {
+ super();
+ this.exchange = exchange;
+ this.callback = callback;
+ }
+ }
}
Modified:
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelJbiComponent.java
URL:
http://svn.apache.org/viewvc/servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelJbiComponent.java?rev=958366&r1=958365&r2=958366&view=diff
==============================================================================
---
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelJbiComponent.java
(original)
+++
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelJbiComponent.java
Sun Jun 27 11:50:24 2010
@@ -24,6 +24,7 @@ import java.util.Map;
import javax.jbi.servicedesc.ServiceEndpoint;
+import org.apache.camel.AsyncProcessor;
import org.apache.camel.Endpoint;
import org.apache.camel.Processor;
import org.apache.servicemix.common.BaseServiceUnitManager;
@@ -116,7 +117,7 @@ public class CamelJbiComponent extends D
Map map = URISupport.parseQuery(uri.getQuery());
String camelUri = uri.getSchemeSpecificPart();
Endpoint camelEndpoint =
jbiComponent.getCamelContext().getEndpoint(camelUri);
- Processor processor =
jbiComponent.createCamelProcessor(camelEndpoint);
+ AsyncProcessor processor =
jbiComponent.createCamelProcessor(camelEndpoint);
CamelProviderEndpoint endpoint =
new CamelProviderEndpoint(getServiceUnit(), camelEndpoint,
jbiComponent.createBinding(camelEndpoint), processor);
Modified:
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelProviderEndpoint.java
URL:
http://svn.apache.org/viewvc/servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelProviderEndpoint.java?rev=958366&r1=958365&r2=958366&view=diff
==============================================================================
---
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelProviderEndpoint.java
(original)
+++
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/CamelProviderEndpoint.java
Sun Jun 27 11:50:24 2010
@@ -26,9 +26,7 @@ import javax.jbi.messaging.MessagingExce
import javax.jbi.messaging.RobustInOnly;
import javax.xml.namespace.QName;
-import org.apache.camel.Endpoint;
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
+import org.apache.camel.*;
import org.apache.camel.spi.Synchronization;
import org.apache.servicemix.common.JbiConstants;
@@ -49,15 +47,15 @@ public class CamelProviderEndpoint exten
private Endpoint camelEndpoint;
- private Processor camelProcessor;
+ private AsyncProcessor camelProcessor;
- public CamelProviderEndpoint(ServiceUnit serviceUnit, QName service,
String endpoint, JbiBinding binding, Processor camelProcessor) {
+ public CamelProviderEndpoint(ServiceUnit serviceUnit, QName service,
String endpoint, JbiBinding binding, AsyncProcessor camelProcessor) {
super(serviceUnit, service, endpoint);
this.camelProcessor = camelProcessor;
this.binding = binding;
}
- public CamelProviderEndpoint(ServiceUnit serviceUnit, Endpoint
camelEndpoint, JbiBinding binding, Processor camelProcessor) {
+ public CamelProviderEndpoint(ServiceUnit serviceUnit, Endpoint
camelEndpoint, JbiBinding binding, AsyncProcessor camelProcessor) {
this(serviceUnit, SERVICE_NAME, camelEndpoint.getEndpointUri(),
binding, camelProcessor);
}
@@ -105,13 +103,18 @@ public class CamelProviderEndpoint exten
final Exchange camelExchange = binding.createExchange(exchange);
camelExchange.setFromEndpoint(camelEndpoint);
camelExchange.addOnCompletion(this);
-
+
binding.runWithCamelContextClassLoader(new Callable<Object>() {
public Object call() throws Exception {
- camelProcessor.process(camelExchange);
+ camelProcessor.process(camelExchange, new AsyncCallback() {
+ public void done(boolean doneSync) {
+ // result processing done by onFailure/onSuccess
methods
+ }
+ });
return null;
}
});
+
} else {
// This is not complaint with the default MEPs
throw new IllegalStateException("Provider exchange is ACTIVE, but
no in or fault is provided");
Modified:
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiBinding.java
URL:
http://svn.apache.org/viewvc/servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiBinding.java?rev=958366&r1=958365&r2=958366&view=diff
==============================================================================
---
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiBinding.java
(original)
+++
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiBinding.java
Sun Jun 27 11:50:24 2010
@@ -103,7 +103,7 @@ public class JbiBinding {
* @param callable the block of code to be run
* @throws Exception exceptions being thrown while running the block of
code
*/
- public void runWithCamelContextClassLoader(Callable<Object> callable)
throws Exception {
+ public<T> T runWithCamelContextClassLoader(Callable<T> callable) throws
Exception {
ClassLoader original = Thread.currentThread().getContextClassLoader();
try {
ClassLoader loader = context.getApplicationContextClassLoader();
@@ -113,7 +113,7 @@ public class JbiBinding {
}
Thread.currentThread().setContextClassLoader(loader);
}
- callable.call();
+ return callable.call();
} finally {
// restore CL
Thread.currentThread().setContextClassLoader(original);
Modified:
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiComponent.java
URL:
http://svn.apache.org/viewvc/servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiComponent.java?rev=958366&r1=958365&r2=958366&view=diff
==============================================================================
---
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiComponent.java
(original)
+++
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiComponent.java
Sun Jun 27 11:50:24 2010
@@ -18,11 +18,7 @@ package org.apache.servicemix.camel;
import javax.xml.namespace.QName;
-import org.apache.camel.CamelContext;
-import org.apache.camel.Component;
-import org.apache.camel.Endpoint;
-import org.apache.camel.FailedToCreateProducerException;
-import org.apache.camel.Processor;
+import org.apache.camel.*;
import org.apache.camel.processor.UnitOfWorkProcessor;
import org.apache.servicemix.common.util.URIResolver;
import org.apache.servicemix.id.IdGenerator;
@@ -82,7 +78,7 @@ public class JbiComponent implements Com
protected CamelProviderEndpoint createJbiEndpointFromCamel(
- Endpoint camelEndpoint, Processor processor) {
+ Endpoint camelEndpoint, AsyncProcessor processor) {
CamelProviderEndpoint jbiEndpoint;
String endpointUri = camelEndpoint.getEndpointUri();
if (camelEndpoint instanceof JbiEndpoint) {
@@ -160,12 +156,12 @@ public class JbiComponent implements Com
*/
public CamelProviderEndpoint createJbiEndpointFromCamel(
Endpoint camelEndpoint) {
- Processor processor = createCamelProcessor(camelEndpoint);
+ AsyncProcessor processor = createCamelProcessor(camelEndpoint);
return createJbiEndpointFromCamel(camelEndpoint, processor);
}
- protected Processor createCamelProcessor(Endpoint camelEndpoint) {
- Processor processor = null;
+ protected AsyncProcessor createCamelProcessor(Endpoint camelEndpoint) {
+ AsyncProcessor processor = null;
try {
processor = new
UnitOfWorkProcessor(camelEndpoint.createProducer());
} catch (Exception e) {
Modified:
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiEndpoint.java
URL:
http://svn.apache.org/viewvc/servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiEndpoint.java?rev=958366&r1=958365&r2=958366&view=diff
==============================================================================
---
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiEndpoint.java
(original)
+++
servicemix/components/engines/servicemix-camel/trunk/src/main/java/org/apache/servicemix/camel/JbiEndpoint.java
Sun Jun 27 11:50:24 2010
@@ -22,17 +22,15 @@ import java.util.concurrent.Callable;
import javax.xml.namespace.QName;
-import org.apache.camel.Consumer;
-import org.apache.camel.Endpoint;
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.Producer;
+import org.apache.camel.*;
import org.apache.camel.impl.DefaultConsumer;
import org.apache.camel.impl.DefaultEndpoint;
import org.apache.camel.impl.DefaultProducer;
+import org.apache.camel.impl.converter.AsyncProcessorTypeConverter;
import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.spi.HeaderFilterStrategyAware;
import org.apache.camel.spi.Registry;
+import org.apache.camel.util.AsyncProcessorHelper;
import org.apache.camel.util.URISupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -81,7 +79,7 @@ public class JbiEndpoint extends Default
return new JbiProducer(this);
}
- protected class JbiProducer extends DefaultProducer {
+ protected class JbiProducer extends DefaultProducer implements
AsyncProcessor {
private final Log log = LogFactory.getLog(JbiProducer.class);
@@ -124,6 +122,20 @@ public class JbiEndpoint extends Default
protected CamelConsumerEndpoint getCamelConsumerEndpoint() {
return consumer;
}
+
+ public boolean process(final Exchange exchange, final AsyncCallback
asyncCallback) {
+ try {
+ return binding.runWithCamelContextClassLoader(new
Callable<Boolean>() {
+
+ public Boolean call() throws Exception {
+ return consumer.process(exchange, asyncCallback);
+ }
+ });
+ } catch (Exception e) {
+ exchange.setException(e);
+ return true;
+ }
+ }
}
@SuppressWarnings("unchecked")
@@ -206,7 +218,7 @@ public class JbiEndpoint extends Default
@Override
protected void doStart() throws Exception {
super.doStart();
- jbiEndpoint =
jbiComponent.createJbiEndpointFromCamel(JbiEndpoint.this, processor);
+ jbiEndpoint =
jbiComponent.createJbiEndpointFromCamel(JbiEndpoint.this,
AsyncProcessorTypeConverter.convert(processor));
jbiComponent.getCamelJbiComponent().activateJbiEndpoint(jbiEndpoint);
}
Added:
servicemix/components/engines/servicemix-camel/trunk/src/test/java/org/apache/servicemix/camel/AsyncJbiMessagingTest.java
URL:
http://svn.apache.org/viewvc/servicemix/components/engines/servicemix-camel/trunk/src/test/java/org/apache/servicemix/camel/AsyncJbiMessagingTest.java?rev=958366&view=auto
==============================================================================
---
servicemix/components/engines/servicemix-camel/trunk/src/test/java/org/apache/servicemix/camel/AsyncJbiMessagingTest.java
(added)
+++
servicemix/components/engines/servicemix-camel/trunk/src/test/java/org/apache/servicemix/camel/AsyncJbiMessagingTest.java
Sun Jun 27 11:50:24 2010
@@ -0,0 +1,100 @@
+/*
+ * 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.servicemix.camel;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.servicemix.client.DefaultServiceMixClient;
+import org.apache.servicemix.client.ServiceMixClient;
+import org.apache.servicemix.executors.impl.ExecutorFactoryImpl;
+import org.apache.servicemix.jbi.container.ActivationSpec;
+import org.apache.servicemix.jbi.container.JBIContainer;
+import org.apache.servicemix.jbi.jaxp.StringSource;
+
+import javax.jbi.messaging.ExchangeStatus;
+import javax.jbi.messaging.InOnly;
+import javax.xml.namespace.QName;
+import java.util.List;
+
+/**
+ * Tests to make sure that even a small thread pool can handle a lot of
servicemix-camel interactions
+ * by avoiding the use of sendSync internally.
+ */
+public class AsyncJbiMessagingTest extends JbiTestSupport {
+
+ private static final String MESSAGE = "<just><a>test</a></just>";
+
+ public void testNoSyncMessagingDeadlock() throws Exception {
+ ServiceMixClient client = new DefaultServiceMixClient(jbiContainer);
+ InOnly exchange = client.createInOnlyExchange();
+ exchange.setService(new QName("urn:test", "service"));
+ exchange.getInMessage().setContent(new StringSource(MESSAGE));
+ client.sendSync(exchange, 5000);
+ assertEquals("Should finish within the designated time-frame,
otherwise we probably caused a thread deadlock",
+ ExchangeStatus.DONE, exchange.getStatus());
+ }
+
+ public void testCallbackExchangesEmptyOnError() throws Exception {
+ // disable the exchange completed listener as it is unable to detect
failed exchanges
+ disableExchangeCompletedListener();
+
+ ServiceMixClient client = new DefaultServiceMixClient(jbiContainer);
+ InOnly exchange = client.createInOnlyExchange();
+ exchange.setService(new QName("urn:test", "error"));
+ exchange.getInMessage().setContent(new StringSource(MESSAGE));
+ client.sendSync(exchange);
+ assertEquals(ExchangeStatus.ERROR, exchange.getStatus());
+
+ // the tearDown method will now ensure that the ContinuationData map
on the endpoint is empty
+ // even though the send failed
+ }
+
+ @Override
+ protected void configureContainer(JBIContainer container) throws Exception
{
+ super.configureContainer(container);
+
+ // let's tune down the default thread pool size to make sure the test
fails with any JBI sync exchange
+ ExecutorFactoryImpl impl = new ExecutorFactoryImpl();
+ impl.getDefaultConfig().setCorePoolSize(1);
+
+ container.setExecutorFactory(impl);
+ }
+
+ @Override
+ protected void appendJbiActivationSpecs(List<ActivationSpec>
activationSpecList) {
+ // no additional activation specs required
+ }
+
+ @Override
+ protected RouteBuilder createRoutes() {
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() throws Exception {
+ // this chain of endpoints is too long to be handled with all
sync calls
+
from("jbi:service:urn:test:service").to("jbi:service:urn:test:service1");
+
from("jbi:service:urn:test:service1").to("jbi:service:urn:test:service2");
+
from("jbi:service:urn:test:service2").to("jbi:service:urn:test:service3");
+
from("jbi:service:urn:test:service3").to("jbi:service:urn:test:service4");
+
from("jbi:service:urn:test:service4").to("jbi:service:urn:test:service5");
+ from("jbi:service:urn:test:service5").to("log:test");
+
+ // deliberately send to an non-existing endpoint
+
from("jbi:service:urn:test:error").to("jbi:service:urn:test:non-existing");
+ }
+ };
+ }
+}
Modified:
servicemix/components/engines/servicemix-camel/trunk/src/test/java/org/apache/servicemix/camel/JbiTestSupport.java
URL:
http://svn.apache.org/viewvc/servicemix/components/engines/servicemix-camel/trunk/src/test/java/org/apache/servicemix/camel/JbiTestSupport.java?rev=958366&r1=958365&r2=958366&view=diff
==============================================================================
---
servicemix/components/engines/servicemix-camel/trunk/src/test/java/org/apache/servicemix/camel/JbiTestSupport.java
(original)
+++
servicemix/components/engines/servicemix-camel/trunk/src/test/java/org/apache/servicemix/camel/JbiTestSupport.java
Sun Jun 27 11:50:24 2010
@@ -20,8 +20,11 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.NotSerializableException;
import java.io.ObjectOutputStream;
+import java.lang.reflect.Field;
import java.util.ArrayList;
+import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -30,12 +33,7 @@ import javax.jbi.JBIException;
import javax.jbi.messaging.MessageExchange;
import javax.xml.namespace.QName;
-import org.apache.camel.CamelContext;
-import org.apache.camel.Endpoint;
-import org.apache.camel.Exchange;
-import org.apache.camel.Message;
-import org.apache.camel.Processor;
-import org.apache.camel.ProducerTemplate;
+import org.apache.camel.*;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.impl.DefaultCamelContext;
@@ -43,6 +41,7 @@ import org.apache.camel.spi.Synchronizat
import org.apache.camel.test.TestSupport;
import org.apache.servicemix.client.DefaultServiceMixClient;
import org.apache.servicemix.client.ServiceMixClient;
+import org.apache.servicemix.common.Registry;
import org.apache.servicemix.jbi.container.ActivationSpec;
import org.apache.servicemix.jbi.container.JBIContainer;
import org.apache.servicemix.jbi.container.SpringJBIContainer;
@@ -74,7 +73,7 @@ public abstract class JbiTestSupport ext
protected ProducerTemplate client;
protected ServiceMixClient servicemixClient;
-
+
/**
* Sends an exchange to the endpoint
*/
@@ -198,8 +197,12 @@ public abstract class JbiTestSupport ext
@Override
protected void tearDown() throws Exception {
exchangeCompletedListener.assertExchangeCompleted();
+
getServicemixClient().close();
client.stop();
+
+ assertCamelConsumerEndpointsDone();
+
camelContext.stop();
super.tearDown();
}
@@ -229,6 +232,44 @@ public abstract class JbiTestSupport ext
});
}
+ protected void disableExchangeCompletedListener() {
+ jbiContainer.removeListener(exchangeCompletedListener);
+ }
+
+ /*
+ * Assert that all CamelConsumerEndpoint are done:
+ * - there should be no more pending ContinuationData instances
+ */
+ private void assertCamelConsumerEndpointsDone() throws Exception {
+ List<CamelConsumerEndpoint> results =
findEndpoints(CamelConsumerEndpoint.class);
+
+ for (CamelConsumerEndpoint endpoint : results) {
+ assertEquals("Continuation data map should be empty on endpoint
for " + endpoint.getJbiEndpoint().getDestinationUri(),
+ 0, endpoint.getContinuationData().size());
+
+ }
+ }
+
+ /*
+ * Find endpoints in the component registry for the type provided
+ */
+ private<E> List<E> findEndpoints(Class<E> type) throws
NoSuchFieldException, IllegalAccessException {
+ List<E> results = new LinkedList<E>();
+
+ // little hack to access the endpoints map in the registry directly
+ Field field = Registry.class.getDeclaredField("endpoints");
+ field.setAccessible(true);
+ Map<String, org.apache.servicemix.common.Endpoint> endpoints = (Map)
field.get(component.getRegistry());
+
+ for (org.apache.servicemix.common.Endpoint endpoint :
endpoints.values()) {
+ if (type.isAssignableFrom(endpoint.getClass())) {
+ results.add((E) endpoint);
+ }
+ }
+
+ return results;
+ }
+
protected abstract void appendJbiActivationSpecs(
List<ActivationSpec> activationSpecList);
