diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java index 4b8bc462ad..159cba39c0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java @@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit; import org.springframework.beans.factory.BeanNameAware; import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.message.Message; +import org.springframework.util.Assert; /** * Simple implementation of a message channel. Each {@link Message} is @@ -33,7 +34,7 @@ import org.springframework.integration.message.Message; */ public class SimpleChannel implements MessageChannel, BeanNameAware { - private static final int DEFAULT_CAPACITY = 25; + private static final int DEFAULT_CAPACITY = 100; private String name; @@ -110,6 +111,7 @@ public class SimpleChannel implements MessageChannel, BeanNameAware { * false if the sending thread is interrupted. */ public boolean send(Message message) { + Assert.notNull(message, "'message' must not be null"); try { queue.put(message); return true; @@ -134,6 +136,7 @@ public class SimpleChannel implements MessageChannel, BeanNameAware { * time or the sending thread is interrupted. */ public boolean send(Message message, long timeout) { + Assert.notNull(message, "'message' must not be null"); try { if (timeout > 0) { return this.queue.offer(message, timeout, TimeUnit.MILLISECONDS); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java index ab2f7176e7..945e4e4417 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java @@ -29,7 +29,9 @@ import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.handler.MessageHandlerNotRunningException; import org.springframework.integration.message.Message; +import org.springframework.integration.message.selector.MessageSelectorRejectedException; import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.Schedule; import org.springframework.util.Assert; 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 b8a0e05c2d..e876cb8eb8 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 @@ -28,15 +28,18 @@ import org.springframework.integration.bus.Subscription; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.dispatcher.MessageHandlerNotRunningException; -import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; -import org.springframework.integration.dispatcher.MessageSelectorRejectedException; import org.springframework.integration.handler.ConcurrentHandler; import org.springframework.integration.handler.MessageHandler; +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.MessageHeader; import org.springframework.integration.message.selector.MessageSelector; +import org.springframework.integration.message.selector.MessageSelectorRejectedException; import org.springframework.integration.util.ErrorHandler; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * Default implementation of the {@link MessageEndpoint} interface. @@ -57,6 +60,8 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA private ErrorHandler errorHandler; + private ReplyHandler replyHandler = new EndpointReplyHandler(); + private String defaultOutputChannelName; private ChannelRegistry channelRegistry; @@ -145,7 +150,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA } public void afterPropertiesSet() { - if (this.concurrencyPolicy != null) { + if (this.concurrencyPolicy != null || this.handler instanceof ConcurrentHandler) { if (!(this.handler instanceof ConcurrentHandler)) { this.handler = new ConcurrentHandler(this.handler, this.concurrencyPolicy); } @@ -153,6 +158,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA if (this.errorHandler != null) { concurrentHandler.setErrorHandler(this.errorHandler); } + concurrentHandler.setReplyHandler(this.replyHandler); concurrentHandler.afterPropertiesSet(); } this.initialized = true; @@ -206,13 +212,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA try { Message replyMessage = handler.handle(message); if (replyMessage != null) { - MessageChannel replyChannel = this.resolveReplyChannel(message); - if (replyChannel == null) { - throw new MessageHandlingException("Unable to determine reply channel for message. " - + "Provide a 'replyChannelName' in the message header or a 'defaultOutputChannelName' " - + "on the message endpoint."); - } - replyChannel.send(replyMessage); + this.replyHandler.handle(replyMessage, message.getHeader()); } } catch (MessageHandlerRejectedExecutionException e) { @@ -228,12 +228,11 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA return null; } - private MessageChannel resolveReplyChannel(Message message) { + private MessageChannel resolveReplyChannel(String replyChannelName) { if (this.channelRegistry == null) { return null; } - String replyChannelName = message.getHeader().getReplyChannelName(); - if (replyChannelName != null && replyChannelName.trim().length() > 0) { + if (StringUtils.hasText(replyChannelName)) { return this.channelRegistry.lookupChannel(replyChannelName); } if (this.defaultOutputChannelName != null) { @@ -242,4 +241,22 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA return null; } + + private class EndpointReplyHandler implements ReplyHandler { + + public void handle(Message replyMessage, MessageHeader originalMessageHeader) { + if (replyMessage == null) { + return; + } + String replyChannelName = originalMessageHeader.getReplyChannelName(); + MessageChannel replyChannel = resolveReplyChannel(replyChannelName); + if (replyChannel == null) { + throw new MessageHandlingException("Unable to determine reply channel for message. " + + "Provide a 'replyChannelName' in the message header or a 'defaultOutputChannelName' " + + "on the message endpoint."); + } + replyChannel.send(replyMessage); + } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java index 3e41d021c6..d6c0a04b43 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java @@ -21,8 +21,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.Lifecycle; -import org.springframework.integration.dispatcher.MessageHandlerNotRunningException; -import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.message.Message; import org.springframework.integration.util.ErrorHandler; @@ -51,6 +49,8 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin private ErrorHandler errorHandler; + private ReplyHandler replyHandler; + private volatile boolean running; private Object lifecycleMonitor = new Object(); @@ -83,6 +83,10 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin this.errorHandler = errorHandler; } + public void setReplyHandler(ReplyHandler replyHandler) { + this.replyHandler = replyHandler; + } + public void afterPropertiesSet() { refreshExecutor(); } @@ -173,7 +177,10 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin public void run() { try { - handler.handle(this.message); + Message reply = handler.handle(this.message); + if (replyHandler != null) { + replyHandler.handle(reply, this.message.getHeader()); + } } catch (Throwable t) { if (errorHandler != null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerNotRunningException.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerNotRunningException.java similarity index 94% rename from spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerNotRunningException.java rename to spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerNotRunningException.java index 0d63fa7fed..b72b582502 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerNotRunningException.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerNotRunningException.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dispatcher; +package org.springframework.integration.handler; import org.springframework.integration.MessageHandlingException; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerRejectedExecutionException.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerRejectedExecutionException.java similarity index 96% rename from spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerRejectedExecutionException.java rename to spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerRejectedExecutionException.java index 07c3b76d7e..ebd4ac608b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerRejectedExecutionException.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerRejectedExecutionException.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dispatcher; +package org.springframework.integration.handler; import org.springframework.integration.MessageHandlingException; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyHandler.java new file mode 100644 index 0000000000..48d155ed2f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyHandler.java @@ -0,0 +1,31 @@ +/* + * Copyright 2002-2007 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.handler; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHeader; + +/** + * Strategy interface for handling reply messages. + * + * @author Mark Fisher + */ +public interface ReplyHandler { + + void handle(Message replyMessage, MessageHeader originalMessageHeader); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageSelectorRejectedException.java b/spring-integration-core/src/main/java/org/springframework/integration/message/selector/MessageSelectorRejectedException.java similarity index 95% rename from spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageSelectorRejectedException.java rename to spring-integration-core/src/main/java/org/springframework/integration/message/selector/MessageSelectorRejectedException.java index b67fd89132..d6f1b3d8b6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageSelectorRejectedException.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/selector/MessageSelectorRejectedException.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dispatcher; +package org.springframework.integration.message.selector; import org.springframework.integration.MessageHandlingException; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java index e25623cbff..6ea86e0dcb 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java @@ -120,6 +120,7 @@ public class ChannelParserTests { assertEquals(DispatcherPolicy.DEFAULT_RECEIVE_TIMEOUT, dispatcherPolicy.getReceiveTimeout()); assertEquals(DispatcherPolicy.DEFAULT_REJECTION_LIMIT, dispatcherPolicy.getRejectionLimit()); assertEquals(DispatcherPolicy.DEFAULT_RETRY_INTERVAL, dispatcherPolicy.getRetryInterval()); + assertTrue(dispatcherPolicy.getShouldFailOnRejectionLimit()); } @Test @@ -133,6 +134,7 @@ public class ChannelParserTests { assertEquals(77, dispatcherPolicy.getReceiveTimeout()); assertEquals(777, dispatcherPolicy.getRejectionLimit()); assertEquals(7777, dispatcherPolicy.getRetryInterval()); + assertFalse(dispatcherPolicy.getShouldFailOnRejectionLimit()); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml index 4ab0159f98..6e06361a2c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml @@ -16,7 +16,11 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java new file mode 100644 index 0000000000..30cdc53d2a --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java @@ -0,0 +1,100 @@ +/* + * Copyright 2002-2007 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.dispatcher; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.Collection; +import java.util.Iterator; + +import org.junit.Test; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class ChannelPollingMessageRetrieverTests { + + @Test + public void testSingleMessagePerRetrieval() { + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setReceiveTimeout(0); + MessageChannel channel = new SimpleChannel(dispatcherPolicy); + ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); + Collection> results = retriever.retrieveMessages(); + assertTrue(results.isEmpty()); + channel.send(new StringMessage("test1")); + channel.send(new StringMessage("test2")); + results = retriever.retrieveMessages(); + assertEquals(1, results.size()); + assertEquals("test1", results.iterator().next().getPayload()); + results = retriever.retrieveMessages(); + assertEquals(1, results.size()); + assertEquals("test2", results.iterator().next().getPayload()); + } + + @Test + public void testMultipleMessagesPerRetrieval() { + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setReceiveTimeout(0); + dispatcherPolicy.setMaxMessagesPerTask(2); + MessageChannel channel = new SimpleChannel(dispatcherPolicy); + ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); + Collection> results = retriever.retrieveMessages(); + assertTrue(results.isEmpty()); + channel.send(new StringMessage("test1")); + channel.send(new StringMessage("test2")); + channel.send(new StringMessage("test3")); + results = retriever.retrieveMessages(); + assertEquals(2, results.size()); + Iterator> iter = results.iterator(); + assertEquals("test1", iter.next().getPayload()); + assertEquals("test2", iter.next().getPayload()); + results = retriever.retrieveMessages(); + assertEquals(1, results.size()); + assertEquals("test3", results.iterator().next().getPayload()); + } + + @Test + public void testMaxMessagesConfiguredDynamically() { + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setReceiveTimeout(0); + dispatcherPolicy.setMaxMessagesPerTask(1); + MessageChannel channel = new SimpleChannel(dispatcherPolicy); + ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); + Collection> results = retriever.retrieveMessages(); + assertTrue(results.isEmpty()); + channel.send(new StringMessage("test1")); + channel.send(new StringMessage("test2")); + channel.send(new StringMessage("test3")); + results = retriever.retrieveMessages(); + assertEquals(1, results.size()); + assertEquals("test1", results.iterator().next().getPayload()); + channel.getDispatcherPolicy().setMaxMessagesPerTask(5); + results = retriever.retrieveMessages(); + assertEquals(2, results.size()); + Iterator> iter = results.iterator(); + assertEquals("test2", iter.next().getPayload()); + assertEquals("test3", iter.next().getPayload()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java index f0b6872449..db1d6bfedc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java @@ -29,12 +29,12 @@ import org.junit.Test; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.dispatcher.DefaultMessageDispatcher; -import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.ConcurrentHandler; import org.springframework.integration.handler.InterceptingMessageHandler; import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.Message; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DispatcherTaskTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DispatcherTaskTests.java new file mode 100644 index 0000000000..994333794b --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DispatcherTaskTests.java @@ -0,0 +1,79 @@ +/* + * Copyright 2002-2007 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.dispatcher; + +import static org.junit.Assert.assertEquals; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.handler.TestHandlers; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class DispatcherTaskTests { + + @Test + public void testSimpleDispatch() throws InterruptedException { + MessageChannel channel = new SimpleChannel(); + DispatcherTask task = new DispatcherTask(channel); + final CountDownLatch latch = new CountDownLatch(1); + task.addHandler(TestHandlers.countDownHandler(latch)); + task.dispatchMessage(new StringMessage("test")); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + } + + @Test + public void testDispatchWithPointToPointChannel() throws InterruptedException { + MessageChannel channel = new SimpleChannel(new DispatcherPolicy(false)); + DispatcherTask task = new DispatcherTask(channel); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicInteger counter1 = new AtomicInteger(); + final AtomicInteger counter2 = new AtomicInteger(); + task.addHandler(TestHandlers.countingCountDownHandler(counter1, latch)); + task.addHandler(TestHandlers.countingCountDownHandler(counter2, latch)); + task.dispatchMessage(new StringMessage("test")); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertEquals("only 1 handler should have received the message", 1, counter1.get() + counter2.get()); + } + + @Test + public void testDispatchWithPublishSubscribeChannel() throws InterruptedException { + MessageChannel channel = new SimpleChannel(new DispatcherPolicy(true)); + DispatcherTask task = new DispatcherTask(channel); + final CountDownLatch latch = new CountDownLatch(2); + final AtomicInteger counter1 = new AtomicInteger(); + final AtomicInteger counter2 = new AtomicInteger(); + task.addHandler(TestHandlers.countingCountDownHandler(counter1, latch)); + task.addHandler(TestHandlers.countingCountDownHandler(counter2, latch)); + task.dispatchMessage(new StringMessage("test")); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertEquals(1, counter1.get()); + assertEquals(1, counter2.get()); + } + +} 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 71170c8539..dbe4c420de 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 @@ -18,6 +18,12 @@ package org.springframework.integration.endpoint; 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 java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; @@ -25,9 +31,15 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.handler.ConcurrentHandler; 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.StringMessage; +import org.springframework.integration.message.selector.MessageSelector; +import org.springframework.integration.message.selector.MessageSelectorRejectedException; +import org.springframework.integration.util.ErrorHandler; /** * @author Mark Fisher @@ -79,4 +91,316 @@ public class DefaultMessageEndpointTests { assertEquals("hello test", reply.getPayload()); } + @Test + public void testCustomErrorHandler() throws InterruptedException { + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); + final CountDownLatch latch = new CountDownLatch(2); + endpoint.setErrorHandler(new ErrorHandler() { + public void handle(Throwable t) { + latch.countDown(); + } + }); + endpoint.setHandler(TestHandlers.rejectingCountDownHandler(latch)); + endpoint.start(); + endpoint.handle(new StringMessage("test")); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("both handler and errorHandler should have been invoked", 0, latch.getCount()); + } + + @Test + public void testConcurrentHandlerWithDefaultReplyChannel() throws InterruptedException { + MessageChannel replyChannel = new SimpleChannel(); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("replyChannel", replyChannel); + final CountDownLatch latch = new CountDownLatch(1); + MessageHandler handler = new MessageHandler() { + public Message handle(Message message) { + latch.countDown(); + return new StringMessage("123", "hello " + message.getPayload()); + } + }; + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.setChannelRegistry(channelRegistry); + endpoint.setHandler(new ConcurrentHandler(handler)); + endpoint.setDefaultOutputChannelName("replyChannel"); + endpoint.start(); + endpoint.handle(new StringMessage(1, "test")); + endpoint.stop(); + latch.await(200, TimeUnit.MILLISECONDS); + assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); + Message reply = replyChannel.receive(100); + assertNotNull(reply); + assertEquals("hello test", reply.getPayload()); + } + + @Test + public void testHandlerReturnsNull() throws InterruptedException { + MessageChannel replyChannel = new SimpleChannel(); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("replyChannel", replyChannel); + final CountDownLatch latch = new CountDownLatch(1); + MessageHandler handler = new MessageHandler() { + public Message handle(Message message) { + latch.countDown(); + return null; + } + }; + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.setChannelRegistry(channelRegistry); + endpoint.setHandler(handler); + endpoint.setDefaultOutputChannelName("replyChannel"); + endpoint.start(); + endpoint.handle(new StringMessage(1, "test")); + endpoint.stop(); + latch.await(200, TimeUnit.MILLISECONDS); + assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); + Message reply = replyChannel.receive(0); + assertNull(reply); + } + + @Test + public void testConcurrentHandlerReturnsNull() throws InterruptedException { + MessageChannel replyChannel = new SimpleChannel(); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("replyChannel", replyChannel); + final CountDownLatch latch = new CountDownLatch(1); + MessageHandler handler = new MessageHandler() { + public Message handle(Message message) { + latch.countDown(); + return null; + } + }; + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.setChannelRegistry(channelRegistry); + endpoint.setHandler(new ConcurrentHandler(handler)); + endpoint.setDefaultOutputChannelName("replyChannel"); + endpoint.start(); + endpoint.handle(new StringMessage(1, "test")); + endpoint.stop(); + latch.await(200, TimeUnit.MILLISECONDS); + assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); + Message reply = replyChannel.receive(0); + assertNull(reply); + } + + @Test + public void testConcurrentHandlerWithExplicitReplyChannel() throws InterruptedException { + MessageChannel replyChannel = new SimpleChannel(); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("replyChannel", replyChannel); + final CountDownLatch latch = new CountDownLatch(1); + MessageHandler handler = new MessageHandler() { + public Message handle(Message message) { + latch.countDown(); + return new StringMessage("123", "hello " + message.getPayload()); + } + }; + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.setChannelRegistry(channelRegistry); + endpoint.setHandler(new ConcurrentHandler(handler)); + endpoint.start(); + StringMessage message = new StringMessage(1, "test"); + message.getHeader().setReplyChannelName("replyChannel"); + endpoint.handle(message); + endpoint.stop(); + latch.await(200, TimeUnit.MILLISECONDS); + assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); + Message reply = replyChannel.receive(100); + assertNotNull(reply); + assertEquals("hello test", reply.getPayload()); + } + + @Test + public void testGeneratedConcurrentHandlerWithDefaultReplyChannel() throws InterruptedException { + MessageChannel replyChannel = new SimpleChannel(); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("replyChannel", replyChannel); + final CountDownLatch latch = new CountDownLatch(1); + MessageHandler handler = new MessageHandler() { + public Message handle(Message message) { + latch.countDown(); + return new StringMessage("123", "hello " + message.getPayload()); + } + }; + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.setChannelRegistry(channelRegistry); + endpoint.setHandler(handler); + endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(3, 14)); + endpoint.setDefaultOutputChannelName("replyChannel"); + endpoint.start(); + endpoint.handle(new StringMessage(1, "test")); + endpoint.stop(); + latch.await(200, TimeUnit.MILLISECONDS); + assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); + Message reply = replyChannel.receive(100); + assertNotNull(reply); + assertEquals("hello test", reply.getPayload()); + } + + @Test + public void testGeneratedConcurrentHandlerWithExplicitReplyChannel() throws InterruptedException { + MessageChannel replyChannel = new SimpleChannel(); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("replyChannel", replyChannel); + final CountDownLatch latch = new CountDownLatch(1); + MessageHandler handler = new MessageHandler() { + public Message handle(Message message) { + latch.countDown(); + return new StringMessage("123", "hello " + message.getPayload()); + } + }; + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.setChannelRegistry(channelRegistry); + endpoint.setHandler(handler); + endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(3, 14)); + endpoint.start(); + StringMessage message = new StringMessage(1, "test"); + message.getHeader().setReplyChannelName("replyChannel"); + endpoint.handle(message); + endpoint.stop(); + latch.await(200, TimeUnit.MILLISECONDS); + assertEquals("handler should have been invoked within allotted time", 0, latch.getCount()); + Message reply = replyChannel.receive(100); + assertNotNull(reply); + assertEquals("hello test", reply.getPayload()); + } + + @Test(expected=MessageHandlerNotRunningException.class) + public void testEndpointDoesNotHandleMessagesWhenNotYetStarted() { + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.handle(new StringMessage("test")); + } + + @Test + public void testEndpointDoesNotHandleMessagesAfterBeingStopped() { + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + AtomicInteger counter = new AtomicInteger(); + boolean exceptionThrown = false; + try { + endpoint.start(); + endpoint.setHandler(TestHandlers.countingHandler(counter)); + endpoint.handle(new StringMessage("test1")); + endpoint.stop(); + endpoint.handle(new StringMessage("test2")); + } + catch (MessageHandlerNotRunningException e) { + exceptionThrown = true; + } + assertEquals("handler should have been invoked exactly once", 1, counter.get()); + assertTrue(exceptionThrown); + } + + @Test(expected=MessageSelectorRejectedException.class) + public void testEndpointWithSelectorRejecting() { + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.addMessageSelector(new MessageSelector() { + public boolean accept(Message message) { + return false; + } + }); + endpoint.start(); + endpoint.handle(new StringMessage("test")); + } + + @Test + public void testEndpointWithSelectorAccepting() throws InterruptedException { + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + endpoint.addMessageSelector(new MessageSelector() { + public boolean accept(Message message) { + return true; + } + }); + CountDownLatch latch = new CountDownLatch(1); + endpoint.setHandler(TestHandlers.countDownHandler(latch)); + endpoint.start(); + endpoint.handle(new StringMessage("test")); + latch.await(100, TimeUnit.MILLISECONDS); + assertEquals("handler should have been invoked", 0, latch.getCount()); + endpoint.stop(); + } + + @Test + public void testEndpointWithMultipleSelectorsAndFirstRejects() { + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + boolean exceptionThrown = false; + final AtomicInteger counter = new AtomicInteger(); + endpoint.addMessageSelector(new MessageSelector() { + public boolean accept(Message message) { + counter.incrementAndGet(); + return false; + } + }); + endpoint.addMessageSelector(new MessageSelector() { + public boolean accept(Message message) { + counter.incrementAndGet(); + return true; + } + }); + endpoint.setHandler(TestHandlers.countingHandler(counter)); + endpoint.start(); + try { + endpoint.handle(new StringMessage("test")); + } + catch (MessageSelectorRejectedException e) { + exceptionThrown = true; + } + assertEquals("only the first selector should have been invoked", 1, counter.get()); + assertTrue(exceptionThrown); + endpoint.stop(); + } + + @Test + public void testEndpointWithMultipleSelectorsAndFirstAccepts() { + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + boolean exceptionThrown = false; + final AtomicInteger counter = new AtomicInteger(); + endpoint.addMessageSelector(new MessageSelector() { + public boolean accept(Message message) { + counter.incrementAndGet(); + return true; + } + }); + endpoint.addMessageSelector(new MessageSelector() { + public boolean accept(Message message) { + counter.incrementAndGet(); + return false; + } + }); + endpoint.setHandler(TestHandlers.countingHandler(counter)); + endpoint.start(); + try { + endpoint.handle(new StringMessage("test")); + } + catch (MessageSelectorRejectedException e) { + exceptionThrown = true; + } + assertEquals("both selectors should have been invoked but not the handler", 2, counter.get()); + assertTrue(exceptionThrown); + endpoint.stop(); + } + + @Test + public void testEndpointWithMultipleSelectorsAndBothAccept() { + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); + final AtomicInteger counter = new AtomicInteger(); + endpoint.addMessageSelector(new MessageSelector() { + public boolean accept(Message message) { + counter.incrementAndGet(); + return true; + } + }); + endpoint.addMessageSelector(new MessageSelector() { + public boolean accept(Message message) { + counter.incrementAndGet(); + return true; + } + }); + endpoint.setHandler(TestHandlers.countingHandler(counter)); + endpoint.start(); + endpoint.handle(new StringMessage("test")); + assertEquals("both selectors and handler should have been invoked", 3, counter.get()); + endpoint.stop(); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java index f0d115b033..9e0de65ded 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java @@ -19,7 +19,6 @@ package org.springframework.integration.handler; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicInteger; -import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; import org.springframework.integration.message.Message; /**