INT-1225. Set RRLBS as a default load-balancing strategy at the instance level of Direct/ExecutorChannel. Adjusted some of the tests

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());
 	}
 
 	/**

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());
 	}
 
 	/**

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;

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<MessageHandler> 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();
 	}
 

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);
 	}
 
 }

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

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);

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);