You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ja...@apache.org on 2011/03/11 20:52:29 UTC
svn commit: r1080742 - in /camel/trunk/components/camel-jms/src:
main/java/org/apache/camel/component/jms/
test/java/org/apache/camel/component/jms/
Author: janstey
Date: Fri Mar 11 19:52:29 2011
New Revision: 1080742
URL: http://svn.apache.org/viewvc?rev=1080742&view=rev
Log:
CAMEL-3782 - add option to force sending original JMS message
Added:
camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsLoadBalanceFailoverWithForceSendOriginalJmsMessageTest.java
Modified:
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsConfiguration.java
Modified: camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java?rev=1080742&r1=1080741&r2=1080742&view=diff
==============================================================================
--- camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java (original)
+++ camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java Fri Mar 11 19:52:29 2011
@@ -261,7 +261,7 @@ public class JmsBinding {
boolean alwaysCopy = endpoint != null && endpoint.getConfiguration().isAlwaysCopyMessage();
if (!alwaysCopy && camelMessage instanceof JmsMessage) {
JmsMessage jmsMessage = (JmsMessage)camelMessage;
- if (!jmsMessage.shouldCreateNewMessage()) {
+ if (!jmsMessage.shouldCreateNewMessage() || endpoint.getConfiguration().isForceSendOriginalMessage()) {
answer = jmsMessage.getJmsMessage();
}
}
@@ -296,7 +296,7 @@ public class JmsBinding {
* Appends the JMS headers from the Camel {@link JmsMessage}
*/
public void appendJmsProperties(Message jmsMessage, Exchange exchange, org.apache.camel.Message in) throws JMSException {
- Set<Map.Entry<String, Object>> entries = in.getHeaders().entrySet();
+ Set<Map.Entry<String, Object>> entries = in.getHeaders().entrySet();
for (Map.Entry<String, Object> entry : entries) {
String headerName = entry.getKey();
Object headerValue = entry.getValue();
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=1080742&r1=1080741&r2=1080742&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 Fri Mar 11 19:52:29 2011
@@ -113,6 +113,9 @@ public class JmsConfiguration implements
private boolean transferExchange;
private boolean transferException;
private boolean testConnectionOnStartup;
+ // if the message is a JmsMessage and mapJmsMessage=false, force the
+ // producer to send the javax.jms.Message body to the next JMS destination
+ private boolean forceSendOriginalMessage;
public JmsConfiguration() {
}
@@ -1075,4 +1078,13 @@ public class JmsConfiguration implements
public void setTestConnectionOnStartup(boolean testConnectionOnStartup) {
this.testConnectionOnStartup = testConnectionOnStartup;
}
+
+ public void setForceSendOriginalMessage(boolean forceSendOriginalMessage) {
+ this.forceSendOriginalMessage = forceSendOriginalMessage;
+ }
+
+ public boolean isForceSendOriginalMessage() {
+ return forceSendOriginalMessage;
+ }
+
}
Added: camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsLoadBalanceFailoverWithForceSendOriginalJmsMessageTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsLoadBalanceFailoverWithForceSendOriginalJmsMessageTest.java?rev=1080742&view=auto
==============================================================================
--- camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsLoadBalanceFailoverWithForceSendOriginalJmsMessageTest.java (added)
+++ camel/trunk/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsLoadBalanceFailoverWithForceSendOriginalJmsMessageTest.java Fri Mar 11 19:52:29 2011
@@ -0,0 +1,129 @@
+/**
+ * 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.activemq.command.ActiveMQTextMessage;
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge;
+
+/**
+ * Unit test for Camel loadbalancer failover with JMS
+ */
+public class JmsLoadBalanceFailoverWithForceSendOriginalJmsMessageTest extends CamelTestSupport {
+ private boolean forceSendOriginalMessage = true;
+
+ @Test
+ public void testFailover() throws Exception {
+ MockEndpoint oneMock = getMockEndpoint("mock:one");
+ MockEndpoint threeMock = getMockEndpoint("mock:three");
+ MockEndpoint twoMock = getMockEndpoint("mock:two");
+ MockEndpoint resultMock = getMockEndpoint("mock:result");
+
+
+ oneMock.expectedMessageCount(1);
+ oneMock.expectedHeaderReceived("foo", "bar");
+ twoMock.expectedMessageCount(1);
+ twoMock.expectedHeaderReceived("foo", "bar");
+ threeMock.expectedMessageCount(0);
+ resultMock.expectedMessageCount(1);
+ resultMock.expectedHeaderReceived("foo", "bar");
+
+ String out = template.requestBodyAndHeader("jms:queue:start?jmsMessageType=Text&forceSendOriginalMessage=" + forceSendOriginalMessage, "Hello World", "foo", "bar", String.class);
+ assertEquals("Hello Back", out);
+
+ assertMockEndpointsSatisfied();
+
+ // we should get an ActiveMQTextMessage with the body and custom header intact
+ assertEquals(ActiveMQTextMessage.class, oneMock.getExchanges().get(0).getIn().getBody().getClass());
+ assertEquals("Hello World", ((ActiveMQTextMessage) oneMock.getExchanges().get(0).getIn().getBody()).getText());
+ assertEquals("bar", ((ActiveMQTextMessage) oneMock.getExchanges().get(0).getIn().getBody()).getStringProperty("foo"));
+
+ assertEquals(ActiveMQTextMessage.class, twoMock.getExchanges().get(0).getIn().getBody().getClass());
+ assertEquals("Hello World", ((ActiveMQTextMessage) twoMock.getExchanges().get(0).getIn().getBody()).getText());
+ assertEquals("bar", ((ActiveMQTextMessage) twoMock.getExchanges().get(0).getIn().getBody()).getStringProperty("foo"));
+
+ // reset mocks
+ oneMock.reset();
+ twoMock.reset();
+ threeMock.reset();
+ resultMock.reset();
+
+ // the round robin should now be at three so one and two should be skipped
+
+ oneMock.expectedMessageCount(0);
+ twoMock.expectedMessageCount(0);
+ threeMock.expectedMessageCount(1);
+ threeMock.expectedHeaderReceived("foo", "bar");
+ resultMock.expectedMessageCount(1);
+ resultMock.expectedHeaderReceived("foo", "bar");
+
+ out = template.requestBodyAndHeader("jms:queue:start?jmsMessageType=Text&forceSendOriginalMessage=" + forceSendOriginalMessage, "Hello World", "foo", "bar", String.class);
+ assertEquals("Bye World", out);
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals(ActiveMQTextMessage.class, threeMock.getExchanges().get(0).getIn().getBody().getClass());
+ assertEquals("Hello World", ((ActiveMQTextMessage) threeMock.getExchanges().get(0).getIn().getBody()).getText());
+ assertEquals("bar", ((ActiveMQTextMessage) threeMock.getExchanges().get(0).getIn().getBody()).getStringProperty("foo"));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("jms:queue:start?mapJmsMessage=false")
+ .loadBalance().failover(-1, false, true)
+ .to("jms:queue:one?transferException=true&forceSendOriginalMessage=" + forceSendOriginalMessage)
+ .to("jms:queue:two?transferException=true&forceSendOriginalMessage=" + forceSendOriginalMessage)
+ .to("jms:queue:three?transferException=true&forceSendOriginalMessage=" + forceSendOriginalMessage)
+ .end()
+ .to("mock:result");
+
+ from("jms:queue:one?transferException=true&mapJmsMessage=false")
+ .to("mock:one")
+ .throwException(new IllegalArgumentException("Damn"));
+
+ from("jms:queue:two?transferException=true&mapJmsMessage=false")
+ .to("mock:two")
+ .transform().simple("Hello Back");
+
+ from("jms:queue:three?transferException=true&mapJmsMessage=false")
+ .to("mock:three")
+ .transform().simple("Bye World");
+ }
+ };
+ }
+
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+
+ ConnectionFactory connectionFactory = CamelJmsTestHelper.createConnectionFactory();
+ camelContext.addComponent("jms", jmsComponentAutoAcknowledge(connectionFactory));
+
+ return camelContext;
+ }
+
+}