diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/ExecutorChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/ExecutorChannel.java index fbb9e3166b..18758284e9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/ExecutorChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/ExecutorChannel.java @@ -18,12 +18,20 @@ package org.springframework.integration.channel; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.dispatcher.LoadBalancingStrategy; import org.springframework.integration.dispatcher.UnicastingDispatcher; /** * An implementation of {@link MessageChannel} that delegates to an instance of - * {@link UnicastingDispatcher} and wraps all send invocations within a - * {@link TaskExecutor}. + * {@link UnicastingDispatcher} which in turn delegates all dispatching + * invocations to a {@link TaskExecutor}. + *

+ * NOTE: unlike DirectChannel, the ExecutorChannel does not support a + * shared transactional context between sender and handler, because the + * {@link TaskExecutor} typically does not block the sender's Thread since it + * uses another Thread for the dispatch. (SyncTaskExecutor is an + * exception but would provide no value for this channel. If synchronous + * dispatching is required, a DirectChannel should be used instead). * * @author Mark Fisher * @since 1.0.3 @@ -33,10 +41,31 @@ public class ExecutorChannel extends AbstractSubscribableChannel { private final UnicastingDispatcher dispatcher; + /** + * Create an ExecutorChannel that delegates to the provided + * {@link TaskExecutor} when dispatching Messages. + */ public ExecutorChannel(TaskExecutor taskExecutor) { this.dispatcher = new UnicastingDispatcher(taskExecutor); } + /** + * Create an ExecutorChannel with a {@link LoadBalancingStrategy}. The + * strategy must not be null. + */ + public ExecutorChannel(TaskExecutor taskExecutor, LoadBalancingStrategy loadBalancingStrategy) { + this(taskExecutor); + this.dispatcher.setLoadBalancingStrategy(loadBalancingStrategy); + } + + + /** + * Specify whether the channel's dispatcher should have failover enabled. + * By default, it will. Set this value to 'false' to disable it. + */ + public void setFailover(boolean failover) { + this.dispatcher.setFailover(failover); + } @Override protected UnicastingDispatcher getDispatcher() { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java new file mode 100644 index 0000000000..4c90c5db9f --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java @@ -0,0 +1,184 @@ +/* + * Copyright 2002-2009 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.channel; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +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.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.integration.core.Message; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; +import org.springframework.integration.message.MessageHandler; +import org.springframework.integration.message.StringMessage; +import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; + +/** + * @author Mark Fisher + */ +public class ExecutorChannelTests { + + @Test + public void verifyDifferentThread() throws Exception { + SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); + taskExecutor.setThreadNamePrefix("test-"); + ExecutorChannel channel = new ExecutorChannel(taskExecutor); + CountDownLatch latch = new CountDownLatch(1); + TestHandler handler = new TestHandler(latch); + channel.subscribe(handler); + channel.send(new StringMessage("test")); + latch.await(1000, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertNotNull(handler.thread); + assertFalse(Thread.currentThread().equals(handler.thread)); + assertEquals("test-1", handler.thread.getName()); + } + + @Test + public void roundRobinLoadBalancing() throws Exception { + int numberOfMessages = 11; + ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor( + Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-"))); + ExecutorChannel channel = new ExecutorChannel( + taskExecutor, new RoundRobinLoadBalancingStrategy()); + CountDownLatch latch = new CountDownLatch(numberOfMessages); + TestHandler handler1 = new TestHandler(latch); + TestHandler handler2 = new TestHandler(latch); + TestHandler handler3 = new TestHandler(latch); + channel.subscribe(handler1); + channel.subscribe(handler2); + channel.subscribe(handler3); + for (int i = 0; i < numberOfMessages; i++) { + channel.send(new StringMessage("test-" + i)); + } + latch.await(3000, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertNotNull(handler1.thread); + assertFalse(Thread.currentThread().equals(handler1.thread)); + assertTrue(handler1.thread.getName().startsWith("test-")); + assertNotNull(handler2.thread); + assertFalse(Thread.currentThread().equals(handler2.thread)); + assertTrue(handler2.thread.getName().startsWith("test-")); + assertNotNull(handler3.thread); + assertFalse(Thread.currentThread().equals(handler3.thread)); + assertTrue(handler3.thread.getName().startsWith("test-")); + assertEquals(4, handler1.count.get()); + assertEquals(4, handler2.count.get()); + assertEquals(3, handler3.count.get()); + } + + @Test + public void verifyFailoverWithLoadBalancing() throws Exception { + int numberOfMessages = 11; + ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor( + Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-"))); + ExecutorChannel channel = new ExecutorChannel( + taskExecutor, new RoundRobinLoadBalancingStrategy()); + CountDownLatch latch = new CountDownLatch(numberOfMessages); + TestHandler handler1 = new TestHandler(latch); + TestHandler handler2 = new TestHandler(latch); + TestHandler handler3 = new TestHandler(latch); + channel.subscribe(handler1); + channel.subscribe(handler2); + channel.subscribe(handler3); + handler2.shouldFail = true; + for (int i = 0; i < numberOfMessages; i++) { + channel.send(new StringMessage("test-" + i)); + } + latch.await(3000, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertNotNull(handler1.thread); + assertFalse(Thread.currentThread().equals(handler1.thread)); + assertTrue(handler1.thread.getName().startsWith("test-")); + assertNotNull(handler2.thread); + assertFalse(Thread.currentThread().equals(handler2.thread)); + assertTrue(handler2.thread.getName().startsWith("test-")); + assertNotNull(handler3.thread); + assertFalse(Thread.currentThread().equals(handler3.thread)); + assertTrue(handler3.thread.getName().startsWith("test-")); + assertEquals(0, handler2.count.get()); + assertEquals(4, handler1.count.get()); + assertEquals(7, handler3.count.get()); + } + + @Test + public void verifyFailoverWithoutLoadBalancing() throws Exception { + int numberOfMessages = 11; + ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor( + Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-"))); + ExecutorChannel channel = new ExecutorChannel(taskExecutor); + CountDownLatch latch = new CountDownLatch(numberOfMessages); + TestHandler handler1 = new TestHandler(latch); + TestHandler handler2 = new TestHandler(latch); + TestHandler handler3 = new TestHandler(latch); + channel.subscribe(handler1); + channel.subscribe(handler2); + channel.subscribe(handler3); + handler1.shouldFail = true; + for (int i = 0; i < numberOfMessages; i++) { + channel.send(new StringMessage("test-" + i)); + } + latch.await(3000, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertNotNull(handler1.thread); + assertFalse(Thread.currentThread().equals(handler1.thread)); + assertTrue(handler1.thread.getName().startsWith("test-")); + assertNotNull(handler2.thread); + assertFalse(Thread.currentThread().equals(handler2.thread)); + assertTrue(handler2.thread.getName().startsWith("test-")); + assertNull(handler3.thread); + assertEquals(0, handler1.count.get()); + assertEquals(0, handler3.count.get()); + assertEquals(numberOfMessages, handler2.count.get()); + } + + + private static class TestHandler implements MessageHandler { + + private final CountDownLatch latch; + + private final AtomicInteger count = new AtomicInteger(); + + private volatile Thread thread; + + private volatile boolean shouldFail; + + public TestHandler(CountDownLatch latch) { + this.latch = latch; + } + + public void handleMessage(Message message) { + this.thread = Thread.currentThread(); + if (this.shouldFail) { + throw new RuntimeException("intentional test failure"); + } + this.count.incrementAndGet(); + this.latch.countDown(); + } + } + +}