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/01/12 07:43:06 UTC
svn commit: r898191 - in /camel/branches/camel-1.x/components/camel-jms/src:
main/java/org/apache/camel/component/jms/JmsBinding.java
test/java/org/apache/camel/component/jms/JmsProducerWithJMSHeaderTest.java
Author: davsclaus
Date: Tue Jan 12 06:43:06 2010
New Revision: 898191
URL: http://svn.apache.org/viewvc?rev=898191&view=rev
Log:
CAMEL-2351: Backported per message JMSPriority and JMSExpiration to be supported by Camel JMS producer.
Modified:
camel/branches/camel-1.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
camel/branches/camel-1.x/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsProducerWithJMSHeaderTest.java
Modified: camel/branches/camel-1.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-1.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java?rev=898191&r1=898190&r2=898191&view=diff
==============================================================================
--- camel/branches/camel-1.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java (original)
+++ camel/branches/camel-1.x/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java Tue Jan 12 06:43:06 2010
@@ -248,6 +248,10 @@
jmsMessage.setJMSReplyTo(ExchangeHelper.convertToType(exchange, Destination.class, headerValue));
} else if (headerName.equals("JMSType")) {
jmsMessage.setJMSType(ExchangeHelper.convertToType(exchange, String.class, headerValue));
+ } else if (headerName.equals("JMSPriority")) {
+ jmsMessage.setJMSPriority(ExchangeHelper.convertToType(exchange, Integer.class, headerValue));
+ } else if (headerName.equals("JMSExpiration")) {
+ jmsMessage.setJMSExpiration(ExchangeHelper.convertToType(exchange, Long.class, headerValue));
} else if (LOG.isDebugEnabled()) {
// The following properties are set by the MessageProducer
// JMSDeliveryMode, JMSDestination, JMSExpiration,
Modified: camel/branches/camel-1.x/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsProducerWithJMSHeaderTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-1.x/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsProducerWithJMSHeaderTest.java?rev=898191&r1=898190&r2=898191&view=diff
==============================================================================
--- camel/branches/camel-1.x/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsProducerWithJMSHeaderTest.java (original)
+++ camel/branches/camel-1.x/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsProducerWithJMSHeaderTest.java Tue Jan 12 06:43:06 2010
@@ -16,14 +16,19 @@
*/
package org.apache.camel.component.jms;
+import java.util.HashMap;
+import java.util.Map;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.camel.CamelContext;
import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.PollingConsumer;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
+
import static org.apache.camel.component.jms.JmsComponent.jmsComponentClientAcknowledge;
/**
@@ -31,6 +36,138 @@
*/
public class JmsProducerWithJMSHeaderTest extends ContextTestSupport {
+ public void testInOnlyJMSPrioritory() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ mock.message(0).header("JMSPriority").isEqualTo(2);
+
+ template.sendBodyAndHeader("activemq:queue:foo?preserveMessageQos=true", "Hello World", "JMSPriority", "2");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testInOnlyJMSPrioritoryZero() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ mock.message(0).header("JMSPriority").isEqualTo(0);
+
+ template.sendBodyAndHeader("activemq:queue:foo?preserveMessageQos=true", "Hello World", "JMSPriority", "0");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testInOnlyJMSPrioritoryNine() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ mock.message(0).header("JMSPriority").isEqualTo(9);
+
+ template.sendBodyAndHeader("activemq:queue:foo?preserveMessageQos=true", "Hello World", "JMSPriority", "9");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testInOnlyJMSExpiration() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ long ttl = System.currentTimeMillis() + 5000;
+ template.sendBodyAndHeader("activemq:queue:bar?preserveMessageQos=true", "Hello World", "JMSExpiration", ttl);
+
+ // sleep just a little
+ Thread.sleep(2000);
+
+ PollingConsumer consumer = context.getEndpoint("activemq:queue:bar").createPollingConsumer();
+ consumer.start();
+
+ // use timeout in case running on slow box
+ Exchange bar = consumer.receive(10000);
+ assertNotNull("Should be a message on queue", bar);
+
+ consumer.stop();
+
+ template.send("activemq:queue:foo", bar);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testInOnlyJMSExpirationNoMessage() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ long ttl = System.currentTimeMillis() + 2000;
+ template.sendBodyAndHeader("activemq:queue:bar?preserveMessageQos=true", "Hello World", "JMSExpiration", ttl);
+
+ // sleep more so the message is expired
+ Thread.sleep(5000);
+
+ PollingConsumer consumer = context.getEndpoint("activemq:queue:bar").createPollingConsumer();
+ consumer.start();
+
+ Exchange bar = consumer.receiveNoWait();
+ assertNull("Should NOT be a message on queue", bar);
+
+ consumer.stop();
+
+ template.sendBody("activemq:queue:foo", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testInOnlyMultipleJMSHeadersAndExpiration() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ mock.message(0).header("JMSPriority").isEqualTo(3);
+
+ long ttl = System.currentTimeMillis() + 2000;
+ Map headers = new HashMap();
+ headers.put("JMSPriority", 3);
+ headers.put("JMSExpiration", ttl);
+ template.sendBodyAndHeaders("activemq:queue:bar?preserveMessageQos=true", "Hello World", headers);
+
+ // sleep just a little
+ Thread.sleep(50);
+
+ PollingConsumer consumer = context.getEndpoint("activemq:queue:bar").createPollingConsumer();
+ consumer.start();
+
+ Exchange bar = consumer.receive(5000);
+ assertNotNull("Should be a message on queue", bar);
+
+ consumer.stop();
+
+ template.send("activemq:queue:foo?preserveMessageQos=true", bar);
+
+ Thread.sleep(1000);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testInOnlyMultipleJMSHeadersAndExpirationNoMessage() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ long ttl = System.currentTimeMillis() + 2000;
+ Map headers = new HashMap();
+ headers.put("JMSPriority", 3);
+ headers.put("JMSExpiration", ttl);
+ template.sendBodyAndHeaders("activemq:queue:bar?preserveMessageQos=true", "Hello World", headers);
+
+ // sleep more so the message is expired
+ Thread.sleep(5000);
+
+ PollingConsumer consumer = context.getEndpoint("activemq:queue:bar").createPollingConsumer();
+ consumer.start();
+
+ Exchange bar = consumer.receiveNoWait();
+ assertNull("Should NOT be a message on queue", bar);
+
+ consumer.stop();
+
+ template.sendBody("activemq:queue:foo", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ }
+
public void testInOnlyJMSDestinationName() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedMessageCount(1);