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

Added a 'replyTimeout' property to DefaultMessageEndpoint with a default value of 1000 milliseconds. When sending to either the 'defaultOutputChannelName' or the reply channel specified by the input MessageHeader's 'returnAddress', a timeout will now cause the reply Message to be wrapped within a MessageDeliveryException and passed to the endpoint's ErrorHandler. The default error handler publishes the exception as an ErrorMessage to the global error channel (INT-108).

Browse Source
This commit is contained in:
Mark Fisher
2008-02-25 20:55:24 +00:00
parent af229e3576
commit f9d01dd568
3 changed files with 123 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java
View File

@@ -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 <code>replyTimeout</code> 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<Runnable> queue = (capacity < 1) ? new SynchronousQueue<Runnable>() :
new ArrayBlockingQueue<Runnable>(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"));
}
}
}

4
spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
View File

@@ -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));

94
spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java
View File

@@ -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<Runnable>());
}
private static class TestErrorHandler implements ErrorHandler {
private volatile Throwable lastError;
public void handle(Throwable t) {
this.lastError = t;
}
Throwable getLastError() {
return this.lastError;
}
}
}