INT-1743 added support for MessageHistory to JMS Channel

This commit is contained in:
Oleg Zhurakousky
2011-01-13 12:59:32 -05:00
parent 1941e53ba7
commit 529cf4686e
4 changed files with 111 additions and 4 deletions

View File

@@ -1,13 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.5.0.201010221000-RELEASE]]></pluginVersion>
<pluginVersion><![CDATA[2.5.2.201101121000-SR1]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
<enableImports><![CDATA[false]]></enableImports>
<configs>
<config>src/test/java/org/springframework/integration/jms/config/GlobalChannelInterceptorTests-context.xml</config>
<config>src/test/java/org/springframework/integration/jms/config/JmsChannelHistory-context.xml</config>
</configs>
<configSets>
</configSets>

View File

@@ -24,6 +24,7 @@ import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.Session;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.SmartLifecycle;
@@ -44,9 +45,10 @@ import org.springframework.util.ErrorHandler;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.0
*/
public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChannel> implements SmartLifecycle, DisposableBean {
public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChannel> implements SmartLifecycle, DisposableBean, BeanNameAware {
private volatile AbstractJmsChannel channel;
@@ -103,6 +105,8 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
private volatile Long receiveTimeout;
private volatile Long recoveryInterval;
private volatile String beanName;
/**
* This value differs from the container implementations' default (which is AUTO_ACKNOWLEDGE)
@@ -322,6 +326,7 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
this.channel.setInterceptors(this.interceptors);
}
this.channel.afterPropertiesSet();
this.channel.setBeanName(this.beanName);
return this.channel;
}
@@ -414,7 +419,9 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
return container;
}
public void setBeanName(String name) {
this.beanName = name;
}
/*
* SmartLifecycle implementation (delegates to the created channel if message-driven)
*/
@@ -457,5 +464,4 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
((SubscribableJmsChannel) this.channel).destroy();
}
}
}

View File

@@ -0,0 +1,38 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
<int:message-history/>
<int-jms:channel id="jmsChannel" queue="requestQueue"/>
<int:service-activator input-channel="jmsChannel" ref="subscriber"/>
<bean id="subscriber" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.integration.jms.config.JmsChannelHistoryTests.JmsService"/>
</bean>
<bean id="requestQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="request.queue"/>
</bean>
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
</bean>
<bean id="jmsTransactionManager"
class="org.springframework.jms.connection.JmsTransactionManager">
<property name="connectionFactory" ref="connectionFactory" />
</bean>
</beans>

View File

@@ -0,0 +1,62 @@
/*
* Copyright 2002-2011 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.config;
import static junit.framework.Assert.assertTrue;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.jms.SubscribableJmsChannel;
import org.springframework.integration.message.GenericMessage;
/**
* @author Oleg Zhurakousky
*
*/
public class JmsChannelHistoryTests {
@Test
public void testReqRepPerformanceWithConsecutiveMessages() throws Exception{
ActiveMqTestUtils.prepare();
ApplicationContext ac = new ClassPathXmlApplicationContext("JmsChannelHistory-context.xml", this.getClass());
SubscribableJmsChannel jmsChannel = ac.getBean("jmsChannel", SubscribableJmsChannel.class);
JmsService service = ac.getBean("subscriber", JmsService.class);
jmsChannel.send(new GenericMessage<String>("hello"));
Thread.sleep(1000);
verify(service, times(1)).handleMessage(Mockito.any(Message.class));
}
public static interface JmsGateway{
public Long echo(String time);
}
public static class JmsService implements MessageHandler{
public void handleMessage(Message<?> message) throws MessagingException {
MessageHistory history = MessageHistory.read(message);
assertTrue(history.contains("jmsChannel"));
}
}
}