From 00bca2d5117fb534e6ff2466fd40973a9328e596 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Sun, 4 Jul 2010 19:47:12 +0000 Subject: [PATCH] INT-1225. Set RRLBS as a default load-balancing strategy at the instance level of Direct/ExecutorChannel. Adjusted some of the tests --- .../integration/channel/DirectChannel.java | 7 +++--- .../integration/channel/ExecutorChannel.java | 3 ++- .../config/xml/PointToPointChannelParser.java | 19 +++++++-------- .../dispatcher/UnicastingDispatcher.java | 23 ++++++++++++++----- .../channel/DirectChannelParserTests.java | 6 +++++ .../channel/DirectChannelTests.java | 9 ++++++++ .../channel/ExecutorChannelTests.java | 2 +- .../annotation/SubscriberOrderTests.java | 6 ++++- 8 files changed, 52 insertions(+), 23 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java index 2848b64d5a..6086b147af 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java @@ -17,6 +17,7 @@ package org.springframework.integration.channel; import org.springframework.integration.dispatcher.LoadBalancingStrategy; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; import org.springframework.integration.dispatcher.UnicastingDispatcher; /** @@ -26,6 +27,7 @@ import org.springframework.integration.dispatcher.UnicastingDispatcher; * @author Dave Syer * @author Mark Fisher * @author Iwein Fuld + * @author Oleg Zhurakousky */ public class DirectChannel extends AbstractSubscribableChannel { @@ -33,11 +35,10 @@ public class DirectChannel extends AbstractSubscribableChannel { /** - * Create a channel with no {@link LoadBalancingStrategy}. - * The dispatcher for such a channel will invoke its - * MessageHandlers in a fixed-order. + * Create a channel with default {@link RoundRobinLoadBalancingStrategy} */ public DirectChannel() { + this(new RoundRobinLoadBalancingStrategy()); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java index 9e9c1e5223..f50037865f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java @@ -20,6 +20,7 @@ import java.util.concurrent.Executor; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.dispatcher.LoadBalancingStrategy; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.integration.util.ErrorHandlingTaskExecutor; import org.springframework.util.Assert; @@ -58,7 +59,7 @@ public class ExecutorChannel extends AbstractSubscribableChannel { * The Executor must not be null. */ public ExecutorChannel(Executor executor) { - this(executor, null); + this(executor, new RoundRobinLoadBalancingStrategy()); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java index 5632ada887..600ee400e6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java @@ -30,12 +30,13 @@ import org.w3c.dom.Element; * * @author Mark Fisher * @author Iwein Fuld + * @author Oleg Zhurakousky */ public class PointToPointChannelParser extends AbstractChannelParser { private static final String CHANNEL_PACKAGE = IntegrationNamespaceUtils.BASE_PACKAGE + ".channel"; - private static final String DISPATCHER_PACKAGE = IntegrationNamespaceUtils.BASE_PACKAGE + ".dispatcher"; + //private static final String DISPATCHER_PACKAGE = IntegrationNamespaceUtils.BASE_PACKAGE + ".dispatcher"; private static final String STORE_PACKAGE = IntegrationNamespaceUtils.BASE_PACKAGE + ".store"; @@ -111,17 +112,14 @@ public class PointToPointChannelParser extends AbstractChannelParser { // this attribute is deprecated, but if set, we need to create a DirectChannel // without any LoadBalancerStrategy and the failover flag set to true (default). builder = BeanDefinitionBuilder.genericBeanDefinition(CHANNEL_PACKAGE + ".DirectChannel"); - if (!"failover".equals(dispatcherAttribute)) { + if ("failover".equals(dispatcherAttribute)) { // round-robin dispatcher is used by default, the "failover" value simply disables it - builder.addConstructorArgValue(new RootBeanDefinition(DISPATCHER_PACKAGE - + ".RoundRobinLoadBalancingStrategy", null, null)); - } + builder.addConstructorArgValue(null); + } } else if (dispatcherElement == null) { // configure the default DirectChannel with a RoundRobinLoadBalancingStrategy builder = BeanDefinitionBuilder.genericBeanDefinition(CHANNEL_PACKAGE + ".DirectChannel"); - builder.addConstructorArgValue(new RootBeanDefinition(DISPATCHER_PACKAGE - + ".RoundRobinLoadBalancingStrategy", null, null)); } else { // configure either an ExecutorChannel or DirectChannel based on existence of 'task-executor' @@ -136,10 +134,9 @@ public class PointToPointChannelParser extends AbstractChannelParser { // unless the 'load-balancer' attribute is explicitly set to 'none', // configure the default RoundRobinLoadBalancingStrategy String loadBalancer = dispatcherElement.getAttribute("load-balancer"); - if (!"none".equals(loadBalancer)) { - builder.addConstructorArgValue(new RootBeanDefinition(DISPATCHER_PACKAGE - + ".RoundRobinLoadBalancingStrategy", null, null)); - } + if ("none".equals(loadBalancer)) { + builder.addConstructorArgValue(null); + } IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, dispatcherElement, "failover"); } return builder; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java index ba0371f641..9dde9e7590 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java @@ -19,12 +19,14 @@ import java.util.ArrayList; import java.util.Iterator; import java.util.List; import java.util.concurrent.Executor; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessagingException; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageHandler; -import org.springframework.util.Assert; /** * Implementation of {@link MessageDispatcher} that will attempt to send a @@ -43,12 +45,13 @@ import org.springframework.util.Assert; * @author Iwein Fuld * @author Mark Fisher * @author Gary Russell + * @author Oleg Zhurakousky * @since 1.0.2 */ public class UnicastingDispatcher extends AbstractDispatcher { private volatile boolean failover = true; - + private ReadWriteLock rwLock = new ReentrantReadWriteLock(); private volatile LoadBalancingStrategy loadBalancingStrategy; private final Executor executor; @@ -76,8 +79,10 @@ public class UnicastingDispatcher extends AbstractDispatcher { * Provide a {@link LoadBalancingStrategy} for this dispatcher. */ public void setLoadBalancingStrategy(LoadBalancingStrategy loadBalancingStrategy) { - Assert.notNull(loadBalancingStrategy, "loadBalancingStrategy must not be null"); + Lock lock = rwLock.writeLock(); + lock.lock(); this.loadBalancingStrategy = loadBalancingStrategy; + lock.unlock(); } public final boolean dispatch(final Message message) { @@ -127,9 +132,15 @@ public class UnicastingDispatcher extends AbstractDispatcher { * it simply returns the Iterator for the existing handler List. */ private Iterator getHandlerIterator(Message message) { - if (this.loadBalancingStrategy != null) { - return this.loadBalancingStrategy.getHandlerIterator(message, this.getHandlers()); - } + Lock lock = rwLock.readLock(); + lock.lock(); + try { + if (this.loadBalancingStrategy != null) { + return this.loadBalancingStrategy.getHandlerIterator(message, this.getHandlers()); + } + } finally { + lock.unlock(); + } return this.getHandlers().iterator(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelParserTests.java index 467c752afb..ed226e3689 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelParserTests.java @@ -17,14 +17,18 @@ package org.springframework.integration.channel; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import org.junit.Test; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; /** * @author Mark Fisher + * @author Oleg Zhurakousky */ public class DirectChannelParserTests { @@ -34,6 +38,8 @@ public class DirectChannelParserTests { "directChannelParserTests.xml", DirectChannelParserTests.class); Object channel = context.getBean("channel"); assertEquals(DirectChannel.class, channel.getClass()); + DirectFieldAccessor dcAccessor = new DirectFieldAccessor(((DirectChannel)channel).getDispatcher()); + assertTrue(dcAccessor.getPropertyValue("loadBalancingStrategy") instanceof RoundRobinLoadBalancingStrategy); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java index 21f7404ac6..507335d890 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java @@ -24,12 +24,16 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.core.Message; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; +import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher + * @author Oleg Zhurakousky */ public class DirectChannelTests { @@ -41,6 +45,11 @@ public class DirectChannelTests { StringMessage message = new StringMessage("test"); assertTrue(channel.send(message)); assertEquals(Thread.currentThread().getName(), target.threadName); + DirectFieldAccessor channelAccessor = new DirectFieldAccessor(channel); + UnicastingDispatcher dispatcher = (UnicastingDispatcher) channelAccessor.getPropertyValue("dispatcher"); + DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher); + Object loadBalancingStrategy = dispatcherAccessor.getPropertyValue("loadBalancingStrategy"); + assertTrue(loadBalancingStrategy instanceof RoundRobinLoadBalancingStrategy); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java index 4c90c5db9f..69be36c72e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java @@ -130,7 +130,7 @@ public class ExecutorChannelTests { int numberOfMessages = 11; ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor( Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-"))); - ExecutorChannel channel = new ExecutorChannel(taskExecutor); + ExecutorChannel channel = new ExecutorChannel(taskExecutor, null); CountDownLatch latch = new CountDownLatch(numberOfMessages); TestHandler handler1 = new TestHandler(latch); TestHandler handler2 = new TestHandler(latch); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java index a2ca5edfbc..944156e98d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java @@ -23,6 +23,8 @@ import java.util.List; import org.junit.Test; +import org.springframework.beans.factory.config.ConstructorArgumentValues; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.support.GenericApplicationContext; import org.springframework.core.annotation.Order; @@ -70,7 +72,9 @@ public class SubscriberOrderTests { public void directChannelAndFailoverDispatcherWithMultipleCallsPerMethod() { GenericApplicationContext context = new GenericApplicationContext(); context.registerBeanDefinition("postProcessor", new RootBeanDefinition(MessagingAnnotationPostProcessor.class)); - RootBeanDefinition channelDefinition = new RootBeanDefinition(DirectChannel.class); + BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.rootBeanDefinition(DirectChannel.class); + channelBuilder.addConstructorArgValue(null); + RootBeanDefinition channelDefinition = (RootBeanDefinition) channelBuilder.getBeanDefinition(); context.registerBeanDefinition("input", channelDefinition); RootBeanDefinition testBeanDefinition = new RootBeanDefinition(TestBean.class); testBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue(2);