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

INT-904, INT-907 added exception mapper support to AbstractMessagingGateway, Changed ChannelPublishingMessageListener to extend form AbstractMesagingGateway, added namespace support, tests, etc...

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-06-25 01:41:30 +00:00
parent c86e24cd34
commit 1e1c0e84b7
12 changed files with 385 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,12 +23,12 @@ import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.gateway.AbstractMessagingGateway;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.destination.DestinationResolver;
@@ -43,8 +43,12 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
* @author Juergen Hoeller
* @author Oleg Zhurakousky
*/
public class ChannelPublishingJmsMessageListener implements SessionAwareMessageListener<javax.jms.Message>, InitializingBean {
public class ChannelPublishingJmsMessageListener extends AbstractMessagingGateway
implements SessionAwareMessageListener<javax.jms.Message>, InitializingBean {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile boolean expectReply;
@@ -68,41 +72,12 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL
private volatile JmsHeaderMapper headerMapper;
private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
/**
* Specify the channel to which request Messages should be sent.
*/
public void setRequestChannel(MessageChannel requestChannel) {
this.channelTemplate.setDefaultChannel(requestChannel);
}
/**
* Specify whether a JMS reply Message is expected.
*/
public void setExpectReply(boolean expectReply) {
this.expectReply = expectReply;
}
/**
* Specify the maximum time to wait when sending a request message to the
* request channel. The default value will be that of
* {@link MessageChannelTemplate}.
*/
public void setRequestTimeout(long requestTimeout) {
this.channelTemplate.setSendTimeout(requestTimeout);
}
/**
* Specify the maximum time to wait for reply Messages. This value is only
* relevant if {@link #expectReply} is <code>true</code>. The default
* value will be that of {@link MessageChannelTemplate}.
*/
public void setReplyTimeout(long replyTimeout) {
this.channelTemplate.setReceiveTimeout(replyTimeout);
}
/**
* Set the default reply destination to send reply messages to. This will
* be applied in case of a request message that does not carry a
@@ -226,8 +201,7 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL
public void setExtractReplyPayload(boolean extractReplyPayload) {
this.extractReplyPayload = extractReplyPayload;
}
public final void afterPropertiesSet() {
public final void onInit() {
if (!(this.messageConverter instanceof HeaderMappingMessageConverter)) {
HeaderMappingMessageConverter hmmc = new HeaderMappingMessageConverter(this.messageConverter, this.headerMapper);
hmmc.setExtractJmsMessageBody(this.extractRequestPayload);
@@ -241,31 +215,31 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL
Message<?> requestMessage = (object instanceof Message<?>) ?
(Message<?>) object : MessageBuilder.withPayload(object).build();
if (!this.expectReply) {
boolean sent = this.channelTemplate.send(requestMessage);
if (!sent) {
throw new MessageDeliveryException(requestMessage, "failed to send Message to request channel");
}
this.send(requestMessage);
}
else {
Message<?> replyMessage = this.channelTemplate.sendAndReceive(requestMessage);
if (replyMessage != null) {
Destination destination = this.getReplyDestination(jmsMessage, session);
javax.jms.Message jmsReply = this.messageConverter.toMessage(replyMessage, session);
if (jmsReply.getJMSCorrelationID() == null) {
jmsReply.setJMSCorrelationID(jmsMessage.getJMSMessageID());
}
MessageProducer producer = session.createProducer(destination);
try {
if (this.explicitQosEnabledForReplies) {
producer.send(jmsReply,
this.replyDeliveryMode, this.replyPriority, this.replyTimeToLive);
Message<?> replyMessage = this.sendAndReceiveMessage(requestMessage);
if (replyMessage != null){
Destination destination = this.getReplyDestination(jmsMessage, session, false);
if (destination != null){
javax.jms.Message jmsReply = this.messageConverter.toMessage(replyMessage, session);
if (jmsReply.getJMSCorrelationID() == null) {
jmsReply.setJMSCorrelationID(jmsMessage.getJMSMessageID());
}
else {
producer.send(jmsReply);
MessageProducer producer = session.createProducer(destination);
try {
if (this.explicitQosEnabledForReplies) {
producer.send(jmsReply,
this.replyDeliveryMode, this.replyPriority, this.replyTimeToLive);
}
else {
producer.send(jmsReply);
}
}
finally {
producer.close();
}
}
finally {
producer.close();
}
}
}
@@ -273,7 +247,9 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL
/**
* Determine a reply destination for the given message.
* <p>This implementation first checks the JMS Reply-To {@link Destination}
* <p>This implementation first checks the boolean 'error' flag which signifies that the reply is an error message.
* If it is it will attempt to get the destination value from {@link JmsHeaders#SEND_ERROR_TO}.
* If reply is not an error it will first check the JMS Reply-To {@link Destination}
* of the supplied request message; if that is not <code>null</code> it is
* returned; if it is <code>null</code>, then the configured
* {@link #resolveDefaultReplyDestination default reply destination}
@@ -287,7 +263,8 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL
* @see #setDefaultReplyDestination
* @see javax.jms.Message#getJMSReplyTo()
*/
private Destination getReplyDestination(javax.jms.Message request, Session session) throws JMSException {
private Destination getReplyDestination(javax.jms.Message request, Session session, boolean error) throws JMSException {
Destination replyTo = request.getJMSReplyTo();
if (replyTo == null) {
replyTo = resolveDefaultReplyDestination(session);
@@ -296,6 +273,7 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL
"Request message does not contain reply-to destination, and no default reply destination set.");
}
}
return replyTo;
}
@@ -337,4 +315,19 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL
}
}
@Override
protected Object fromMessage(Message<?> message) {
throw new UnsupportedOperationException("'fromMessage' is not supported within this instance");
}
@Override
protected Message<?> toMessage(Object object) {
Message<?> message = null;
if (object instanceof Throwable){
message = super.toMessage(object);
} else if (object instanceof Message<?>) {
message = (Message<?>) object;
}
return message;
}
}

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

@@ -156,6 +156,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-request-payload");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-reply-payload");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "exception-mapper");
int defaults = 0;
if (StringUtils.hasText(element.getAttribute(DEFAULT_REPLY_DESTINATION_ATTRIB))) {
defaults++;

1
spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd
View File

@@ -468,6 +468,7 @@
</xsd:attribute>
<xsd:attribute name="request-destination-name" type="xsd:string"/>
<xsd:attribute name="request-pub-sub-domain" type="xsd:string"/>
<xsd:attribute name="exception-mapper" type="xsd:string"/>
<xsd:attribute name="default-reply-destination" type="xsd:string">
<xsd:annotation>
<xsd:documentation>

66
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/Exception-nonSiProducer-siConsumer.xml Normal file
View File

@@ -0,0 +1,66 @@
<?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.ExceptionHandlingSiConsumerTests$SampleGateway"
default-request-channel="outbound-channel">
</int:gateway>
<int:channel id="outbound-channel"/>
<int-jms:outbound-gateway request-channel="outbound-channel" request-destination="requestQueue"/>
<int-jms:inbound-gateway request-destination="requestQueue"
request-channel="jmsinputchannel"
exception-mapper="errorMessageMapper"/>
<bean id="errorMessageMapper" class="org.springframework.integration.jms.config.ExceptionHandlingSiConsumerTests$SampleErrorMessageMapper"/>
<int:channel id="jmsinputchannel"/>
<int:router input-channel="jmsinputchannel" expression="payload"/>
<int:service-activator input-channel="echoWithExceptionChannel" method="echoWithException">
<bean class="org.springframework.integration.jms.config.ExceptionHandlingSiConsumerTests$SampleService"/>
</int:service-activator>
<int:service-activator input-channel="echoChannel" method="echo">
<bean class="org.springframework.integration.jms.config.ExceptionHandlingSiConsumerTests$SampleService"/>
</int:service-activator>
<bean id="requestQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="request.queue"/>
</bean>
<bean id="replyQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="reply.queue"/>
</bean>
<bean id="replyTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="reply.topic"/>
</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"/>
<property name="cacheProducers" value="false"/>
</bean>
</beans>

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

@@ -0,0 +1,142 @@
/*
* 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 java.io.File;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.message.InboundMessageMapper;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
/**
* @author Oleg Zhurakousky
*
*/
public class ExceptionHandlingSiConsumerTests {
@Test
public void nonSiProducer_siConsumer_sync_withReturn() throws Exception {
ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("Exception-nonSiProducer-siConsumer.xml", ExceptionHandlingSiConsumerTests.class);
JmsTemplate jmsTemplate = new JmsTemplate(applicationContext.getBean("connectionFactory", ConnectionFactory.class));
Destination request = applicationContext.getBean("requestQueue", Destination.class);
final Destination reply = applicationContext.getBean("replyQueue", Destination.class);
jmsTemplate.send(request, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("echoChannel");
message.setJMSReplyTo(reply);
return message;
}
});
Message message = jmsTemplate.receive(reply);
Assert.assertNotNull(message);
applicationContext.close();
}
@Test
public void nonSiProducer_siConsumer_sync_withReturnNoException() throws Exception {
ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("Exception-nonSiProducer-siConsumer.xml", ExceptionHandlingSiConsumerTests.class);
JmsTemplate jmsTemplate = new JmsTemplate(applicationContext.getBean("connectionFactory", ConnectionFactory.class));
Destination request = applicationContext.getBean("requestQueue", Destination.class);
final Destination reply = applicationContext.getBean("replyQueue", Destination.class);
jmsTemplate.send(request, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("echoWithExceptionChannel");
message.setJMSReplyTo(reply);
return message;
}
});
Message message = jmsTemplate.receive(reply);
Assert.assertNotNull(message);
applicationContext.close();
}
@Test
public void nonSiProducer_siConsumer_sync_withOutboundGateway() throws Exception{
final ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("Exception-nonSiProducer-siConsumer.xml", ExceptionHandlingSiConsumerTests.class);
SampleGateway gateway = applicationContext.getBean("sampleGateway", SampleGateway.class);
String reply = gateway.echo("echoWithExceptionChannel");
System.out.println("Reply: " + reply);
applicationContext.close();
}
@Before
public void prepare() throws Exception {
System.out.println("####### Refreshing ActiveMq ########");
File activeMqTempDir = new File("activemq-data");
this.deleteDir(activeMqTempDir);
}
/*
*
*/
private void deleteDir(File directory){
if (directory.exists()){
String[] children = directory.list();
if (children != null){
for (int i=0; i < children.length; i++) {
deleteDir(new File(directory, children[i]));
}
}
}
directory.delete();
}
public static class SampleService{
public String echoWithException(String value){
throw new SampleException("echoWithException");
}
public String echo(String value){
return value;
}
}
@SuppressWarnings("serial")
public static class SampleException extends RuntimeException{
public SampleException(String message){
super(message);
}
}
public static interface SampleGateway{
public String echo(String value);
}
public static class SampleErrorMessageMapper implements InboundMessageMapper<Throwable>{
public org.springframework.integration.core.Message<?> toMessage(
Throwable t) throws Exception {
return MessageBuilder.withPayload(t.getCause().getMessage()).build();
}
}
}