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

INT-3004 Detect Timeout in MessagingTemplate

Browse Source 
https://jira.springsource.org/browse/INT-3004

Previously, reply messages were silently ignored if the request thread
times out or caught an exception, or had already handled a reply.

Add a WARN log if the client times out or catches an exception,
or has already received a reply.

Optionally make these fatal; at this time, we are not exposing
this boolean on the namespaces see INT-3005.

Polishing - PR Comments

INT-3004 Polishing

Change MessagingTemplate property to throwExceptionOnLateReply.
This commit is contained in:
Gary Russell
2013-05-03 13:57:03 -04:00
committed by Mark Fisher
parent 698cb10fe4
commit 91f040f1c7
2 changed files with 162 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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.
@@ -21,6 +21,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
@@ -41,9 +42,10 @@ import org.springframework.util.Assert;
* This is the central class for invoking message exchange operations across
* {@link MessageChannel}s. It supports one-way send and receive calls as well
* as request/reply.
*
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
*/
public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, InitializingBean {
@@ -63,6 +65,8 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware,
private final Object initializationMonitor = new Object();
private volatile boolean throwExceptionOnLateReply = false;
/**
* Create a MessagingTemplate with no default channel. Note, that one
@@ -110,7 +114,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware,
/**
* Specify the timeout value to use for send operations.
*
*
* @param sendTimeout the send timeout in milliseconds
*/
public void setSendTimeout(long sendTimeout) {
@@ -119,7 +123,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware,
/**
* Specify the timeout value to use for receive operations.
*
*
* @param receiveTimeout the receive timeout in milliseconds
*/
public void setReceiveTimeout(long receiveTimeout) {
@@ -132,6 +136,18 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware,
}
}
/**
* Specify whether or not an attempt to send on the reply channel throws an exception
* if no receiving thread will actually receive the reply. This can occur
* if the receiving thread has already timed out, or will never call receive()
* because it caught an exception, or has already received a reply.
* (default false - just a WARN log is emitted in these cases).
* @param throwExceptionOnLateReply TRUE or FALSE.
*/
public void setThrowExceptionOnLateReply(boolean throwExceptionOnLateReply) {
this.throwExceptionOnLateReply = throwExceptionOnLateReply;
}
public void afterPropertiesSet() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
@@ -310,12 +326,18 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware,
private <S, R> Message<R> doSendAndReceive(MessageChannel channel, Message<S> requestMessage) {
Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel();
Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel();
TemporaryReplyChannel replyChannel = new TemporaryReplyChannel(this.receiveTimeout);
TemporaryReplyChannel replyChannel = new TemporaryReplyChannel(this.receiveTimeout, this.throwExceptionOnLateReply);
requestMessage = MessageBuilder.fromMessage(requestMessage)
.setReplyChannel(replyChannel)
.setErrorChannel(replyChannel)
.build();
this.doSend(channel, requestMessage);
try {
this.doSend(channel, requestMessage);
}
catch (RuntimeException e) {
replyChannel.setClientWontReceive(true);
throw e;
}
Message<R> reply = this.doReceive(replyChannel);
if (reply != null) {
reply = MessageBuilder.fromMessage(reply)
@@ -356,15 +378,30 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware,
private static class TemporaryReplyChannel implements PollableChannel {
private static final Log logger = LogFactory.getLog(TemporaryReplyChannel.class);
private volatile Message<?> message;
private final long receiveTimeout;
private final CountDownLatch latch = new CountDownLatch(1);
private final boolean throwExceptionOnLateReply;
public TemporaryReplyChannel(long receiveTimeout) {
private volatile boolean clientTimedOut;
private volatile boolean clientWontReceive;
private volatile boolean clientHasReceived;
public TemporaryReplyChannel(long receiveTimeout, boolean throwExceptionOnLateReply) {
this.receiveTimeout = receiveTimeout;
this.throwExceptionOnLateReply = throwExceptionOnLateReply;
}
public void setClientWontReceive(boolean clientWontReceive) {
this.clientWontReceive = clientWontReceive;
}
@@ -376,9 +413,15 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware,
try {
if (this.receiveTimeout < 0) {
this.latch.await();
this.clientHasReceived = true;
}
else {
this.latch.await(this.receiveTimeout, TimeUnit.MILLISECONDS);
if (this.latch.await(this.receiveTimeout, TimeUnit.MILLISECONDS)) {
this.clientHasReceived = true;
}
else {
this.clientTimedOut = true;
}
}
}
catch (InterruptedException e) {
@@ -394,6 +437,25 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware,
public boolean send(Message<?> message, long timeout) {
this.message = message;
this.latch.countDown();
if (this.clientTimedOut || this.clientHasReceived || this.clientWontReceive) {
String exceptionMessage = "";
if (this.clientTimedOut) {
exceptionMessage = "Reply message being sent, but the receiving thread has already timed out";
}
else if (this.clientHasReceived) {
exceptionMessage = "Reply message being sent, but the receiving thread has already received a reply";
}
else if (this.clientWontReceive) {
exceptionMessage = "Reply message being sent, but the receiving thread has already caught an exception and won't receive";
}
if (logger.isWarnEnabled()) {
logger.warn(exceptionMessage + ":" + message);
}
if (this.throwExceptionOnLateReply) {
throw new MessageDeliveryException(message, exceptionMessage);
}
}
return true;
}
}

100
spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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.
@@ -20,11 +20,13 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.After;
import org.junit.Before;
@@ -34,6 +36,7 @@ import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -51,10 +54,11 @@ import org.springframework.integration.test.util.TestUtils.TestApplicationContex
/**
* @author Mark Fisher
* @author Gary Russell
*/
public class MessagingTemplateTests {
private TestApplicationContext context = TestUtils.createTestApplicationContext();
private final TestApplicationContext context = TestUtils.createTestApplicationContext();
private QueueChannel requestChannel;
@@ -305,16 +309,16 @@ public class MessagingTemplateTests {
DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory();
beanFactory.registerSingleton("testChannel", testChannel);
MessagingTemplate template = new MessagingTemplate();
template.setBeanFactory(beanFactory);
template.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("test").build();
template.send("testChannel", message);
assertEquals(message, testChannel.receive(0));
template.setChannelResolver(new ChannelResolver() {
template.setChannelResolver(new ChannelResolver() {
public MessageChannel resolveChannelName(String channelName) {
return anotherChannel;
}
@@ -405,7 +409,7 @@ public class MessagingTemplateTests {
template.convertAndSend(channel, "test");
Message<?> reply = channel.receive(0);
assertNotNull(reply);
assertEquals("test", reply.getPayload());
assertEquals("test", reply.getPayload());
}
@Test
@@ -452,7 +456,7 @@ public class MessagingTemplateTests {
template.convertAndSend(channel, "test");
Message<?> reply = channel.receive(0);
assertNotNull(reply);
assertEquals("to:test", reply.getPayload());
assertEquals("to:test", reply.getPayload());
}
@Test
@@ -524,6 +528,86 @@ public class MessagingTemplateTests {
assertEquals("from:TO:TEST", result);
}
@Test
public void testLateReply() {
MessagingTemplate template = new MessagingTemplate();
QueueChannel channel = new QueueChannel();
template.setDefaultChannel(channel);
template.setReceiveTimeout(1);
template.setThrowExceptionOnLateReply(true);
Object result = template.sendAndReceive(new GenericMessage<String>("foo"));
assertNull(result);
Message<?> message = channel.receive();
try {
((MessageChannel) message.getHeaders().getReplyChannel()).send(new GenericMessage<String>("bar"));
fail("Exception expected");
}
catch (MessagingException e) {
assertEquals("Reply message being sent, but the receiving thread has already timed out", e.getMessage());
}
}
@Test
public void testNeverReceive() {
MessagingTemplate template = new MessagingTemplate();
DirectChannel channel = new DirectChannel();
final AtomicReference<MessageChannel> replyChannel = new AtomicReference<MessageChannel>();
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
replyChannel.set((MessageChannel) message.getHeaders().getReplyChannel());
throw new MessagingException("foo");
}
});
template.setDefaultChannel(channel);
template.setReceiveTimeout(10000);
template.setThrowExceptionOnLateReply(true);
try {
template.sendAndReceive(new GenericMessage<String>("foo"));
fail("Exception expected");
}
catch (MessagingException e) {
assertTrue(e.getMessage().equals("foo"));
}
try {
replyChannel.get().send(new GenericMessage<String>("bar"));
fail("Exception expected");
}
catch (MessagingException e) {
assertEquals(
"Reply message being sent, but the receiving thread has already caught an exception and won't receive",
e.getMessage());
}
}
@Test
public void testTwoReplies() {
MessagingTemplate template = new MessagingTemplate();
DirectChannel channel = new DirectChannel();
final AtomicReference<MessageChannel> replyChannel = new AtomicReference<MessageChannel>();
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
MessageChannel replyChannelHeader = (MessageChannel) message.getHeaders().getReplyChannel();
replyChannel.set(replyChannelHeader);
replyChannelHeader.send(new GenericMessage<String>("bar"));
}
});
template.setDefaultChannel(channel);
template.setReceiveTimeout(10000);
template.setThrowExceptionOnLateReply(true);
Message<?> reply = template.sendAndReceive(new GenericMessage<String>("foo"));
assertTrue(reply.getPayload().equals("bar"));
try {
replyChannel.get().send(new GenericMessage<String>("baz"));
fail("Exception expected");
}
catch (MessagingException e) {
assertEquals(
"Reply message being sent, but the receiving thread has already received a reply",
e.getMessage());
}
}
private static class TestMapper implements InboundMessageMapper<Object>, OutboundMessageMapper<Object> {