diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index 64b2c6074d..c54428243c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -40,6 +40,7 @@ import org.springframework.integration.handler.MessageHandlerNotRunningException import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; import org.springframework.integration.handler.ReplyHandler; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.MessageHeader; import org.springframework.integration.message.selector.MessageSelector; @@ -72,6 +73,8 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA private final ReplyHandler replyHandler = new EndpointReplyHandler(); + private volatile long replyTimeout = 1000; + private volatile String defaultOutputChannelName; private volatile ChannelRegistry channelRegistry; @@ -146,12 +149,24 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA return (this.errorHandler != null); } + /** + * Set the timeout in milliseconds to be enforced when this endpoint sends a + * reply message. If the message is not sent successfully within the + * allotted time, then it will be sent within a MessageDeliveryException to + * the error handler instead. The default replyTimeout value + * is 1000 milliseconds. + */ + public void setReplyTimeout(long replyTimeout) { + this.replyTimeout = replyTimeout; + } + public String getDefaultOutputChannelName() { return this.defaultOutputChannelName; } /** - * Set the name of the channel to which this endpoint can send reply messages by default. + * Set the name of the channel to which this endpoint should send reply + * messages by default. */ public void setDefaultOutputChannelName(String defaultOutputChannelName) { this.defaultOutputChannelName = defaultOutputChannelName; @@ -170,9 +185,9 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA int capacity = concurrencyPolicy.getQueueCapacity(); BlockingQueue queue = (capacity < 1) ? new SynchronousQueue() : new ArrayBlockingQueue(capacity); - ExecutorService executor = new ThreadPoolExecutor(concurrencyPolicy.getCoreSize(), - concurrencyPolicy.getMaxSize(), concurrencyPolicy.getKeepAliveSeconds(), - TimeUnit.SECONDS, queue); + ExecutorService executor = new ThreadPoolExecutor( + concurrencyPolicy.getCoreSize(), concurrencyPolicy.getMaxSize(), + concurrencyPolicy.getKeepAliveSeconds(), TimeUnit.SECONDS, queue); this.handler = new ConcurrentHandler(this.handler, executor); } ConcurrentHandler concurrentHandler = (ConcurrentHandler) this.handler; @@ -232,7 +247,10 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA if (logger.isDebugEnabled()) { logger.debug("endpoint '" + this + "' sending to output channel '" + outputChannel + "', message: " + message); } - outputChannel.send(message); + if (!outputChannel.send(message, this.replyTimeout)) { + this.errorHandler.handle(new MessageDeliveryException(message, + "unable to send output message within alloted timeout of " + replyTimeout + " milliseconds")); + } return null; } try { @@ -291,7 +309,10 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA if (logger.isDebugEnabled()) { logger.debug("endpoint '" + DefaultMessageEndpoint.this + "' replying to channel '" + replyChannel + "' with message: " + replyMessage); } - replyChannel.send(replyMessage); + if (!replyChannel.send(replyMessage, replyTimeout)) { + errorHandler.handle(new MessageDeliveryException(replyMessage, + "unable to send reply message within alloted timeout of " + replyTimeout + " milliseconds")); + } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 7be5bbade8..bf1c2fcd7e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -157,8 +157,8 @@ public class MessageBusTests { bus.registerHandler("handler2", handler2, new Subscription(inputChannel)); bus.start(); inputChannel.send(new StringMessage(1, "testing")); - Message message1 = outputChannel1.receive(200); - Message message2 = outputChannel2.receive(200); + Message message1 = outputChannel1.receive(500); + Message message2 = outputChannel2.receive(500); bus.stop(); assertTrue("both handlers should have received and replied to the message", (message1 != null && message2 != null)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java index 32b4c7806b..d83b32f8ec 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java @@ -38,6 +38,7 @@ import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerNotRunningException; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.selector.MessageSelector; import org.springframework.integration.message.selector.MessageSelectorRejectedException; @@ -442,9 +443,102 @@ public class DefaultMessageEndpointTests { endpoint.stop(); } + @Test + public void testDefaultOutputChannelTimeoutSendsToErrorHandler() { + SimpleChannel output = new SimpleChannel(1); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("output", output); + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() { + public Message handle(Message message) { + return message; + } + }); + endpoint.setDefaultOutputChannelName("output"); + endpoint.setChannelRegistry(channelRegistry); + TestErrorHandler errorHandler = new TestErrorHandler(); + endpoint.setErrorHandler(errorHandler); + endpoint.setReplyTimeout(0); + endpoint.start(); + endpoint.handle(new StringMessage("test1")); + assertNull(errorHandler.getLastError()); + endpoint.handle(new StringMessage("test2")); + Throwable error = errorHandler.getLastError(); + assertNotNull(error); + assertEquals(MessageDeliveryException.class, error.getClass()); + assertEquals("test2", ((MessageDeliveryException) error).getUndeliveredMessage().getPayload()); + } + + @Test + public void testReturnAddressChannelTimeoutSendsToErrorHandler() { + SimpleChannel replyChannel = new SimpleChannel(1); + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() { + public Message handle(Message message) { + return message; + } + }); + TestErrorHandler errorHandler = new TestErrorHandler(); + endpoint.setErrorHandler(errorHandler); + endpoint.setReplyTimeout(0); + endpoint.start(); + Message message1 = new StringMessage("test1"); + message1.getHeader().setReturnAddress(replyChannel); + endpoint.handle(message1); + assertNull(errorHandler.getLastError()); + Message message2 = new StringMessage("test2"); + message2.getHeader().setReturnAddress(replyChannel); + endpoint.handle(message2); + Throwable error = errorHandler.getLastError(); + assertNotNull(error); + assertEquals(MessageDeliveryException.class, error.getClass()); + assertEquals(message2, ((MessageDeliveryException) error).getUndeliveredMessage()); + } + + @Test + public void testReturnAddressChannelNameTimeoutSendsToErrorHandler() { + SimpleChannel replyChannel = new SimpleChannel(1); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("replyChannel", replyChannel); + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() { + public Message handle(Message message) { + return message; + } + }); + endpoint.setChannelRegistry(channelRegistry); + TestErrorHandler errorHandler = new TestErrorHandler(); + endpoint.setErrorHandler(errorHandler); + endpoint.setReplyTimeout(10); + endpoint.start(); + Message message1 = new StringMessage("test1"); + message1.getHeader().setReturnAddress("replyChannel"); + endpoint.handle(message1); + assertNull(errorHandler.getLastError()); + Message message2 = new StringMessage("test2"); + message2.getHeader().setReturnAddress("replyChannel"); + endpoint.handle(message2); + Throwable error = errorHandler.getLastError(); + assertNotNull(error); + assertEquals(MessageDeliveryException.class, error.getClass()); + assertEquals(message2, ((MessageDeliveryException) error).getUndeliveredMessage()); + } + private static ExecutorService createExecutor() { return new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, new SynchronousQueue()); } + + private static class TestErrorHandler implements ErrorHandler { + + private volatile Throwable lastError; + + + public void handle(Throwable t) { + this.lastError = t; + } + + Throwable getLastError() { + return this.lastError; + } + } + }