diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java index d4cb92a818..41df197255 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java @@ -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 Message doSendAndReceive(MessageChannel channel, Message 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 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; } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java index 669b81c302..9c3ce73aae 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java @@ -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("foo")); + assertNull(result); + Message message = channel.receive(); + try { + ((MessageChannel) message.getHeaders().getReplyChannel()).send(new GenericMessage("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 replyChannel = new AtomicReference(); + 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("foo")); + fail("Exception expected"); + } + catch (MessagingException e) { + assertTrue(e.getMessage().equals("foo")); + } + try { + replyChannel.get().send(new GenericMessage("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 replyChannel = new AtomicReference(); + 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("bar")); + } + }); + template.setDefaultChannel(channel); + template.setReceiveTimeout(10000); + template.setThrowExceptionOnLateReply(true); + Message reply = template.sendAndReceive(new GenericMessage("foo")); + assertTrue(reply.getPayload().equals("bar")); + try { + replyChannel.get().send(new GenericMessage("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, OutboundMessageMapper {