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/14 08:48:26 UTC
svn commit: r1157475 - in /camel/trunk/components/camel-jms/src:
main/java/org/apache/camel/component/jms/
main/java/org/apache/camel/component/jms/reply/
test/java/org/apache/camel/component/jms/
Author: davsclaus
Date: Sun Aug 14 06:48:25 2011
New Revision: 1157475
URL: http://svn.apache.org/viewvc?rev=1157475&view=rev
Log:
CAMEL-4202: Added ReplyToType option. Added new Exclusive replyToType to support per producer exclusive persistent reply queues for request/reply over JMS. This allows to run faster than the Shared option.
Added:
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/ReplyToType.java
- copied, changed from r1157369, camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/ConsumerType.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ExclusivePersistentQueueMessageListenerContainer.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/SharedPersistentQueueMessageListenerContainer.java
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToComponentTest.java
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToTest.java (contents, props changed)
- copied, changed from r1157369, camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyFixedReplyToInEndpointTest.java
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplySharedReplyToTest.java
Modified:
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsProducer.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyManager.java
Modified: camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java?rev=1157475&r1=1157474&r2=1157475&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java (original)
+++ camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsComponent.java Sun Aug 14 06:48:25 2011
@@ -332,6 +332,14 @@ public class JmsComponent extends Defaul
getConfiguration().setDestinationResolver(destinationResolver);
}
+ public ReplyToType getReplyToType() {
+ return configuration.getReplyToType();
+ }
+
+ public void setReplyToType(ReplyToType replyToType) {
+ configuration.setReplyToType(replyToType);
+ }
+
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
Modified: camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java?rev=1157475&r1=1157474&r2=1157475&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java (original)
+++ camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java Sun Aug 14 06:48:25 2011
@@ -124,6 +124,7 @@ public class JmsConfiguration implements
private boolean forceSendOriginalMessage;
// to force disabling time to live (works in both in-only or in-out mode)
private boolean disableTimeToLive;
+ private ReplyToType replyToType;
public JmsConfiguration() {
}
@@ -1176,4 +1177,19 @@ public class JmsConfiguration implements
public void setDisableTimeToLive(boolean disableTimeToLive) {
this.disableTimeToLive = disableTimeToLive;
}
+
+ /**
+ * Gets the reply to type.
+ * <p/>
+ * Will only return a value if this option has been explicit configured.
+ *
+ * @return the reply type if configured, otherwise <tt>null</tt>
+ */
+ public ReplyToType getReplyToType() {
+ return replyToType;
+ }
+
+ public void setReplyToType(ReplyToType replyToType) {
+ this.replyToType = replyToType;
+ }
}
Modified: camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java?rev=1157475&r1=1157474&r2=1157475&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java (original)
+++ camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsEndpoint.java Sun Aug 14 06:48:25 2011
@@ -1009,6 +1009,16 @@ public class JmsEndpoint extends Default
configuration.setDisableTimeToLive(disableTimeToLive);
}
+ @ManagedAttribute
+ public ReplyToType getReplyToType() {
+ return configuration.getReplyToType();
+ }
+
+ @ManagedAttribute
+ public void setReplyToType(ReplyToType replyToType) {
+ configuration.setReplyToType(replyToType);
+ }
+
@ManagedAttribute(description = "Camel id")
public String getCamelId() {
return getCamelContext().getName();
@@ -1048,7 +1058,4 @@ public class JmsEndpoint extends Default
return super.createEndpointUri();
}
-
-
-
}
Modified: camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsProducer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsProducer.java?rev=1157475&r1=1157474&r2=1157475&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsProducer.java (original)
+++ camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsProducer.java Sun Aug 14 06:48:25 2011
@@ -18,7 +18,6 @@ package org.apache.camel.component.jms;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
-
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
@@ -67,6 +66,15 @@ public class JmsProducer extends Default
return;
}
try {
+ // validate that replyToType and replyTo is configured accordingly
+ if (endpoint.getReplyToType() != null) {
+ // setting temporary with a fixed replyTo is not supported
+ if (endpoint.getReplyTo() != null && endpoint.getReplyToType() == ReplyToType.Temporary) {
+ throw new IllegalArgumentException("ReplyToType " + ReplyToType.Temporary
+ + " is not supported when replyTo " + endpoint.getReplyTo() + " is also configured.");
+ }
+ }
+
if (endpoint.getReplyTo() != null) {
replyManager = endpoint.getReplyManager(endpoint.getReplyTo());
LOG.info("Using JmsReplyManager: " + replyManager + " to process replies from: " + endpoint.getReplyTo());
Copied: camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/ReplyToType.java (from r1157369, camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/ConsumerType.java)
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/ReplyToType.java?p2=camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/ReplyToType.java&p1=camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/ConsumerType.java&r1=1157369&r2=1157475&rev=1157475&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/ConsumerType.java (original)
+++ camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/ReplyToType.java Sun Aug 14 06:48:25 2011
@@ -17,8 +17,10 @@
package org.apache.camel.component.jms;
/**
+ * Types for replyTo queues
+ *
* @version
*/
-public enum ConsumerType {
- Simple, Default
+public enum ReplyToType {
+ Temporary, Shared, Exclusive
}
Added: camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ExclusivePersistentQueueMessageListenerContainer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ExclusivePersistentQueueMessageListenerContainer.java?rev=1157475&view=auto
==============================================================================
--- camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ExclusivePersistentQueueMessageListenerContainer.java (added)
+++ camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ExclusivePersistentQueueMessageListenerContainer.java Sun Aug 14 06:48:25 2011
@@ -0,0 +1,38 @@
+/**
+ * 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.jms.reply;
+
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+
+/**
+ * This {@link org.springframework.jms.listener.DefaultMessageListenerContainer} is used for persistent reply queues
+ * which are exclusive.
+ * <p/>
+ * Mind that exclusive persistent reply queues is per producer, so if you run in a clustered environment then
+ * each producer should use an unique reply queue destination name. If not then other nodes may steal reply
+ * messages which was intended for another. For clustered environments it may be safer to use shared persistent queues
+ * as each node will only consume reply messages which are intended for itself.
+ * <p/>
+ * See more details at <a href="http://camel.apache.org/jms">camel-jms</a>.
+ *
+ * @see SharedPersistentQueueMessageListenerContainer
+ */
+public class ExclusivePersistentQueueMessageListenerContainer extends DefaultMessageListenerContainer {
+
+ // no need to override any methods currently
+
+}
Modified: camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyManager.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyManager.java?rev=1157475&r1=1157474&r2=1157475&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyManager.java (original)
+++ camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyManager.java Sun Aug 14 06:48:25 2011
@@ -25,6 +25,7 @@ import javax.jms.Session;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
+import org.apache.camel.component.jms.ReplyToType;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.support.destination.DestinationResolver;
@@ -124,45 +125,35 @@ public class PersistentQueueReplyManager
}
};
- private final class PersistentQueueMessageListenerContainer extends DefaultMessageListenerContainer {
-
- private String fixedMessageSelector;
- private MessageSelectorCreator creator;
-
- private PersistentQueueMessageListenerContainer(String fixedMessageSelector) {
- this.fixedMessageSelector = fixedMessageSelector;
- }
-
- private PersistentQueueMessageListenerContainer(MessageSelectorCreator creator) {
- this.creator = creator;
- }
-
- @Override
- public String getMessageSelector() {
- String id = null;
- if (fixedMessageSelector != null) {
- id = fixedMessageSelector;
- } else if (creator != null) {
- id = creator.get();
- }
- log.trace("Using MessageSelector[{}]", id);
- return id;
- }
- }
-
protected AbstractMessageListenerContainer createListenerContainer() throws Exception {
DefaultMessageListenerContainer answer;
- String replyToSelectorName = endpoint.getReplyToDestinationSelectorName();
- if (replyToSelectorName != null) {
- // create a random selector value we will use for the persistent reply queue
- replyToSelectorValue = "ID:" + new BigInteger(24 * 8, new Random()).toString(16);
- String fixedMessageSelector = replyToSelectorName + "='" + replyToSelectorValue + "'";
- answer = new PersistentQueueMessageListenerContainer(fixedMessageSelector);
+ ReplyToType type = endpoint.getConfiguration().getReplyToType();
+ if (type == null) {
+ // use shared by default for persistent reply queues
+ type = ReplyToType.Shared;
+ }
+
+ if (ReplyToType.Shared == type) {
+ // shared reply to queues support either a fixed or dynamic JMS message selector
+ String replyToSelectorName = endpoint.getReplyToDestinationSelectorName();
+ if (replyToSelectorName != null) {
+ // create a random selector value we will use for the persistent reply queue
+ replyToSelectorValue = "ID:" + new BigInteger(24 * 8, new Random()).toString(16);
+ String fixedMessageSelector = replyToSelectorName + "='" + replyToSelectorValue + "'";
+ answer = new SharedPersistentQueueMessageListenerContainer(fixedMessageSelector);
+ log.debug("Using shared queue: " + endpoint.getReplyTo() + " with fixed message selector [" + fixedMessageSelector + "] as reply listener: " + answer);
+ } else {
+ // use a dynamic message selector which will select the message we want to receive as reply
+ dynamicMessageSelector = new MessageSelectorCreator();
+ answer = new SharedPersistentQueueMessageListenerContainer(dynamicMessageSelector);
+ log.debug("Using shared queue: " + endpoint.getReplyTo() + " with dynamic message selector as reply listener: " + answer);
+ }
+ } else if (ReplyToType.Exclusive == type) {
+ answer = new ExclusivePersistentQueueMessageListenerContainer();
+ log.debug("Using exclusive queue:" + endpoint.getReplyTo() + " as reply listener: " + answer);
} else {
- // use a dynamic message selector which will select the message we want to receive as reply
- dynamicMessageSelector = new MessageSelectorCreator();
- answer = new PersistentQueueMessageListenerContainer(dynamicMessageSelector);
+ throw new IllegalArgumentException("ReplyToType " + type + " is not supported for persistent reply queues");
}
DestinationResolver resolver = endpoint.getDestinationResolver();
Added: camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/SharedPersistentQueueMessageListenerContainer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/SharedPersistentQueueMessageListenerContainer.java?rev=1157475&view=auto
==============================================================================
--- camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/SharedPersistentQueueMessageListenerContainer.java (added)
+++ camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/SharedPersistentQueueMessageListenerContainer.java Sun Aug 14 06:48:25 2011
@@ -0,0 +1,71 @@
+/**
+ * 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.jms.reply;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+
+/**
+ * This {@link DefaultMessageListenerContainer} is used for persistent reply queues which are shared.
+ * <p/>
+ * This implementation supports using a fixed or dynamic JMS Message Selector to pickup the
+ * designated reply messages from the shared queue. Since the queue is shared, then we can only
+ * pickup the reply messages which is intended for us, so to support that we must use JMS
+ * Message Selectors.
+ *
+ * @see ExclusivePersistentQueueMessageListenerContainer
+ */
+public class SharedPersistentQueueMessageListenerContainer extends DefaultMessageListenerContainer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SharedPersistentQueueMessageListenerContainer.class);
+
+ private String fixedMessageSelector;
+ private MessageSelectorCreator creator;
+
+ /**
+ * Use a fixed JMS message selector
+ *
+ * @param fixedMessageSelector the fixed selector
+ */
+ public SharedPersistentQueueMessageListenerContainer(String fixedMessageSelector) {
+ this.fixedMessageSelector = fixedMessageSelector;
+ }
+
+ /**
+ * Use a dynamic JMS message selector
+ *
+ * @param creator the create to create the dynamic selector
+ */
+ public SharedPersistentQueueMessageListenerContainer(MessageSelectorCreator creator) {
+ this.creator = creator;
+ }
+
+ @Override
+ public String getMessageSelector() {
+ // override this method and return the appropriate selector
+ String id = null;
+ if (fixedMessageSelector != null) {
+ id = fixedMessageSelector;
+ } else if (creator != null) {
+ id = creator.get();
+ }
+ LOG.trace("Using MessageSelector[{}]", id);
+ return id;
+ }
+
+}
Added: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToComponentTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToComponentTest.java?rev=1157475&view=auto
==============================================================================
--- camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToComponentTest.java (added)
+++ camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToComponentTest.java Sun Aug 14 06:48:25 2011
@@ -0,0 +1,71 @@
+/**
+ * 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.jms;
+
+import javax.jms.ConnectionFactory;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.camel.util.StopWatch;
+import org.junit.Test;
+
+import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge;
+
+/**
+ * Using exclusive fixed replyTo queues should be faster as there is no need for
+ * JMSMessage selectors.
+ *
+ * @version
+ */
+public class JmsRequestReplyExclusiveReplyToComponentTest extends CamelTestSupport {
+
+ @Test
+ public void testJmsRequestReplyExclusiveFixedReplyTo() throws Exception {
+ StopWatch watch = new StopWatch();
+
+ assertEquals("Hello A", template.requestBody("activemq:queue:foo?replyTo=bar", "A"));
+ assertEquals("Hello B", template.requestBody("activemq:queue:foo?replyTo=bar", "B"));
+ assertEquals("Hello C", template.requestBody("activemq:queue:foo?replyTo=bar", "C"));
+ assertEquals("Hello D", template.requestBody("activemq:queue:foo?replyTo=bar", "D"));
+ assertEquals("Hello E", template.requestBody("activemq:queue:foo?replyTo=bar", "E"));
+
+ long delta = watch.stop();
+ assertTrue("Should be faster than about 4 seconds, was: " + delta, delta < 4200);
+ }
+
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ ConnectionFactory connectionFactory = CamelJmsTestHelper.createConnectionFactory();
+ // mark the reply to type as exclusive on the component
+ JmsComponent jms = jmsComponentAutoAcknowledge(connectionFactory);
+ jms.setReplyToType(ReplyToType.Exclusive);
+ camelContext.addComponent("activemq", jms);
+ return camelContext;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("activemq:queue:foo")
+ .transform(body().prepend("Hello "));
+ }
+ };
+ }
+}
\ No newline at end of file
Copied: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToTest.java (from r1157369, camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyFixedReplyToInEndpointTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToTest.java?p2=camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToTest.java&p1=camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyFixedReplyToInEndpointTest.java&r1=1157369&r2=1157475&rev=1157475&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyFixedReplyToInEndpointTest.java (original)
+++ camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToTest.java Sun Aug 14 06:48:25 2011
@@ -19,63 +19,47 @@ package org.apache.camel.component.jms;
import javax.jms.ConnectionFactory;
import org.apache.camel.CamelContext;
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.FailedToCreateProducerException;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.camel.util.StopWatch;
import org.junit.Test;
import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge;
/**
+ * Using exclusive fixed replyTo queues should be faster as there is no need for
+ * JMSMessage selectors.
+ *
* @version
*/
-public class JmsRequestReplyFixedReplyToInEndpointTest extends CamelTestSupport {
+public class JmsRequestReplyExclusiveReplyToTest extends CamelTestSupport {
@Test
- public void testJmsRequestReplyTempReplyTo() throws Exception {
- Exchange reply = template.request("activemq:queue:foo", new Processor() {
- public void process(Exchange exchange) throws Exception {
- exchange.getIn().setBody("World");
- }
- });
- assertEquals("Hello World", reply.getOut().getBody());
- assertTrue("Should have headers", reply.getOut().hasHeaders());
- String replyTo = reply.getOut().getHeader("JMSReplyTo", String.class);
- assertTrue("Should be a temp queue", replyTo.startsWith("temp-queue"));
- }
+ public void testJmsRequestReplyExclusiveFixedReplyTo() throws Exception {
+ StopWatch watch = new StopWatch();
- @Test
- public void testJmsRequestReplyFixedReplyToInEndpoint() throws Exception {
- Exchange reply = template.request("activemq:queue:foo?replyTo=bar", new Processor() {
- public void process(Exchange exchange) throws Exception {
- exchange.getIn().setBody("World");
- }
- });
- assertEquals("Hello World", reply.getOut().getBody());
- assertTrue("Should have headers", reply.getOut().hasHeaders());
- assertEquals("queue://bar", reply.getOut().getHeader("JMSReplyTo", String.class));
+ assertEquals("Hello A", template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Exclusive", "A"));
+ assertEquals("Hello B", template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Exclusive", "B"));
+ assertEquals("Hello C", template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Exclusive", "C"));
+ assertEquals("Hello D", template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Exclusive", "D"));
+ assertEquals("Hello E", template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Exclusive", "E"));
+
+ long delta = watch.stop();
+ assertTrue("Should be faster than about 4 seconds, was: " + delta, delta < 4200);
}
@Test
- public void testJmsRequestReplyFixedReplyToInEndpointTwoMessages() throws Exception {
- Exchange reply = template.request("activemq:queue:foo?replyTo=bar", new Processor() {
- public void process(Exchange exchange) throws Exception {
- exchange.getIn().setBody("World");
- }
- });
- assertEquals("Hello World", reply.getOut().getBody());
- assertTrue("Should have headers", reply.getOut().hasHeaders());
- assertEquals("queue://bar", reply.getOut().getHeader("JMSReplyTo", String.class));
-
- reply = template.request("activemq:queue:foo?replyTo=bar", new Processor() {
- public void process(Exchange exchange) throws Exception {
- exchange.getIn().setBody("Moon");
- }
- });
- assertEquals("Hello Moon", reply.getOut().getBody());
- assertTrue("Should have headers", reply.getOut().hasHeaders());
- assertEquals("queue://bar", reply.getOut().getHeader("JMSReplyTo", String.class));
+ public void testInvalidConfiguration() throws Exception {
+ try {
+ template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Temporary", "Hello World");
+ fail("Should have thrown exception");
+ } catch (CamelExecutionException e) {
+ assertIsInstanceOf(FailedToCreateProducerException.class, e.getCause());
+ assertIsInstanceOf(IllegalArgumentException.class, e.getCause().getCause());
+ assertEquals("ReplyToType Temporary is not supported when replyTo bar is also configured.", e.getCause().getCause().getMessage());
+ }
}
protected CamelContext createCamelContext() throws Exception {
Propchange: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplyExclusiveReplyToTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplySharedReplyToTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplySharedReplyToTest.java?rev=1157475&view=auto
==============================================================================
--- camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplySharedReplyToTest.java (added)
+++ camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsRequestReplySharedReplyToTest.java Sun Aug 14 06:48:25 2011
@@ -0,0 +1,68 @@
+/**
+ * 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.jms;
+
+import javax.jms.ConnectionFactory;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.camel.util.StopWatch;
+import org.junit.Test;
+
+import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge;
+
+/**
+ * @version
+ */
+public class JmsRequestReplySharedReplyToTest extends CamelTestSupport {
+
+ @Test
+ public void testJmsRequestReplySharedReplyTo() throws Exception {
+ StopWatch watch = new StopWatch();
+
+ // shared is more slower than exclusive, due it need to use a JMS Message Selector
+ // and has a receiveTimeout of 1 sec per default, so it react slower to new messages
+
+ assertEquals("Hello A", template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Shared", "A"));
+ assertEquals("Hello B", template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Shared", "B"));
+ assertEquals("Hello C", template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Shared", "C"));
+ assertEquals("Hello D", template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Shared", "D"));
+ assertEquals("Hello E", template.requestBody("activemq:queue:foo?replyTo=bar&replyToType=Shared", "E"));
+
+ long delta = watch.stop();
+ assertTrue("Should be slower than about 3 seconds, was: " + delta, delta > 3200);
+ }
+
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ ConnectionFactory connectionFactory = CamelJmsTestHelper.createConnectionFactory();
+ camelContext.addComponent("activemq", jmsComponentAutoAcknowledge(connectionFactory));
+ return camelContext;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("activemq:queue:foo")
+ .transform(body().prepend("Hello "));
+ }
+ };
+ }
+}
\ No newline at end of file