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 2010/07/27 10:51:33 UTC
svn commit: r979591 - 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/
test/java/org/apache/camel/component/jms/issues/ test/ja...
Author: davsclaus
Date: Tue Jul 27 08:51:32 2010
New Revision: 979591
URL: http://svn.apache.org/viewvc?rev=979591&view=rev
Log:
CAMEL-2970: request-reply over JMS cant use TX.
Added:
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutPersistentReplyQueueTest.java (contents, props changed)
- copied, changed from r979551, camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutIssueTest.java
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/tx/JMSTXInOutPersistentQueueTest.java
Modified:
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/reply/PersistentQueueReplyManager.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsEndpointConfigurationTest.java
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=979591&r1=979590&r2=979591&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 Tue Jul 27 08:51:32 2010
@@ -84,7 +84,7 @@ public class JmsConfiguration implements
private long receiveTimeout = -1;
private long requestTimeout = 20000L;
private int idleTaskExecutionLimit = 1;
- private int maxConcurrentConsumers = 1;
+ private int maxConcurrentConsumers = 0;
// JmsTemplate only
private Boolean explicitQosEnabled;
private boolean deliveryPersistent = true;
@@ -869,7 +869,11 @@ public class JmsConfiguration implements
if (idleTaskExecutionLimit >= 0) {
listenerContainer.setIdleTaskExecutionLimit(idleTaskExecutionLimit);
}
- if (maxConcurrentConsumers >= 0) {
+ if (maxConcurrentConsumers > 0) {
+ if (maxConcurrentConsumers < concurrentConsumers) {
+ throw new IllegalArgumentException("Property maxConcurrentConsumers: " + maxConcurrentConsumers
+ + " must be higher than concurrentConsumers: " + concurrentConsumers);
+ }
listenerContainer.setMaxConcurrentConsumers(maxConcurrentConsumers);
}
if (maxMessagesPerTask >= 0) {
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=979591&r1=979590&r2=979591&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 Tue Jul 27 08:51:32 2010
@@ -19,18 +19,15 @@ package org.apache.camel.component.jms.r
import java.math.BigInteger;
import java.util.Random;
import javax.jms.Destination;
-import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
-import org.springframework.core.task.TaskExecutor;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.support.destination.DestinationResolver;
-import org.springframework.transaction.PlatformTransactionManager;
/**
* A {@link ReplyManager} when using persistent queues.
@@ -189,48 +186,25 @@ public class PersistentQueueReplyManager
answer.setMessageListener(this);
answer.setPubSubDomain(false);
answer.setSubscriptionDurable(false);
- answer.setConcurrentConsumers(endpoint.getConcurrentConsumers());
-
- ExceptionListener exceptionListener = endpoint.getExceptionListener();
- if (exceptionListener != null) {
- answer.setExceptionListener(exceptionListener);
- }
-
- answer.setSessionTransacted(endpoint.isTransacted());
- if (endpoint.isTransacted()) {
- answer.setSessionAcknowledgeMode(Session.SESSION_TRANSACTED);
- } else {
- if (endpoint.getAcknowledgementMode() >= 0) {
- answer.setSessionAcknowledgeMode(endpoint.getAcknowledgementMode());
- } else if (endpoint.getAcknowledgementModeName() != null) {
- answer.setSessionAcknowledgeModeName(endpoint.getAcknowledgementModeName());
- }
- }
-
answer.setConcurrentConsumers(1);
+ // must use cache level session
answer.setCacheLevel(DefaultMessageListenerContainer.CACHE_SESSION);
+ // we cannot do request-reply over JMS with transaction
+ answer.setSessionTransacted(false);
+
+ // other optional properties
+ if (endpoint.getExceptionListener() != null) {
+ answer.setExceptionListener(endpoint.getExceptionListener());
+ }
if (endpoint.getReceiveTimeout() >= 0) {
answer.setReceiveTimeout(endpoint.getReceiveTimeout());
}
if (endpoint.getRecoveryInterval() >= 0) {
answer.setRecoveryInterval(endpoint.getRecoveryInterval());
}
- TaskExecutor taskExecutor = endpoint.getTaskExecutor();
- if (taskExecutor != null) {
- answer.setTaskExecutor(taskExecutor);
- }
- PlatformTransactionManager tm = endpoint.getTransactionManager();
- if (tm != null) {
- answer.setTransactionManager(tm);
- } else if (endpoint.isTransacted()) {
- throw new IllegalArgumentException("Property transacted is enabled but a transactionManager was not injected!");
- }
- if (endpoint.getTransactionName() != null) {
- answer.setTransactionName(endpoint.getTransactionName());
- }
- if (endpoint.getTransactionTimeout() >= 0) {
- answer.setTransactionTimeout(endpoint.getTransactionTimeout());
+ if (endpoint.getTaskExecutor() != null) {
+ answer.setTaskExecutor(endpoint.getTaskExecutor());
}
return answer;
Modified: camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java?rev=979591&r1=979590&r2=979591&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java (original)
+++ camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java Tue Jul 27 08:51:32 2010
@@ -105,8 +105,9 @@ public class TemporaryQueueReplyManager
answer.setMessageListener(this);
answer.setPubSubDomain(false);
answer.setSubscriptionDurable(false);
- answer.setConcurrentConsumers(endpoint.getConcurrentConsumers());
+ answer.setConcurrentConsumers(1);
answer.setConnectionFactory(endpoint.getConnectionFactory());
+ answer.setSessionTransacted(false);
String clientId = endpoint.getClientId();
if (clientId != null) {
clientId += ".CamelReplyManager";
Modified: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsEndpointConfigurationTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsEndpointConfigurationTest.java?rev=979591&r1=979590&r2=979591&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsEndpointConfigurationTest.java (original)
+++ camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsEndpointConfigurationTest.java Tue Jul 27 08:51:32 2010
@@ -26,6 +26,7 @@ import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.ResolveEndpointFailedException;
+import org.apache.camel.processor.Logger;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.junit.Test;
import org.springframework.jms.connection.UserCredentialsConnectionFactoryAdapter;
@@ -34,6 +35,7 @@ import org.springframework.jms.core.JmsT
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.support.converter.SimpleMessageConverter;
+
import static org.apache.camel.component.jms.JmsComponent.jmsComponentClientAcknowledge;
/**
@@ -147,6 +149,17 @@ public class JmsEndpointConfigurationTes
}
@Test
+ public void testInvalidMaxConcurrentConsumers() throws Exception {
+ JmsEndpoint endpoint = (JmsEndpoint) resolveMandatoryEndpoint("jms:queue:Foo?concurrentConsumers=5&maxConcurrentConsumers=2");
+ try {
+ endpoint.createConsumer(new Logger());
+ fail("Should have thrown exception");
+ } catch (IllegalArgumentException e) {
+ assertEquals("Property maxConcurrentConsumers: 2 must be higher than concurrentConsumers: 5", e.getMessage());
+ }
+ }
+
+ @Test
public void testConcurrentConsumers() throws Exception {
JmsEndpoint endpoint = (JmsEndpoint) resolveMandatoryEndpoint("jms:queue:Foo?concurrentConsumers=4");
assertEquals(4, endpoint.getConcurrentConsumers());
@@ -195,7 +208,7 @@ public class JmsEndpointConfigurationTes
assertEquals(null, endpoint.getJmsMessageType());
assertNull(endpoint.getJmsOperations());
assertNotNull(endpoint.getListenerConnectionFactory());
- assertEquals(1, endpoint.getMaxConcurrentConsumers());
+ assertEquals(0, endpoint.getMaxConcurrentConsumers());
assertEquals(-1, endpoint.getMaxMessagesPerTask());
assertEquals(null, endpoint.getMessageConverter());
assertNotNull(endpoint.getMetadataJmsOperations());
Copied: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutPersistentReplyQueueTest.java (from r979551, camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutIssueTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutPersistentReplyQueueTest.java?p2=camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutPersistentReplyQueueTest.java&p1=camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutIssueTest.java&r1=979551&r2=979591&rev=979591&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutIssueTest.java (original)
+++ camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutPersistentReplyQueueTest.java Tue Jul 27 08:51:32 2010
@@ -16,12 +16,7 @@
*/
package org.apache.camel.component.jms.issues;
-import java.util.concurrent.Future;
-
import org.apache.camel.CamelContext;
-import org.apache.camel.Exchange;
-import org.apache.camel.ExchangePattern;
-import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.junit.Test;
@@ -31,50 +26,18 @@ import static org.apache.activemq.camel.
/**
* @version $Revision$
*/
-public class JmsInOutIssueTest extends CamelTestSupport {
-
- @Test
- public void testInOutWithRequestBody() throws Exception {
- String reply = template.requestBody("activemq:queue:in", "Hello World", String.class);
- assertEquals("Bye World", reply);
- }
-
- @Test
- public void testInOutTwoTimes() throws Exception {
- String reply = template.requestBody("activemq:queue:in", "Hello World", String.class);
- assertEquals("Bye World", reply);
-
- reply = template.requestBody("activemq:queue:in", "Hello Camel", String.class);
- assertEquals("Bye World", reply);
- }
+public class JmsInOutPersistentReplyQueueTest extends CamelTestSupport {
@Test
- public void testInOutWithAsyncRequestBody() throws Exception {
- Future<String> reply = template.asyncRequestBody("activemq:queue:in", "Hello World", String.class);
- assertEquals("Bye World", reply.get());
- }
-
- @Test
- public void testInOutWithSendExchange() throws Exception {
- Exchange out = template.send("activemq:queue:in", ExchangePattern.InOut, new Processor() {
- public void process(Exchange exchange) throws Exception {
- exchange.getIn().setBody("Hello World");
- }
- });
+ public void testInOutPersistentReplyQueue() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceivedInAnyOrder("Bye A", "Bye B", "Bye C", "Bye D");
- assertEquals("Bye World", out.getOut().getBody());
- }
-
- @Test
- public void testInOutWithAsyncSendExchange() throws Exception {
- Future<Exchange> out = template.asyncSend("activemq:queue:in", new Processor() {
- public void process(Exchange exchange) throws Exception {
- exchange.setPattern(ExchangePattern.InOut);
- exchange.getIn().setBody("Hello World");
- }
- });
+ template.sendBody("seda:start", "A");
+ template.sendBody("seda:start", "B");
+ template.sendBody("seda:start", "C");
+ template.sendBody("seda:start", "D");
- assertEquals("Bye World", out.get().getOut().getBody());
+ assertMockEndpointsSatisfied();
}
protected CamelContext createCamelContext() throws Exception {
@@ -86,11 +49,19 @@ public class JmsInOutIssueTest extends C
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
public void configure() throws Exception {
- from("activemq:queue:in").process(new Processor() {
- public void process(Exchange exchange) throws Exception {
- exchange.getOut().setBody("Bye World");
- }
- });
+ from("seda:start")
+ .log("Sending ${body}")
+ .inOut("activemq:queue:foo?replyTo=myReplies")
+ // process the remainder of the route concurrently
+ .threads(5)
+ .log("Reply ${body}")
+ .delay(2000)
+ .to("mock:result");
+
+ from("activemq:queue:foo")
+ .to("mock:foo")
+ .transform(body().prepend("Bye "))
+ .log("Sending back reply ${body}");
}
};
}
Propchange: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutPersistentReplyQueueTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsInOutPersistentReplyQueueTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/tx/JMSTXInOutPersistentQueueTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/tx/JMSTXInOutPersistentQueueTest.java?rev=979591&view=auto
==============================================================================
--- camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/tx/JMSTXInOutPersistentQueueTest.java (added)
+++ camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/tx/JMSTXInOutPersistentQueueTest.java Tue Jul 27 08:51:32 2010
@@ -0,0 +1,90 @@
+/**
+ * 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.tx;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit4.CamelSpringTestSupport;
+import org.junit.Test;
+import org.springframework.context.support.AbstractXmlApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+/**
+ * @version $Revision$
+ */
+public class JMSTXInOutPersistentQueueTest extends CamelSpringTestSupport {
+
+ private static int counter = 0;
+
+ @Override
+ protected AbstractXmlApplicationContext createApplicationContext() {
+ return new ClassPathXmlApplicationContext("org/apache/camel/component/jms/tx/JmsTransacted-context.xml");
+ }
+
+ @Override
+ protected int getExpectedRouteCount() {
+ // no routes in Spring XML
+ return 0;
+ }
+
+ @Test
+ public void testJMSTXInOutPersistentQueueWithClientRedelivery() throws Exception {
+ getMockEndpoint("mock:foo").expectedBodiesReceived("World", "World", "World");
+ getMockEndpoint("mock:reply").expectedBodiesReceived("Bye World", "Bye World", "Bye World");
+
+ try {
+ template.sendBody("direct:start", "World");
+ fail("Should thrown an exception");
+ } catch (Exception e) {
+ // ignore
+ }
+
+ // let client re-try
+ try {
+ template.sendBody("direct:start", "World");
+ fail("Should thrown an exception");
+ } catch (Exception e) {
+ // ignore
+ }
+
+ // now we succeed
+ template.sendBody("direct:start", "World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start").inOut("activemq:queue:foo?replyTo=myReplies")
+ .to("mock:reply")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ if (counter++ < 2) {
+ throw new IllegalArgumentException("Damn");
+ }
+ }
+ }).to("mock:result");
+
+ from("activemq:queue:foo").to("mock:foo").transform(body().prepend("Bye "));
+ }
+ };
+ }
+}