INT-592 Committed first version of JMS-backed MessageChannel implementation.
This commit is contained in:
Mark Fisher
@@ -6,6 +6,8 @@
|
||||
<classpathentry kind="src" path="src/test/resources" />
|
||||
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER" />
|
||||
<classpathentry kind="var" path="IVY_CACHE/javax.jms/com.springsource.javax.jms/1.1.0/com.springsource.javax.jms-1.1.0.jar" sourcepath="/IVY_CACHE/javax.jms/com.springsource.javax.jms/1.1.0/com.springsource.javax.jms-sources-1.1.0.jar" />
|
||||
<classpathentry kind="var" path="IVY_CACHE/org.apache.activemq/com.springsource.org.apache.activemq/5.2.0/com.springsource.org.apache.activemq-5.2.0.jar" sourcepath="IVY_CACHE/org.apache.activemq/com.springsource.org.apache.activemq/5.2.0/com.springsource.org.apache.activemq-sources-5.2.0.jar"/>
|
||||
<classpathentry kind="var" path="IVY_CACHE/org.apache.geronimo.specs/com.springsource.javax.management.j2ee/1.0.1/com.springsource.javax.management.j2ee-1.0.1.jar" sourcepath="IVY_CACHE/org.apache.geronimo.specs/com.springsource.javax.management.j2ee/1.0.1/com.springsource.javax.management.j2ee-sources-1.0.1.jar"/>
|
||||
<classpathentry kind="var" path="IVY_CACHE/org.apache.commons/com.springsource.org.apache.commons.logging/1.1.1/com.springsource.org.apache.commons.logging-1.1.1.jar" sourcepath="/IVY_CACHE/org.apache.commons/com.springsource.org.apache.commons.logging/1.1.1/com.springsource.org.apache.commons.logging-sources-1.1.1.jar" />
|
||||
<classpathentry kind="var" path="IVY_CACHE/org.junit/com.springsource.org.junit/4.6.0/com.springsource.org.junit-4.6.0.jar" sourcepath="/IVY_CACHE/org.junit/com.springsource.org.junit/4.6.0/com.springsource.org.junit-sources-4.6.0.jar" />
|
||||
<classpathentry kind="var" path="IVY_CACHE/org.springframework/org.springframework.asm/3.0.0.RC1/org.springframework.asm-3.0.0.RC1.jar" sourcepath="/IVY_CACHE/org.springframework/org.springframework.asm/3.0.0.RC1/org.springframework.asm-sources-3.0.0.RC1.jar" />
|
||||
|
||||
@@ -21,6 +21,8 @@
|
||||
|
||||
<dependencies>
|
||||
<dependency org="javax.jms" name="com.springsource.javax.jms" rev="1.1.0" conf="provided->runtime"/>
|
||||
<dependency org="org.apache.activemq" name="com.springsource.org.apache.activemq" rev="5.2.0" conf="test->runtime"/>
|
||||
<dependency org="org.apache.geronimo.specs" name="com.springsource.javax.management.j2ee" rev="1.0.1" conf="test->runtime"/>
|
||||
<dependency org="org.junit" name="com.springsource.org.junit" rev="${junit.version}" conf="test->runtime"/>
|
||||
<dependency org="org.springframework" name="org.springframework.jms" rev="${spring.version}" conf="compile->runtime"/>
|
||||
<dependency org="org.springframework" name="org.springframework.test" rev="${spring.version}" conf="test->runtime"/>
|
||||
|
||||
@@ -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<Object> 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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<Message<?>> receivedList1 = new ArrayList<Message<?>>();
|
||||
MessageHandler handler1 = new MessageHandler() {
|
||||
public void handleMessage(Message<?> message) {
|
||||
receivedList1.add(message);
|
||||
latch.countDown();
|
||||
}
|
||||
};
|
||||
final List<Message<?>> receivedList2 = new ArrayList<Message<?>>();
|
||||
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<Message<?>> receivedList1 = new ArrayList<Message<?>>();
|
||||
MessageHandler handler1 = new MessageHandler() {
|
||||
public void handleMessage(Message<?> message) {
|
||||
receivedList1.add(message);
|
||||
latch.countDown();
|
||||
}
|
||||
};
|
||||
final List<Message<?>> receivedList2 = new ArrayList<Message<?>>();
|
||||
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<Message<?>> receivedList1 = new ArrayList<Message<?>>();
|
||||
MessageHandler handler1 = new MessageHandler() {
|
||||
public void handleMessage(Message<?> message) {
|
||||
receivedList1.add(message);
|
||||
latch.countDown();
|
||||
}
|
||||
};
|
||||
final List<Message<?>> receivedList2 = new ArrayList<Message<?>>();
|
||||
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<Message<?>> receivedList1 = new ArrayList<Message<?>>();
|
||||
MessageHandler handler1 = new MessageHandler() {
|
||||
public void handleMessage(Message<?> message) {
|
||||
receivedList1.add(message);
|
||||
latch.countDown();
|
||||
}
|
||||
};
|
||||
final List<Message<?>> receivedList2 = new ArrayList<Message<?>>();
|
||||
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());
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user