From 6a4a246e31c5290c90e2271c5f0ca0bea93e6754 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 17 Nov 2008 17:07:11 +0000 Subject: [PATCH] JmsOutboundGateway now throws a MessageTimeoutException in case a JMS reply Message is not received within the alloted timeout. The null object is no longer passed to the MessageConverter where a misleading Exception ("payload must not be null") was being thrown (INT-479). --- .../integration/jms/JmsOutboundGateway.java | 4 ++ .../endpoint/AbstractPollingEndpoint.java | 16 ++++--- .../message/MessageTimeoutException.java | 42 +++++++++++++++++++ 3 files changed, 53 insertions(+), 9 deletions(-) create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/message/MessageTimeoutException.java diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 437d6a594d..053cf06e9b 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -35,6 +35,7 @@ import org.springframework.integration.handler.AbstractReplyProducingMessageHand import org.springframework.integration.handler.ReplyMessageHolder; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.message.MessageTimeoutException; import org.springframework.jms.connection.ConnectionFactoryUtils; import org.springframework.jms.support.JmsUtils; import org.springframework.jms.support.converter.MessageConverter; @@ -199,6 +200,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp final Message requestMessage = MessageBuilder.fromMessage(message).build(); try { javax.jms.Message jmsReply = JmsOutboundGateway.this.sendAndReceive(requestMessage); + if (jmsReply == null) { + throw new MessageTimeoutException(message, "failed to receive JMS response within timeout of: " + this.receiveTimeout + "ms"); + } Object result = this.messageConverter.fromMessage(jmsReply); replyMessageHolder.set(result); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index 350ef622f0..56e35c457e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -55,6 +55,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private volatile TaskExecutor taskExecutor; + private volatile ErrorHandler errorHandler; + private volatile PlatformTransactionManager transactionManager; private volatile TransactionDefinition transactionDefinition; @@ -70,8 +72,6 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private volatile Runnable poller; private volatile boolean initialized; - - private volatile ErrorHandler errorHandler; private final Object initializationMonitor = new Object(); @@ -96,7 +96,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement public void setTaskExecutor(TaskExecutor taskExecutor) { this.taskExecutor = taskExecutor; } - + public void setErrorHandler(ErrorHandler errorHandler){ this.errorHandler = errorHandler; } @@ -148,13 +148,11 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.transactionManager, this.transactionDefinition); } this.poller = this.createPoller(); - if(this.taskExecutor != null){ - if(this.errorHandler == null){ - taskExecutor = new ErrorHandlingTaskExecutor( - new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(getBeanFactory())),taskExecutor); - } else { - taskExecutor = new ErrorHandlingTaskExecutor(errorHandler, taskExecutor); + if (this.taskExecutor != null) { + if (this.errorHandler == null) { + this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(getBeanFactory())); } + this.taskExecutor = new ErrorHandlingTaskExecutor(this.errorHandler, this.taskExecutor); } this.initialized = true; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageTimeoutException.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageTimeoutException.java new file mode 100644 index 0000000000..d568f2d3bc --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageTimeoutException.java @@ -0,0 +1,42 @@ +/* + * Copyright 2002-2008 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.message; + +import org.springframework.integration.core.Message; + +/** + * @author Mark Fisher + */ +public class MessageTimeoutException extends MessageHandlingException { + + public MessageTimeoutException(Message failedMessage, String description, Throwable cause) { + super(failedMessage, description, cause); + } + + public MessageTimeoutException(Message failedMessage, String description) { + super(failedMessage, description); + } + + public MessageTimeoutException(Message failedMessage, Throwable cause) { + super(failedMessage, cause); + } + + public MessageTimeoutException(Message failedMessage) { + super(failedMessage); + } + +}