Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1260 added support for JMS componentTypes with transport prefix (jms:outbound-gateway), Changed AbstractJmsTemplateVasedAdapter to subclass IntegrationObjectSupport so it could be added to the history

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-07-26 04:37:52 +00:00
parent 27ba48abf0
commit 29e5b6174b
12 changed files with 358 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-jms/pom.xml
View File

@@ -35,6 +35,11 @@
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>

6
spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsTemplateBasedAdapter.java
View File

@@ -20,7 +20,7 @@ import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
public abstract class AbstractJmsTemplateBasedAdapter implements InitializingBean {
public abstract class AbstractJmsTemplateBasedAdapter extends IntegrationObjectSupport {
private volatile boolean extractPayload = true;
@@ -181,7 +181,7 @@ public abstract class AbstractJmsTemplateBasedAdapter implements InitializingBea
return this.jmsTemplate;
}
public void afterPropertiesSet() {
public void onInit() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;

13
spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.jms;
import java.util.Map;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.InvalidDestinationException;
@@ -67,6 +69,10 @@ public class ChannelPublishingJmsMessageListener extends AbstractMessagingGatewa
private volatile DestinationResolver destinationResolver = new DynamicDestinationResolver();
private volatile JmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper();
public String getComponentType(){
return "jms:inbound-gateway";
}
/**
* Specify whether a JMS reply Message is expected.
@@ -208,10 +214,15 @@ public class ChannelPublishingJmsMessageListener extends AbstractMessagingGatewa
super.onInit();
}
@SuppressWarnings("unchecked")
public void onMessage(javax.jms.Message jmsMessage, Session session) throws JMSException {
Object object = this.messageConverter.fromMessage(jmsMessage);
Map<String, Object> headers = (Map<String, Object>) headerMapper.toHeaders(jmsMessage);
Message<?> requestMessage = (object instanceof Message<?>) ?
(Message<?>) object : MessageBuilder.withPayload(object).build();
MessageBuilder.fromMessage((Message<?>) object).copyHeaders(headers).build() :
MessageBuilder.withPayload(object).copyHeaders(headers).build();
this.writeMessageHistory(requestMessage, this);
if (!this.expectReply) {
this.send(requestMessage);
}

6
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java
View File

@@ -53,13 +53,16 @@ public class JmsDestinationPollingSource extends AbstractJmsTemplateBasedAdapter
super(connectionFactory, destinationName);
}
public String getComponentType(){
return "jms:inbound-channel-adapter";
}
/**
* Specify a JMS Message Selector expression to use when receiving Messages.
*/
public void setMessageSelector(String messageSelector) {
this.messageSelector = messageSelector;
}
/**
* Will receive JMS {@link javax.jms.Message} converting and returning it as
* Spring Integration(SI) {@link Message}.
@@ -86,6 +89,7 @@ public class JmsDestinationPollingSource extends AbstractJmsTemplateBasedAdapter
} else {
convertedMessage = MessageBuilder.withPayload(convertedObject).build();
}
this.writeMessageHistory(convertedMessage, this);
} catch (Exception e) {
throw new MessagingException(e.getMessage(), e);
}

1
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java
View File

@@ -53,6 +53,7 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements Dispos
if (!this.listenerContainer.isActive()) {
this.listenerContainer.afterPropertiesSet();
}
listener.setComponentName(this.getComponentName());
}
@Override

4
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

@@ -88,6 +88,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
public String getComponentType(){
return "jms:outbound-gateway";
}
/**

5
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java
View File

@@ -35,7 +35,9 @@ public class JmsSendingMessageHandler extends AbstractJmsTemplateBasedAdapter im
private volatile int order = Ordered.LOWEST_PRECEDENCE;
public String getComponentType(){
return "jms:outbound-channel-adapter";
}
public JmsSendingMessageHandler(JmsTemplate jmsTemplate) {
super(jmsTemplate);
}
@@ -60,6 +62,7 @@ public class JmsSendingMessageHandler extends AbstractJmsTemplateBasedAdapter im
if (message == null) {
throw new IllegalArgumentException("message must not be null");
}
this.writeMessageHistory(message, this);
this.getJmsTemplate().convertAndSend(message, new MessagePostProcessor() {
public javax.jms.Message postProcessMessage(javax.jms.Message jmsMessage)
throws JMSException {

4
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java
View File

@@ -48,6 +48,10 @@ public class JmsInboundChannelAdapterParser extends AbstractPollingInboundChanne
Object source = parserContext.extractSource(element);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.jms.JmsDestinationPollingSource");
String componentName = this.resolveId(element, builder.getBeanDefinition(), parserContext);
if (StringUtils.hasText(componentName)){
builder.addPropertyValue("componentName", componentName);
}
String jmsTemplate = element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE);
String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE);
String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE);

188
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageHistoryTests.java Normal file
View File

@@ -0,0 +1,188 @@
/*
* Copyright 2002-2010 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.assertEquals;
import java.util.Iterator;
import java.util.Map;
import java.util.StringTokenizer;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.context.NamedComponent;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.history.MessageHistoryEvent;
import org.springframework.integration.jms.DefaultJmsHeaderMapper;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.StringMessage;
import org.springframework.util.StringUtils;
/**
* @author Oleg Zhurakousky
*
*/
public class JmsMessageHistoryTests {
@SuppressWarnings("unchecked")
@Test
public void testInboundAdapter() throws Exception{
ActiveMqTestUtils.prepare();
ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("MessageHistoryTests-context.xml", JmsMessageHistoryTests.class);
SampleGateway gateway = applicationContext.getBean("sampleGateway", SampleGateway.class);
PollableChannel jmsInputChannel = applicationContext.getBean("jmsInputChannel", PollableChannel.class);
gateway.send("hello");
Message<String> message = (Message<String>) jmsInputChannel.receive(5000);
Iterator<MessageHistoryEvent> historyIterator = message.getHeaders().getHistory().iterator();
MessageHistoryEvent event = historyIterator.next();
assertEquals("jms:inbound-channel-adapter", event.getType());
assertEquals("sampleJmsInboundAdapter", event.getName());
event = historyIterator.next();
assertEquals("queue-channel", event.getType());
assertEquals("jmsInputChannel", event.getName());
}
@SuppressWarnings("unchecked")
@Test
public void testWithHeaderMapperPropagatingOutboundHistory() throws Exception{
ActiveMqTestUtils.prepare();
ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("MessageHistoryTests-withHeaderMapper.xml", JmsMessageHistoryTests.class);
DirectChannel input = applicationContext.getBean("outbound-channel", DirectChannel.class);
PollableChannel jmsInputChannel = applicationContext.getBean("jmsInputChannel", PollableChannel.class);
input.send(new StringMessage("hello"));
Message<String> message = (Message<String>) jmsInputChannel.receive(50000);
Iterator<MessageHistoryEvent> historyIterator = message.getHeaders().getHistory().iterator();
MessageHistoryEvent event = historyIterator.next();
assertEquals("channel", event.getType());
assertEquals("outbound-channel", event.getName());
event = historyIterator.next();
assertEquals("jms:outbound-channel-adapter", event.getType());
assertEquals("jmsOutbound", event.getName());
event = historyIterator.next();
assertEquals("jms:inbound-channel-adapter", event.getType());
assertEquals("sampleJmsInboundAdapter", event.getName());
event = historyIterator.next();
assertEquals("queue-channel", event.getType());
assertEquals("jmsInputChannel", event.getName());
}
@SuppressWarnings("unchecked")
@Test
public void testWithHeaderMapperPropagatingOutboundHistoryWithGateways() throws Exception{
ActiveMqTestUtils.prepare();
ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("MessageHistoryTests-gateways.xml", JmsMessageHistoryTests.class);
SampleGateway gateway = applicationContext.getBean("sampleGateway", SampleGateway.class);
SubscribableChannel inboundJmsChannel = applicationContext.getBean("inbound-jms-channel", SubscribableChannel.class);
MessageHandler handler = new MessageHandler() {
public void handleMessage(Message<?> message)
throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
Iterator<MessageHistoryEvent> historyIterator = message.getHeaders().getHistory().iterator();
MessageHistoryEvent event = historyIterator.next();
assertEquals("gateway", event.getType());
assertEquals("sampleGateway", event.getName());
event = historyIterator.next();
assertEquals("pub-sub-channel", event.getType());
assertEquals("channel-a", event.getName());
event = historyIterator.next();
assertEquals("jms:outbound-gateway", event.getType());
assertEquals("jmsOutbound", event.getName());
event = historyIterator.next();
assertEquals("jms:inbound-gateway", event.getType());
assertEquals("jmsInbound", event.getName());
event = historyIterator.next();
assertEquals("pub-sub-channel", event.getType());
assertEquals("inbound-jms-channel", event.getName());
event = historyIterator.next();
assertEquals("service-activator", event.getType());
assertEquals("sampleService-a", event.getName());
}
};
handler = Mockito.spy(handler);
inboundJmsChannel.subscribe(handler);
gateway.echo("hello");
Mockito.verify(handler, Mockito.times(1)).handleMessage(Mockito.any(Message.class));
}
public static interface SampleGateway{
public void send(String value);
public Message<?> echo(String value);
}
public static class SampleService{
public Message<?> echoMessage(String value){
return new StringMessage(value);
}
}
public static class SampleHeaderMapper extends DefaultJmsHeaderMapper {
public void fromHeaders(MessageHeaders headers, javax.jms.Message jmsMessage){
super.fromHeaders(headers, jmsMessage);
String messageHistory = headers.getHistory().toString();
try {
jmsMessage.setStringProperty("outbound_history", messageHistory);
} catch (Exception e) {
throw new MessagingException("Problem setting JMS properties", e);
}
}
public Map<String, Object> toHeaders(javax.jms.Message jmsMessage){
Map<String, Object> headers = super.toHeaders(jmsMessage);
MessageHistory history = new MessageHistory();
String outboundHistory = (String) headers.get("outbound_history");
StringTokenizer tok = new StringTokenizer(outboundHistory, ",[] ");
while (tok.hasMoreTokens()) {
String historyItem = tok.nextToken();
String[] parsedHistory = StringUtils.split(historyItem, "@");
String type = null;
String name = historyItem;
if (parsedHistory != null){
name = parsedHistory[1];
type = parsedHistory[0];
}
history.addEvent(new SampleComponent(name, type));
}
headers.put(MessageHeaders.HISTORY, history);
headers.remove("outbound_history");
return headers;
}
}
public static class SampleComponent implements NamedComponent{
private String name;
private String type;
public SampleComponent(String name, String type){
this.name = name;
this.type = type;
}
public String getComponentName() {
return name;
}
public String getComponentType() {
return type;
}
}
}

46
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/MessageHistoryTests-context.xml Normal file
View File

@@ -0,0 +1,46 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
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/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:task="http://www.springframework.org/schema/task">
<int:gateway id="sampleGateway"
service-interface="org.springframework.integration.jms.config.JmsMessageHistoryTests$SampleGateway"
default-request-channel="outbound-channel">
</int:gateway>
<int:channel id="outbound-channel"/>
<int-jms:outbound-channel-adapter id="jmsOutbound" channel="outbound-channel" destination-name="request.queue_c"/>
<int-jms:inbound-channel-adapter id="sampleJmsInboundAdapter" channel="jmsInputChannel" destination-name="request.queue_c"/>
<int:channel id="jmsInputChannel">
<int:queue capacity="2"/>
</int:channel>
<int:poller id="poller" default="true">
<int:interval-trigger interval="10"/>
</int:poller>
<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"/>
<property name="cacheProducers" value="false"/>
</bean>
<bean id="historyWriter" class="org.springframework.integration.history.MessageHistoryWriter"/>
</beans>

44
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/MessageHistoryTests-gateways.xml Normal file
View File

@@ -0,0 +1,44 @@
<?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:aop="http://www.springframework.org/schema/aop"
xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
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"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms">
<int:gateway id="sampleGateway" default-request-channel="channel-a" service-interface="org.springframework.integration.jms.config.JmsMessageHistoryTests.SampleGateway"/>
<int:publish-subscribe-channel id="channel-a"/>
<int-jms:outbound-gateway id="jmsOutbound" request-channel="channel-a" request-destination-name="request.queue_b" header-mapper="headerMapper"/>
<int-jms:inbound-gateway id="jmsInbound" request-channel="inbound-jms-channel" request-destination-name="request.queue_b" header-mapper="headerMapper"/>
<int:publish-subscribe-channel id="inbound-jms-channel"/>
<int:service-activator id="sampleService-a" input-channel="inbound-jms-channel">
<bean class="org.springframework.integration.jms.config.JmsMessageHistoryTests.SampleService"/>
</int:service-activator>
<bean id="historyWriter" class="org.springframework.integration.history.MessageHistoryWriter"/>
<bean id="headerMapper" class="org.springframework.integration.jms.config.JmsMessageHistoryTests$SampleHeaderMapper"/>
<int:poller id="poller" default="true">
<int:interval-trigger interval="10"/>
</int:poller>
<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"/>
<property name="cacheProducers" value="false"/>
</bean>
</beans>

42
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/MessageHistoryTests-withHeaderMapper.xml Normal file
View File

@@ -0,0 +1,42 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
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/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:task="http://www.springframework.org/schema/task">
<int:channel id="outbound-channel"/>
<int-jms:outbound-channel-adapter id="jmsOutbound" channel="outbound-channel" destination-name="request.queue_a" header-mapper="headerMapper"/>
<int-jms:inbound-channel-adapter id="sampleJmsInboundAdapter" channel="jmsInputChannel" destination-name="request.queue_a" header-mapper="headerMapper"/>
<int:channel id="jmsInputChannel">
<int:queue capacity="2"/>
</int:channel>
<int:poller id="poller" default="true">
<int:interval-trigger interval="10"/>
</int:poller>
<bean id="headerMapper" class="org.springframework.integration.jms.config.JmsMessageHistoryTests$SampleHeaderMapper"/>
<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"/>
<property name="cacheProducers" value="false"/>
</bean>
<bean id="historyWriter" class="org.springframework.integration.history.MessageHistoryWriter"/>
</beans>