From 1e1c0e84b7e249b88ecdf94de92aef20e3554de0 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 25 Jun 2010 01:41:30 +0000 Subject: [PATCH] INT-904, INT-907 added exception mapper support to AbstractMessagingGateway, Changed ChannelPublishingMessageListener to extend form AbstractMesagingGateway, added namespace support, tests, etc... --- .../integration/config/xml/GatewayParser.java | 2 +- .../gateway/AbstractMessagingGateway.java | 49 +++++- .../gateway/GatewayProxyFactoryBean.java | 12 ++ .../gateway/SimpleMessagingGateway.java | 33 ++-- .../config/xml/spring-integration-2.0.xsd | 10 ++ ...wayInvokingMessageHandlerTests-context.xml | 27 ++-- .../GatewayInvokingMessageHandlerTests.java | 54 +++---- .../ChannelPublishingJmsMessageListener.java | 109 +++++++------- .../JmsMessageDrivenEndpointParser.java | 1 + .../jms/config/spring-integration-jms-2.0.xsd | 1 + .../Exception-nonSiProducer-siConsumer.xml | 66 ++++++++ .../ExceptionHandlingSiConsumerTests.java | 142 ++++++++++++++++++ 12 files changed, 385 insertions(+), 121 deletions(-) create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/config/Exception-nonSiProducer-siConsumer.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExceptionHandlingSiConsumerTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java index ebd839298b..aedd49d8b7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java @@ -39,7 +39,7 @@ import org.w3c.dom.Element; public class GatewayParser extends AbstractSimpleBeanDefinitionParser { private static String[] referenceAttributes = new String[] { - "default-request-channel", "default-reply-channel", "message-mapper" + "default-request-channel", "default-reply-channel", "message-mapper", "exception-mapper" }; private static String[] innerAttributes = new String[] { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java index 6106af5a6e..86bf1b8697 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java @@ -27,6 +27,8 @@ import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.message.ErrorMessage; +import org.springframework.integration.message.InboundMessageMapper; +import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.scheduling.support.PeriodicTrigger; @@ -41,6 +43,8 @@ import org.springframework.util.Assert; * @author Mark Fisher */ public abstract class AbstractMessagingGateway extends AbstractEndpoint { + + private volatile InboundMessageMapper exceptionMapper; private volatile MessageChannel requestChannel; @@ -120,7 +124,7 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { } } - public void send(Object object) { + protected void send(Object object) { this.initializeIfNecessary(); Assert.state(this.requestChannel != null, "send is not supported, because no request channel has been configured"); @@ -131,7 +135,7 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { } } - public Object receive() { + protected Object receive() { this.initializeIfNecessary(); Assert.state(this.replyChannel != null && (this.replyChannel instanceof PollableChannel), "receive is not supported, because no pollable reply channel has been configured"); @@ -147,15 +151,15 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { } } - public Object sendAndReceive(Object object) { + protected Object sendAndReceive(Object object) { return this.sendAndReceive(object, true); } - public Message sendAndReceiveMessage(Object object) { + protected Message sendAndReceiveMessage(Object object) { return (Message) this.sendAndReceive(object, false); } - private Object sendAndReceive(Object object, boolean shouldMapMessage) { + Object sendAndReceive(Object object, boolean shouldMapMessage) { Message request = this.toMessage(object); Message reply = this.sendAndReceiveMessage(request); if (!shouldMapMessage) { @@ -174,7 +178,19 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { if (this.replyChannel != null && this.replyMessageCorrelator == null) { this.registerReplyMessageCorrelator(); } - Message reply = this.channelTemplate.sendAndReceive(message, this.requestChannel); + Message reply = null; + try { + reply = this.channelTemplate.sendAndReceive(message, this.requestChannel); + } catch (Exception e) { + logger.warn("Execution of endpoint by the MessageListener resulted in : " + e); + reply = this.toMessage(e); + if (reply == null){ // if reply wasn't mapped re-throw + if (e instanceof RuntimeException){ + throw (RuntimeException)e; + } + } + } + if (reply != null && this.shouldThrowErrors && reply instanceof ErrorMessage) { Throwable error = ((ErrorMessage) reply).getPayload(); if (error instanceof RuntimeException) { @@ -226,10 +242,29 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { } } + public InboundMessageMapper getExceptionMapper() { + return exceptionMapper; + } + + public void setExceptionMapper(InboundMessageMapper exceptionMapper) { + this.exceptionMapper = exceptionMapper; + } + /** * Subclasses must implement this to map from an Object to a Message. */ - protected abstract Message toMessage(Object object); + protected Message toMessage(Object object){ + if (object instanceof Throwable){ + if (this.exceptionMapper != null){ + try { + return exceptionMapper.toMessage((Throwable) object); + } catch (Exception e2) { + logger.warn("Problem mapping " + object + " to message with: " + exceptionMapper); + } + } + } + return null; + } /** * Subclasses must implement this to map from a Message to an Object. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 9409d098ba..ca99f989db 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -35,6 +35,7 @@ import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.handler.ArgumentArrayMessageMapper; +import org.springframework.integration.message.InboundMessageMapper; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.StringUtils; @@ -48,6 +49,8 @@ import org.springframework.util.StringUtils; */ public class GatewayProxyFactoryBean extends AbstractEndpoint implements FactoryBean, MethodInterceptor, BeanClassLoaderAware { + private volatile InboundMessageMapper exceptionMapper; + private volatile Class serviceInterface; private volatile MessageChannel defaultRequestChannel; @@ -287,6 +290,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Factory } ArgumentArrayMessageMapper messageMapper = new ArgumentArrayMessageMapper(method, staticHeaders); SimpleMessagingGateway gateway = new SimpleMessagingGateway(messageMapper, new SimpleMessageMapper()); + gateway.setExceptionMapper(exceptionMapper); if (this.getTaskScheduler() != null) { gateway.setTaskScheduler(this.getTaskScheduler()); } @@ -332,5 +336,13 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Factory public void setMethodToChannelMap(Map methodToChannelMap) { this.methodToChannelMap = methodToChannelMap; } + + public InboundMessageMapper getExceptionMapper() { + return exceptionMapper; + } + + public void setExceptionMapper(InboundMessageMapper exceptionMapper) { + this.exceptionMapper = exceptionMapper; + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java index 8795d238f4..b6ba447cbf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java @@ -52,6 +52,14 @@ public class SimpleMessagingGateway extends AbstractMessagingGateway { this.inboundMapper = inboundMapper; this.outboundMapper = outboundMapper; } + + public Message sendAndReceiveMessage(Object object) { + return (Message) super.sendAndReceive(object, false); + } + + public Object sendAndReceive(Object object) { + return super.sendAndReceive(object, true); + } @Override @@ -69,19 +77,24 @@ public class SimpleMessagingGateway extends AbstractMessagingGateway { @Override protected Message toMessage(Object object) { - try { - Message message = this.inboundMapper.toMessage(object); - if (message != null) { - message.getHeaders().getHistory().addEvent(this); + Message message = null; + if (object instanceof Throwable){ + message = super.toMessage(object); + } else { + try { + message = this.inboundMapper.toMessage(object); + if (message != null) { + message.getHeaders().getHistory().addEvent(this); + } } - return message; - } - catch (Exception e) { - if (e instanceof RuntimeException) { - throw (RuntimeException) e; + catch (Exception e) { + if (e instanceof RuntimeException) { + throw (RuntimeException) e; + } + throw new MessagingException("failed to create Message", e); } - throw new MessagingException("failed to create Message", e); } + return message; } } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index 7e5c6a6b73..616596dae1 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -533,6 +533,16 @@ + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests-context.xml index b5586e9502..e6dbc34122 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests-context.xml @@ -13,14 +13,15 @@ - + + + + - - - - - @@ -29,18 +30,6 @@ - - - - - - - - - - - - @@ -75,5 +64,7 @@ + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java index 84da8ebc50..a6c8be97c4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java @@ -24,6 +24,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.channel.SubscribableChannel; import org.springframework.integration.core.Message; +import org.springframework.integration.message.InboundMessageMapper; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageHandlingException; @@ -51,6 +52,11 @@ public class GatewayInvokingMessageHandlerTests { @Qualifier("gatewayWithError") SimpleGateway gatewayWithError; + @Autowired + @Qualifier("gatewayWithErrorAndMapper") + SimpleGateway gatewayWithErrorAndMapper; + + @Autowired @Qualifier("inputB") SubscribableChannel output; @@ -82,39 +88,44 @@ public class GatewayInvokingMessageHandlerTests { @Test public void validateGatewayWithErrorMessageReturned() { + try { - gatewayWithError.sendRecieve("echoWithErrorMessageChannel"); - Assert.fail(); + String result = gatewayWithErrorAndMapper.sendRecieve("echoWithRuntimeExceptionChannel"); + Assert.assertNotNull(result); + Assert.assertEquals("Error happened in message: echoWithRuntimeExceptionChannel", result); } catch (Exception e) { - Assert.assertEquals("echoWithErrorMessageChannel", e.getMessage()); + Assert.fail(); } try { gatewayWithError.sendRecieve("echoWithRuntimeExceptionChannel"); Assert.fail(); - } catch (Exception e) { - Assert.assertEquals("echoWithRuntimeExceptionChannel", e.getMessage()); + } catch (MessageHandlingException e) { + Assert.assertEquals("echoWithRuntimeExceptionChannel", e.getFailedMessage().getPayload()); } - try { gatewayWithError.sendRecieve("echoWithMessagingExceptionChannel"); Assert.fail(); } catch (MessageHandlingException e) { Assert.assertEquals("echoWithMessagingExceptionChannel", e.getFailedMessage().getPayload()); } - try { - gatewayWithError.sendRecieve("echoWithCheckedExceptionChannel"); - Assert.fail(); + String result = gatewayWithErrorAndMapper.sendRecieve("echoWithMessagingExceptionChannel"); + Assert.assertNotNull(result); + Assert.assertEquals("Error happened in message: echoWithMessagingExceptionChannel", result); } catch (Exception e) { - Assert.assertEquals("echoWithCheckedExceptionChannel", e.getCause().getMessage()); + Assert.fail(); } - - //String result = gatewayWithError.sendRecieve("echoWithErrorMessageChannel"); - //System.out.println("Result: " + result); - //Assert.assertEquals("echo:echo:echo:hello", result); } + + public static class SampleExceptionMapper implements InboundMessageMapper{ + public Message toMessage(Throwable object) throws Exception { + MessageHandlingException ex = (MessageHandlingException) object; + return MessageBuilder.withPayload("Error happened in message: " + ex.getFailedMessage().getPayload()).build(); + } + + } public static interface SimpleGateway { public String sendRecieve(String str); @@ -124,22 +135,11 @@ public class GatewayInvokingMessageHandlerTests { public String echo(String value) { return "echo:" + value; } - public Message echoWithErrorMessage(String value) { - return MessageBuilder.withPayload(new RuntimeException(value)).build(); - } public RuntimeException echoWithRuntimeException(String value) { - return new RuntimeException(value); - } - public MessageHandlingException echoWithMessagingException(String value) { - return new MessageHandlingException(new StringMessage(value)); - } - public SampleCheckedException echoWithCheckedException(String value) { - return new SampleCheckedException(value); - } - public String echoWithRuntimeExceptionThrown(String value) { throw new RuntimeException(value); } - public String echoWithMessagingExceptionThrown(String value) { + + public MessageHandlingException echoWithMessagingException(String value) { throw new MessageHandlingException(new StringMessage(value)); } } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index 76f806b0cd..0a5153cece 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -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, InitializingBean { +public class ChannelPublishingJmsMessageListener extends AbstractMessagingGateway + implements SessionAwareMessageListener, 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 true. 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. - *

This implementation first checks the JMS Reply-To {@link Destination} + *

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 null it is * returned; if it is null, 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; + } } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java index d74c310099..85e3fb9762 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java @@ -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++; diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd index 2095216a2d..68db03dadb 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd @@ -468,6 +468,7 @@ + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/Exception-nonSiProducer-siConsumer.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/Exception-nonSiProducer-siConsumer.xml new file mode 100644 index 0000000000..829634f883 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/Exception-nonSiProducer-siConsumer.xml @@ -0,0 +1,66 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExceptionHandlingSiConsumerTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExceptionHandlingSiConsumerTests.java new file mode 100644 index 0000000000..bddb6e90dd --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/ExceptionHandlingSiConsumerTests.java @@ -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{ + public org.springframework.integration.core.Message toMessage( + Throwable t) throws Exception { + return MessageBuilder.withPayload(t.getCause().getMessage()).build(); + } + } +}