You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ch...@apache.org on 2007/08/29 18:08:03 UTC
svn commit: r570851 - in /activemq/camel/trunk/components/camel-jms/src:
main/java/org/apache/camel/component/jms/
test/java/org/apache/camel/component/jms/ test/resources/
Author: chirino
Date: Wed Aug 29 09:08:02 2007
New Revision: 570851
URL: http://svn.apache.org/viewvc?rev=570851&view=rev
Log:
Added support for preserving the original message QoS options when a message gets re-sent to a JMS destination
https://issues.apache.org/activemq/browse/CAMEL-129
Added:
activemq/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsQosRouteTest.java (with props)
Modified:
activemq/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
activemq/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
activemq/camel/trunk/components/camel-jms/src/test/resources/log4j.properties
Modified: activemq/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java?rev=570851&r1=570850&r2=570851&view=diff
==============================================================================
--- activemq/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java (original)
+++ activemq/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java Wed Aug 29 09:08:02 2007
@@ -80,8 +80,15 @@
* @throws JMSException if the message could not be created
*/
public Message makeJmsMessage(Exchange exchange, Session session) throws JMSException {
- Message answer = createJmsMessage(exchange.getIn().getBody(), session);
- appendJmsProperties(answer, exchange);
+ Message answer = null;
+ if( exchange instanceof JmsExchange ) {
+ JmsExchange jmsExchange = (JmsExchange)exchange;
+ answer = jmsExchange.getIn().getJmsMessage();
+ }
+ if( answer == null ) {
+ answer = createJmsMessage(exchange.getIn().getBody(), session);
+ appendJmsProperties(answer, exchange);
+ }
return answer;
}
Modified: activemq/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java?rev=570851&r1=570850&r2=570851&view=diff
==============================================================================
--- activemq/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java (original)
+++ activemq/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java Wed Aug 29 09:08:02 2007
@@ -18,6 +18,11 @@
import javax.jms.ConnectionFactory;
import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.QueueSender;
+import javax.jms.TopicPublisher;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.util.ObjectHelper;
@@ -85,6 +90,7 @@
private PlatformTransactionManager transactionManager;
private String transactionName;
private int transactionTimeout = -1;
+ private boolean preserveMessageQos;
public JmsConfiguration() {
}
@@ -106,7 +112,59 @@
public JmsOperations createJmsOperations(boolean pubSubDomain, String destination) {
ConnectionFactory factory = getTemplateConnectionFactory();
- JmsTemplate template = useVersion102 ? new JmsTemplate102(factory, pubSubDomain) : new JmsTemplate(factory);
+
+ // I whish the spring templates had built in support for preserving the message
+ // qos when doing a send. :(
+ JmsTemplate template = useVersion102 ? new JmsTemplate102(factory, pubSubDomain) {
+ /**
+ * Override so we can support preserving the Qos settings that have
+ * been set on the message.
+ */
+ @Override
+ protected void doSend(MessageProducer producer, Message message) throws JMSException {
+ if (preserveMessageQos) {
+ long ttl = message.getJMSExpiration();
+ if (ttl != 0) {
+ ttl = ttl - System.currentTimeMillis();
+ // Message had expired.. so set the ttl as small as
+ // possible
+ if (ttl <= 0) {
+ ttl = 1;
+ }
+ }
+ if (isPubSubDomain()) {
+ ((TopicPublisher)producer).publish(message, message.getJMSDeliveryMode(), message.getJMSPriority(), ttl);
+ } else {
+ ((QueueSender)producer).send(message, message.getJMSDeliveryMode(), message.getJMSPriority(), ttl);
+ }
+ } else {
+ super.doSend(producer, message);
+ }
+ }
+ } : new JmsTemplate(factory) {
+ /**
+ * Override so we can support preserving the Qos settings that have
+ * been set on the message.
+ */
+ @Override
+ protected void doSend(MessageProducer producer, Message message) throws JMSException {
+ if (preserveMessageQos) {
+ long ttl = message.getJMSExpiration();
+ if (ttl != 0) {
+ ttl = ttl - System.currentTimeMillis();
+ // Message had expired.. so set the ttl as small as
+ // possible
+ if (ttl <= 0) {
+ ttl = 1;
+ }
+ }
+ producer.send(message, message.getJMSDeliveryMode(), message.getJMSPriority(), ttl);
+ } else {
+ super.doSend(producer, message);
+ }
+ }
+ };
+
template.setPubSubDomain(pubSubDomain);
template.setDefaultDestinationName(destination);
@@ -609,5 +667,20 @@
*/
protected ConnectionFactory createTemplateConnectionFactory() {
return getConnectionFactory();
+ }
+
+ public boolean isPreserveMessageQos() {
+ return preserveMessageQos;
+ }
+
+ /**
+ * Set to true if you want to send message using the QoS settings specified
+ * on the message. Normally the QoS settings used are the one configured
+ * on this Object.
+ *
+ * @param preserveMessageQos
+ */
+ public void setPreserveMessageQos(boolean preserveMessageQos) {
+ this.preserveMessageQos = preserveMessageQos;
}
}
Added: activemq/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsQosRouteTest.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsQosRouteTest.java?rev=570851&view=auto
==============================================================================
--- activemq/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsQosRouteTest.java (added)
+++ activemq/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsQosRouteTest.java Wed Aug 29 09:08:02 2007
@@ -0,0 +1,101 @@
+/**
+ * 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.activemq.ActiveMQConnectionFactory;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+
+import static org.apache.camel.component.jms.JmsComponent.jmsComponentClientAcknowledge;
+
+/**
+ * @version $Revision$
+ */
+public class JmsQosRouteTest extends ContextTestSupport {
+ protected String componentName = "activemq";
+
+ public void testJmsRoutePreserveQos() throws Exception {
+
+ MockEndpoint preserveEndpoint1 = (MockEndpoint) context.getEndpoint("mock:preserve-1");
+ preserveEndpoint1.expectedMessageCount(1);
+ preserveEndpoint1.message(0).header("JMSPriority").isEqualTo(1);
+
+ MockEndpoint preserveEndpoint2 = (MockEndpoint) context.getEndpoint("mock:preserve-2");
+ preserveEndpoint2.expectedMessageCount(1);
+ preserveEndpoint2.message(0).header("JMSPriority").isEqualTo(2);
+
+ template.sendBody(componentName + ":queue:p1?explicitQosEnabled=true&priority=1", "test");
+ template.sendBody(componentName + ":queue:p2?explicitQosEnabled=true&priority=2", "test");
+
+ MockEndpoint.assertIsSatisfied(preserveEndpoint1, preserveEndpoint2);
+ }
+
+ public void testJmsRouteNormalQos() throws Exception {
+
+ MockEndpoint regularEndpoint1 = (MockEndpoint) context.getEndpoint("mock:regular-1");
+ regularEndpoint1.expectedMessageCount(1);
+ regularEndpoint1.message(0).header("JMSPriority").isEqualTo(4);
+
+ MockEndpoint regularEndpoint2 = (MockEndpoint) context.getEndpoint("mock:regular-2");
+ regularEndpoint2.expectedMessageCount(1);
+ regularEndpoint2.message(0).header("JMSPriority").isEqualTo(4);
+
+ template.sendBody(componentName + ":queue:r1?explicitQosEnabled=true&priority=1", "test");
+ template.sendBody(componentName + ":queue:r2?explicitQosEnabled=true&priority=2", "test");
+
+ MockEndpoint.assertIsSatisfied(regularEndpoint1, regularEndpoint2);
+ }
+
+
+ @Override
+ protected void setUp() throws Exception {
+ super.setUp();
+ }
+
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+
+ ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
+ camelContext.addComponent(componentName, jmsComponentClientAcknowledge(connectionFactory));
+
+ return camelContext;
+ }
+
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() throws Exception {
+
+ // Messages should arrive at mock:preserve with their priorities preserved.
+ from(componentName + ":queue:p1").to(componentName + ":queue:preserve-1?preserveMessageQos=true");
+ from(componentName + ":queue:preserve-1").to("mock:preserve-1");
+ from(componentName + ":queue:p2").to(componentName + ":queue:preserve-2?preserveMessageQos=true");
+ from(componentName + ":queue:preserve-2").to("mock:preserve-2");
+
+ // Messages will NOT arrive at mock:regular with their priorities preserved.
+ from(componentName + ":queue:r1").to(componentName + ":queue:regular-1");
+ from(componentName + ":queue:regular-1").to("mock:regular-1");
+ from(componentName + ":queue:r2").to(componentName + ":queue:regular-2");
+ from(componentName + ":queue:regular-2").to("mock:regular-2");
+
+ }
+ };
+ }
+}
Propchange: activemq/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsQosRouteTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: activemq/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsQosRouteTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: activemq/camel/trunk/components/camel-jms/src/test/resources/log4j.properties
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-jms/src/test/resources/log4j.properties?rev=570851&r1=570850&r2=570851&view=diff
==============================================================================
--- activemq/camel/trunk/components/camel-jms/src/test/resources/log4j.properties (original)
+++ activemq/camel/trunk/components/camel-jms/src/test/resources/log4j.properties Wed Aug 29 09:08:02 2007
@@ -15,17 +15,39 @@
## limitations under the License.
## ------------------------------------------------------------------------
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
#
-# The logging properties used for eclipse testing, We want to see debug output on the console.
+# The logging properties used during tests..
#
log4j.rootLogger=INFO, out
-log4j.logger.org.springframework=WARN
-log4j.logger.org.apache.activemq=WARN
-#log4j.logger.org.apache.camel=DEBUG
+log4j.logger.org.apache.activemq.spring=WARN
# CONSOLE appender not used by default
-log4j.appender.out=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+
+# File appender
+log4j.appender.out=org.apache.log4j.FileAppender
log4j.appender.out.layout=org.apache.log4j.PatternLayout
-log4j.appender.out.layout.ConversionPattern=[%30.30t] %-30.30c{1} %-5p %m%n
-#log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+log4j.appender.out.file=target/camel-test.log
+log4j.appender.out.append=true
+