diff --git a/org.springframework.integration.jms/.classpath b/org.springframework.integration.jms/.classpath index ab9b27b384..46328184a6 100644 --- a/org.springframework.integration.jms/.classpath +++ b/org.springframework.integration.jms/.classpath @@ -6,6 +6,8 @@ + + diff --git a/org.springframework.integration.jms/ivy.xml b/org.springframework.integration.jms/ivy.xml index e2e9b21734..19f3104067 100644 --- a/org.springframework.integration.jms/ivy.xml +++ b/org.springframework.integration.jms/ivy.xml @@ -21,6 +21,8 @@ + + diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannel.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannel.java new file mode 100644 index 0000000000..cf86c9cd75 --- /dev/null +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannel.java @@ -0,0 +1,161 @@ +/* + * Copyright 2002-2009 the original author or authors. + * + * Licensed 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.springframework.integration.jms; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.MessageListener; +import javax.jms.Topic; + +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.channel.SubscribableChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.MessagingException; +import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; +import org.springframework.integration.dispatcher.UnicastingDispatcher; +import org.springframework.integration.gateway.SimpleMessageMapper; +import org.springframework.integration.message.InboundMessageMapper; +import org.springframework.integration.message.MessageHandler; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.jms.support.destination.DestinationResolver; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.Assert; + +/** + * A {@link MessageChannel} implementation that is actually backed by a JMS + * Destination. This class is useful as a drop-in replacement for any + * Spring Integration channel. The benefit of using this channel is that + * the full power of any JMS provider is available with only minimal + * configuration changes and without requiring any code changes. The most + * obvious benefit is the ability to delegate message persistence to the + * JMS provider. + * + * @author Mark Fisher + * @since 2.0 + */ +public class JmsDestinationBackedMessageChannel implements + SubscribableChannel, MessageListener, BeanNameAware, InitializingBean { + + private final JmsTemplate jmsTemplate = new JmsTemplate(); + + private final DefaultMessageListenerContainer container = new DefaultMessageListenerContainer(); + + private final InboundMessageMapper mapper = new SimpleMessageMapper(); + + private volatile MessageDispatcher dispatcher; + + private volatile String name; + + + public JmsDestinationBackedMessageChannel(ConnectionFactory connectionFactory, Destination destination) { + Assert.notNull(connectionFactory, "connectionFactory must not be null"); + Assert.notNull(destination, "destination must not be null"); + this.jmsTemplate.setConnectionFactory(connectionFactory); + this.jmsTemplate.setDefaultDestination(destination); + this.initDispatcher(destination instanceof Topic); + } + + public JmsDestinationBackedMessageChannel(ConnectionFactory connectionFactory, String destinationName, boolean isPubSub) { + this(connectionFactory, destinationName, isPubSub, null); + } + + public JmsDestinationBackedMessageChannel(ConnectionFactory connectionFactory, String destinationName, boolean isPubSub, DestinationResolver destinationResolver) { + Assert.notNull(connectionFactory, "connectionFactory must not be null"); + Assert.hasText(destinationName, "destinationName is required"); + this.jmsTemplate.setConnectionFactory(connectionFactory); + if (destinationResolver != null) { + this.jmsTemplate.setDestinationResolver(destinationResolver); + } + this.jmsTemplate.setDefaultDestinationName(destinationName); + this.jmsTemplate.setPubSubDomain(isPubSub); + this.initDispatcher(isPubSub); + } + + + public void setBeanName(String beanName) { + this.name = beanName; + } + + private void initDispatcher(boolean isPubSub) { + if (isPubSub) { + this.dispatcher = new BroadcastingDispatcher(); + } + else { + UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(); + unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); + this.dispatcher = unicastingDispatcher; + } + } + + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.container.setTransactionManager(transactionManager); + } + + public void afterPropertiesSet() throws Exception { + this.container.setConnectionFactory(this.jmsTemplate.getConnectionFactory()); + Destination destination = this.jmsTemplate.getDefaultDestination(); + if (destination != null) { + this.container.setDestination(destination); + } + else { + this.container.setDestinationName(this.jmsTemplate.getDefaultDestinationName()); + this.container.setPubSubDomain(this.jmsTemplate.isPubSubDomain()); + } + this.container.setMessageListener(this); + this.container.afterPropertiesSet(); + } + + public String getName() { + return this.name; + } + + public boolean subscribe(MessageHandler handler) { + return this.dispatcher.addHandler(handler); + } + + public boolean unsubscribe(MessageHandler handler) { + return this.dispatcher.removeHandler(handler); + } + + public boolean send(Message message) { + this.jmsTemplate.convertAndSend(message); + return true; + } + + public boolean send(Message message, long timeout) { + return this.send(message); + } + + + // MessageListener implementation + + public void onMessage(javax.jms.Message message) { + try { + Object o = this.jmsTemplate.getMessageConverter().fromMessage(message); + this.dispatcher.dispatch(this.mapper.toMessage(o)); + } + catch (Exception e) { + throw new MessagingException("failed to handle incoming JMS Message", e); + } + } + +} diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannelTests.java b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannelTests.java new file mode 100644 index 0000000000..911e852182 --- /dev/null +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannelTests.java @@ -0,0 +1,192 @@ +/* + * Copyright 2002-2009 the original author or authors. + * + * Licensed 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.springframework.integration.jms; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import javax.jms.Destination; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.activemq.command.ActiveMQTopic; + +import org.junit.Before; +import org.junit.Test; + +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageHandler; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class JmsDestinationBackedMessageChannelTests { + + private ActiveMQConnectionFactory connectionFactory; + + private Destination topic; + + private Destination queue; + + + @Before + public void setup() throws Exception { + this.connectionFactory = new ActiveMQConnectionFactory(); + this.connectionFactory.setBrokerURL("vm://localhost?broker.persistent=false"); + this.topic = new ActiveMQTopic("testTopic"); + this.queue = new ActiveMQQueue("testQueue"); + } + + @Test + public void queueReference() throws Exception { + final CountDownLatch latch = new CountDownLatch(2); + final List> receivedList1 = new ArrayList>(); + MessageHandler handler1 = new MessageHandler() { + public void handleMessage(Message message) { + receivedList1.add(message); + latch.countDown(); + } + }; + final List> receivedList2 = new ArrayList>(); + MessageHandler handler2 = new MessageHandler() { + public void handleMessage(Message message) { + receivedList2.add(message); + latch.countDown(); + } + }; + JmsDestinationBackedMessageChannel channel = + new JmsDestinationBackedMessageChannel(this.connectionFactory, this.queue); + channel.afterPropertiesSet(); + channel.subscribe(handler1); + channel.subscribe(handler2); + channel.send(new StringMessage("foo")); + channel.send(new StringMessage("bar")); + latch.await(3000, TimeUnit.MILLISECONDS); + assertEquals(1, receivedList1.size()); + assertNotNull(receivedList1.get(0)); + assertEquals("foo", receivedList1.get(0).getPayload()); + assertEquals(1, receivedList2.size()); + assertNotNull(receivedList2.get(0)); + assertEquals("bar", receivedList2.get(0).getPayload()); + } + + @Test + public void topicReference() throws Exception { + final CountDownLatch latch = new CountDownLatch(4); + final List> receivedList1 = new ArrayList>(); + MessageHandler handler1 = new MessageHandler() { + public void handleMessage(Message message) { + receivedList1.add(message); + latch.countDown(); + } + }; + final List> receivedList2 = new ArrayList>(); + MessageHandler handler2 = new MessageHandler() { + public void handleMessage(Message message) { + receivedList2.add(message); + latch.countDown(); + } + }; + JmsDestinationBackedMessageChannel channel = + new JmsDestinationBackedMessageChannel(this.connectionFactory, this.topic); + channel.afterPropertiesSet(); + channel.subscribe(handler1); + channel.subscribe(handler2); + channel.send(new StringMessage("foo")); + channel.send(new StringMessage("bar")); + latch.await(3000, TimeUnit.MILLISECONDS); + assertEquals(2, receivedList1.size()); + assertEquals("foo", receivedList1.get(0).getPayload()); + assertEquals("bar", receivedList1.get(1).getPayload()); + assertEquals(2, receivedList2.size()); + assertEquals("foo", receivedList2.get(0).getPayload()); + assertEquals("bar", receivedList2.get(1).getPayload()); + } + + @Test + public void queueName() throws Exception { + final CountDownLatch latch = new CountDownLatch(2); + final List> receivedList1 = new ArrayList>(); + MessageHandler handler1 = new MessageHandler() { + public void handleMessage(Message message) { + receivedList1.add(message); + latch.countDown(); + } + }; + final List> receivedList2 = new ArrayList>(); + MessageHandler handler2 = new MessageHandler() { + public void handleMessage(Message message) { + receivedList2.add(message); + latch.countDown(); + } + }; + JmsDestinationBackedMessageChannel channel = + new JmsDestinationBackedMessageChannel(this.connectionFactory, "dynamicQueue", false); + channel.afterPropertiesSet(); + channel.subscribe(handler1); + channel.subscribe(handler2); + channel.send(new StringMessage("foo")); + channel.send(new StringMessage("bar")); + latch.await(3000, TimeUnit.MILLISECONDS); + assertEquals(1, receivedList1.size()); + assertNotNull(receivedList1.get(0)); + assertEquals("foo", receivedList1.get(0).getPayload()); + assertEquals(1, receivedList2.size()); + assertNotNull(receivedList2.get(0)); + assertEquals("bar", receivedList2.get(0).getPayload()); + } + + @Test + public void topicName() throws Exception { + final CountDownLatch latch = new CountDownLatch(4); + final List> receivedList1 = new ArrayList>(); + MessageHandler handler1 = new MessageHandler() { + public void handleMessage(Message message) { + receivedList1.add(message); + latch.countDown(); + } + }; + final List> receivedList2 = new ArrayList>(); + MessageHandler handler2 = new MessageHandler() { + public void handleMessage(Message message) { + receivedList2.add(message); + latch.countDown(); + } + }; + JmsDestinationBackedMessageChannel channel = + new JmsDestinationBackedMessageChannel(this.connectionFactory, "dynamicTopic", true); + channel.afterPropertiesSet(); + channel.subscribe(handler1); + channel.subscribe(handler2); + channel.send(new StringMessage("foo")); + channel.send(new StringMessage("bar")); + latch.await(3000, TimeUnit.MILLISECONDS); + assertEquals(2, receivedList1.size()); + assertEquals("foo", receivedList1.get(0).getPayload()); + assertEquals("bar", receivedList1.get(1).getPayload()); + assertEquals(2, receivedList2.size()); + assertEquals("foo", receivedList2.get(0).getPayload()); + assertEquals("bar", receivedList2.get(1).getPayload()); + } + +}