diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java index 6496aea328..2848b64d5a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java @@ -16,9 +16,8 @@ package org.springframework.integration.channel; -import org.springframework.integration.dispatcher.AbstractUnicastDispatcher; -import org.springframework.integration.dispatcher.RoundRobinDispatcher; -import org.springframework.util.Assert; +import org.springframework.integration.dispatcher.LoadBalancingStrategy; +import org.springframework.integration.dispatcher.UnicastingDispatcher; /** * A channel that invokes a single subscriber for each sent Message. @@ -30,21 +29,36 @@ import org.springframework.util.Assert; */ public class DirectChannel extends AbstractSubscribableChannel { - private final AbstractUnicastDispatcher dispatcher; + private final UnicastingDispatcher dispatcher = new UnicastingDispatcher(); + /** + * Create a channel with no {@link LoadBalancingStrategy}. + * The dispatcher for such a channel will invoke its + * MessageHandlers in a fixed-order. + */ public DirectChannel() { - this.dispatcher = new RoundRobinDispatcher(); } - public DirectChannel(AbstractUnicastDispatcher dispatcher) { - Assert.notNull(dispatcher, "dispatcher must not be null"); - this.dispatcher = dispatcher; + /** + * Create a DirectChannel with a {@link LoadBalancingStrategy}. The + * strategy must not be null. + */ + public DirectChannel(LoadBalancingStrategy loadBalancingStrategy) { + 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 AbstractUnicastDispatcher getDispatcher() { + protected UnicastingDispatcher getDispatcher() { return this.dispatcher; } 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 79ff312cd2..fbb9e3166b 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 @@ -16,21 +16,13 @@ package org.springframework.integration.channel; -import java.util.concurrent.Executors; - import org.springframework.core.task.TaskExecutor; -import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.dispatcher.AbstractUnicastDispatcher; -import org.springframework.integration.dispatcher.MessageDispatcher; -import org.springframework.integration.dispatcher.RoundRobinDispatcher; -import org.springframework.integration.message.MessageHandler; -import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; -import org.springframework.util.Assert; +import org.springframework.integration.dispatcher.UnicastingDispatcher; /** * An implementation of {@link MessageChannel} that delegates to an instance of - * {@link AbstractUnicastDispatcher} and wraps all send invocations within a + * {@link UnicastingDispatcher} and wraps all send invocations within a * {@link TaskExecutor}. * * @author Mark Fisher @@ -38,66 +30,17 @@ import org.springframework.util.Assert; */ public class ExecutorChannel extends AbstractSubscribableChannel { - private final ExecutorDecoratingDispatcher dispatcher; + private final UnicastingDispatcher dispatcher; - public ExecutorChannel() { - this(null, null); - } - public ExecutorChannel(TaskExecutor taskExecutor) { - this(null, taskExecutor); - } - - public ExecutorChannel(AbstractUnicastDispatcher dispatcher) { - this(dispatcher, null); - } - - public ExecutorChannel(AbstractUnicastDispatcher dispatcher, TaskExecutor taskExecutor) { - if (dispatcher == null) { - dispatcher = new RoundRobinDispatcher(); - } - this.dispatcher = new ExecutorDecoratingDispatcher(dispatcher, taskExecutor); + this.dispatcher = new UnicastingDispatcher(taskExecutor); } @Override - protected MessageDispatcher getDispatcher() { + protected UnicastingDispatcher getDispatcher() { return this.dispatcher; } - - private static class ExecutorDecoratingDispatcher implements MessageDispatcher { - - private final AbstractUnicastDispatcher targetDispatcher; - - private final TaskExecutor taskExecutor; - - - ExecutorDecoratingDispatcher(AbstractUnicastDispatcher dispatcher, TaskExecutor taskExecutor) { - Assert.notNull(dispatcher, "'dispatcher' must not be null"); - this.targetDispatcher = dispatcher; - this.taskExecutor = taskExecutor != null ? taskExecutor - : new ConcurrentTaskExecutor(Executors.newSingleThreadExecutor()); - } - - - public boolean addHandler(MessageHandler handler) { - return this.targetDispatcher.addHandler(handler); - } - - public boolean removeHandler(MessageHandler handler) { - return this.targetDispatcher.removeHandler(handler); - } - - public final boolean dispatch(final Message message) { - this.taskExecutor.execute(new Runnable() { - public void run() { - targetDispatcher.dispatch(message); - } - }); - return true; - } - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java index 6fc85f9c6f..7a53569f96 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java @@ -18,9 +18,8 @@ package org.springframework.integration.config.xml; import org.w3c.dom.Element; -import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; +import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; @@ -37,6 +36,7 @@ public class PointToPointChannelParser extends AbstractChannelParser { private static final String DISPATCHER_PACKAGE = IntegrationNamespaceUtils.BASE_PACKAGE + ".dispatcher"; + @Override protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = null; @@ -81,23 +81,16 @@ public class PointToPointChannelParser extends AbstractChannelParser { else { builder = BeanDefinitionBuilder.genericBeanDefinition(CHANNEL_PACKAGE + ".DirectChannel"); } - parseDispatcher(element.getAttribute("dispatcher"), builder, parserContext); - } - return builder; - } - - - private void parseDispatcher(String dispatcherAttribute, BeanDefinitionBuilder builder, ParserContext parserContext) { - if (dispatcherAttribute != null) { - if (dispatcherAttribute.equals("failover")) { - BeanDefinitionBuilder dispatcherBuilder = BeanDefinitionBuilder - .genericBeanDefinition(DISPATCHER_PACKAGE + ".FailOverDispatcher"); - dispatcherBuilder.setRole(BeanDefinition.ROLE_SUPPORT); - builder.addConstructorArgReference(BeanDefinitionReaderUtils.registerWithGeneratedName(dispatcherBuilder - .getBeanDefinition(), parserContext.getRegistry())); + // this attribute is deprecated, but if set, we need to create a UnicastingDispatcher + // without any LoadBalancerStrategy and the failover flag set to true (default). + String dispatcherAttribute = element.getAttribute("dispatcher"); + if (!"failover".equals(dispatcherAttribute)) { + // round-robin dispatcher by default, but TODO first we need to check for the dispatcher element. + builder.addConstructorArgValue(new RootBeanDefinition( + DISPATCHER_PACKAGE + ".RoundRobinLoadBalancingStrategy", null, null)); } - } - // rely on the default for round-robin + } + return builder; } private boolean parseQueueCapacity(BeanDefinitionBuilder builder, Element queueElement) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliverException.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliveryException.java similarity index 84% rename from org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliverException.java rename to org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliveryException.java index 21e5dc35d6..72348acaa4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliverException.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliveryException.java @@ -16,6 +16,7 @@ package org.springframework.integration.dispatcher; +import java.util.Collections; import java.util.List; import org.springframework.integration.core.Message; @@ -30,12 +31,12 @@ import org.springframework.integration.message.MessageDeliveryException; * @since 1.0.3 */ @SuppressWarnings("serial") -public class AggregateMessageDeliverException extends MessageDeliveryException { +public class AggregateMessageDeliveryException extends MessageDeliveryException { private final List aggregatedExceptions; - public AggregateMessageDeliverException(Message undeliveredMessage, + public AggregateMessageDeliveryException(Message undeliveredMessage, String description, List aggregatedExceptions) { super(undeliveredMessage, description); this.aggregatedExceptions = aggregatedExceptions; @@ -43,7 +44,7 @@ public class AggregateMessageDeliverException extends MessageDeliveryException { public List getAggregatedExceptions() { - return this.aggregatedExceptions; + return Collections.unmodifiableList(this.aggregatedExceptions); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/LoadBalancingStrategy.java similarity index 53% rename from org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java rename to org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/LoadBalancingStrategy.java index ac7c4ae746..22707cad5e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/LoadBalancingStrategy.java @@ -1,5 +1,4 @@ -/* - * Copyright 2002-2009 the original author or authors. +/* 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. @@ -16,29 +15,20 @@ package org.springframework.integration.dispatcher; +import java.util.Iterator; import java.util.List; import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageHandler; /** - * {@link AbstractUnicastDispatcher} that will try its handlers in the - * same order every dispatch. + * Strategy for determining the iteration order of a MessageHandler list. * * @author Mark Fisher - * @author Iwein Fuld + * @since 1.0.3 */ -public class FailOverDispatcher extends AbstractUnicastDispatcher { +public interface LoadBalancingStrategy { - @Override - protected void handleExceptions(List allExceptions, - Message message, boolean isLast) { - if (isLast) { - if (allExceptions != null && allExceptions.size() == 1) { - throw allExceptions.get(0); - } - throw new AggregateMessageDeliverException(message, - "All attempts to deliver Message to MessageHandlers failed.", allExceptions); - } - } + public Iterator getHandlerIterator(Message message, List handlers); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinLoadBalancingStrategy.java similarity index 73% rename from org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinDispatcher.java rename to org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinLoadBalancingStrategy.java index f5b8f58196..5085dd67fb 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinLoadBalancingStrategy.java @@ -25,32 +25,26 @@ import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageHandler; /** - * Round-robin implementation of {@link AbstractUnicastDispatcher}. This + * Round-robin implementation of {@link LoadBalancingStrategy}. This * implementation will keep track of the index of the handler that has been * tried first and use a different starting handler every dispatch. * * @author Iwein Fuld + * @author Mark Fisher + * @since 1.0.3 */ -public class RoundRobinDispatcher extends AbstractUnicastDispatcher { - - private volatile boolean failover = true; +public class RoundRobinLoadBalancingStrategy implements LoadBalancingStrategy { private final AtomicInteger currentHandlerIndex = new AtomicInteger(); - public void setFailover(boolean failover) { - this.failover = failover; - } - /** * Returns an iterator that starts at a new point in the list every time the * first part of the list that is skipped will be used at the end of the * iteration, so it guarantees all handlers are returned once on subsequent * next() invocations. */ - @Override - protected Iterator getHandlerIterator() { - List handlers = this.getHandlers(); + public final Iterator getHandlerIterator(final Message message, final List handlers) { int size = handlers.size(); if (size == 0) { return handlers.iterator(); @@ -72,16 +66,4 @@ public class RoundRobinDispatcher extends AbstractUnicastDispatcher { return indexTail < 0 ? indexTail + size : indexTail; } - @Override - protected void handleExceptions(List allExceptions, - Message message, boolean isLast) { - if (isLast || !this.failover) { - if (allExceptions != null && allExceptions.size() == 1) { - throw allExceptions.get(0); - } - throw new AggregateMessageDeliverException(message, - "Failed to deliver Message to any MessageHandler.", allExceptions); - } - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java new file mode 100644 index 0000000000..068f877e5c --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java @@ -0,0 +1,148 @@ +/* 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.dispatcher; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.core.Message; +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 + * {@link Message} to at most one of its handlers. The handlers will be tried + * as determined by the {@link LoadBalancingStrategy} if one is configured. As + * soon as one of the handlers accepts the Message, the dispatcher will + * return true and ignore the rest of its handlers. + *

+ * If the dispatcher has no handlers, a {@link MessageDeliveryException} will be + * thrown. If all handlers throw Exceptions, the dispatcher will throw an + * {@link AggregateMessageDeliveryException}. + *

+ * A load-balancing strategy may be provided to this class to control the order in + * which the handlers will be tried. + * + * @author Iwein Fuld + * @author Mark Fisher + * @since 1.0.2 + */ +public class UnicastingDispatcher extends AbstractDispatcher { + + private volatile boolean failover = true; + + private volatile LoadBalancingStrategy loadBalancingStrategy; + + private final TaskExecutor taskExecutor; + + + public UnicastingDispatcher() { + this.taskExecutor = null; + } + + public UnicastingDispatcher(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + + /** + * Specify whether this dispatcher should failover when a single + * {@link MessageHandler} throws an Exception. The default value is + * true. + */ + public void setFailover(boolean failover) { + this.failover = failover; + } + + /** + * Provide a {@link LoadBalancingStrategy} for this dispatcher. + */ + public void setLoadBalancingStrategy(LoadBalancingStrategy loadBalancingStrategy) { + Assert.notNull(loadBalancingStrategy, "loadBalancingStrategy must not be null"); + this.loadBalancingStrategy = loadBalancingStrategy; + } + + public final boolean dispatch(final Message message) { + if (this.taskExecutor != null) { + this.taskExecutor.execute(new Runnable() { + public void run() { + doDispatch(message); + } + }); + return true; + } + return this.doDispatch(message); + } + + private boolean doDispatch(Message message) { + boolean success = false; + Iterator handlerIterator = this.getHandlerIterator(message); + if (!handlerIterator.hasNext()) { + throw new MessageDeliveryException(message, "Dispatcher has no subscribers."); + } + List exceptions = new ArrayList(); + while (success == false && handlerIterator.hasNext()) { + MessageHandler handler = handlerIterator.next(); + try { + handler.handleMessage(message); + success = true; // we have a winner. + } + catch (Exception e) { + RuntimeException runtimeException = (e instanceof RuntimeException) + ? (RuntimeException) e + : new MessageDeliveryException(message, + "Dispatcher failed to deliver Message.", e); + exceptions.add(runtimeException); + this.handleExceptions(exceptions, message, !handlerIterator.hasNext()); + } + } + return success; + } + + /** + * Returns the iterator that will be used to loop over the handlers. + * Delegates to a {@link LoadBalancingStrategy} if available. Otherwise, + * 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()); + } + return this.getHandlers().iterator(); + } + + /** + * Handles Exceptions that occur while dispatching. If this dispatcher has + * failover enabled, it will only throw an Exception when the handler list + * is exhausted. The 'isLast' flag will be true if the + * Exception occurred during the final iteration of the MessageHandlers. + * If failover is disabled for this dispatcher, it will re-throw any + * Exception immediately. + */ + private void handleExceptions(List allExceptions, Message message, boolean isLast) { + if (isLast || !this.failover) { + if (allExceptions != null && allExceptions.size() == 1) { + throw allExceptions.get(0); + } + throw new AggregateMessageDeliveryException(message, + "All attempts to deliver Message to MessageHandlers failed.", allExceptions); + } + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java index 2503bbf447..34c2a1a5e4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java @@ -20,6 +20,7 @@ import static org.hamcrest.CoreMatchers.is; 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.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; @@ -39,8 +40,8 @@ import org.springframework.integration.config.TestChannelInterceptor; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessagePriority; -import org.springframework.integration.dispatcher.FailOverDispatcher; -import org.springframework.integration.dispatcher.RoundRobinDispatcher; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; +import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageDeliveryException; @@ -79,17 +80,22 @@ public class ChannelParserTests { MessageChannel channel = (MessageChannel) context.getBean("defaultChannel"); assertEquals(DirectChannel.class, channel.getClass()); DirectFieldAccessor accessor = new DirectFieldAccessor(channel); - assertThat(accessor.getPropertyValue("dispatcher"), is(RoundRobinDispatcher.class)); + Object dispatcher = accessor.getPropertyValue("dispatcher"); + assertThat(dispatcher, is(UnicastingDispatcher.class)); + assertThat(new DirectFieldAccessor(dispatcher).getPropertyValue("loadBalancingStrategy"), + is(RoundRobinLoadBalancingStrategy.class)); } @Test - public void channelWithRoundRobinDispatcher() throws Exception { + public void channelWithFailoverDispatcherAttribute() throws Exception { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this .getClass()); - MessageChannel channel = (MessageChannel) context.getBean("failOverChannel"); + MessageChannel channel = (MessageChannel) context.getBean("channelWithFailoverAttribute"); assertEquals(DirectChannel.class, channel.getClass()); DirectFieldAccessor accessor = new DirectFieldAccessor(channel); - assertThat(accessor.getPropertyValue("dispatcher"), is(FailOverDispatcher.class)); + Object dispatcher = accessor.getPropertyValue("dispatcher"); + assertThat(dispatcher, is(UnicastingDispatcher.class)); + assertNull(new DirectFieldAccessor(dispatcher).getPropertyValue("loadBalancingStrategy")); } @Test diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml index c62a2e70d4..265167434c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml @@ -12,7 +12,7 @@ - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java index 41fdcf6ad4..a2ca5edfbc 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SubscriberOrderTests.java @@ -31,7 +31,7 @@ import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.dispatcher.FailOverDispatcher; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; import org.springframework.integration.message.MessageRejectedException; import org.springframework.integration.message.StringMessage; @@ -45,7 +45,6 @@ public class SubscriberOrderTests { GenericApplicationContext context = new GenericApplicationContext(); context.registerBeanDefinition("postProcessor", new RootBeanDefinition(MessagingAnnotationPostProcessor.class)); RootBeanDefinition channelDefinition = new RootBeanDefinition(DirectChannel.class); - channelDefinition.getConstructorArgumentValues().addGenericArgumentValue(new FailOverDispatcher()); context.registerBeanDefinition("input", channelDefinition); RootBeanDefinition testBeanDefinition = new RootBeanDefinition(TestBean.class); testBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue(1); @@ -72,7 +71,6 @@ public class SubscriberOrderTests { GenericApplicationContext context = new GenericApplicationContext(); context.registerBeanDefinition("postProcessor", new RootBeanDefinition(MessagingAnnotationPostProcessor.class)); RootBeanDefinition channelDefinition = new RootBeanDefinition(DirectChannel.class); - channelDefinition.getConstructorArgumentValues().addGenericArgumentValue(new FailOverDispatcher()); context.registerBeanDefinition("input", channelDefinition); RootBeanDefinition testBeanDefinition = new RootBeanDefinition(TestBean.class); testBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue(2); @@ -112,6 +110,7 @@ public class SubscriberOrderTests { GenericApplicationContext context = new GenericApplicationContext(); context.registerBeanDefinition("postProcessor", new RootBeanDefinition(MessagingAnnotationPostProcessor.class)); RootBeanDefinition channelDefinition = new RootBeanDefinition(DirectChannel.class); + channelDefinition.getConstructorArgumentValues().addGenericArgumentValue(new RoundRobinLoadBalancingStrategy()); context.registerBeanDefinition("input", channelDefinition); RootBeanDefinition testBeanDefinition = new RootBeanDefinition(TestBean.class); testBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue(1000); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/FailOverDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/FailOverDispatcherTests.java index ecc3ee4b70..e3ed03fa65 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/FailOverDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/FailOverDispatcherTests.java @@ -41,7 +41,7 @@ public class FailOverDispatcherTests { @Test public void singleMessage() throws InterruptedException { - FailOverDispatcher dispatcher = new FailOverDispatcher(); + UnicastingDispatcher dispatcher = new UnicastingDispatcher(); final CountDownLatch latch = new CountDownLatch(1); dispatcher.addHandler(createConsumer(TestHandlers.countDownHandler(latch))); dispatcher.dispatch(new StringMessage("test")); @@ -51,7 +51,7 @@ public class FailOverDispatcherTests { @Test public void pointToPoint() throws InterruptedException { - FailOverDispatcher dispatcher = new FailOverDispatcher(); + UnicastingDispatcher dispatcher = new UnicastingDispatcher(); final CountDownLatch latch = new CountDownLatch(1); final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); @@ -65,7 +65,7 @@ public class FailOverDispatcherTests { @Test public void noDuplicateSubscriptions() { - FailOverDispatcher dispatcher = new FailOverDispatcher(); + UnicastingDispatcher dispatcher = new UnicastingDispatcher(); final AtomicInteger counter = new AtomicInteger(); MessageHandler target = new CountingTestEndpoint(counter, false); dispatcher.addHandler(target); @@ -81,7 +81,7 @@ public class FailOverDispatcherTests { @Test public void removeConsumerBeforeSend() { - FailOverDispatcher dispatcher = new FailOverDispatcher(); + UnicastingDispatcher dispatcher = new UnicastingDispatcher(); final AtomicInteger counter = new AtomicInteger(); MessageHandler target1 = new CountingTestEndpoint(counter, false); MessageHandler target2 = new CountingTestEndpoint(counter, false); @@ -101,7 +101,7 @@ public class FailOverDispatcherTests { @Test public void removeConsumerBetweenSends() { - FailOverDispatcher dispatcher = new FailOverDispatcher(); + UnicastingDispatcher dispatcher = new UnicastingDispatcher(); final AtomicInteger counter = new AtomicInteger(); MessageHandler target1 = new CountingTestEndpoint(counter, false); MessageHandler target2 = new CountingTestEndpoint(counter, false); @@ -136,7 +136,7 @@ public class FailOverDispatcherTests { @Test(expected = MessageDeliveryException.class) public void removeConsumerLastTargetCausesDeliveryException() { - FailOverDispatcher dispatcher = new FailOverDispatcher(); + UnicastingDispatcher dispatcher = new UnicastingDispatcher(); final AtomicInteger counter = new AtomicInteger(); MessageHandler target = new CountingTestEndpoint(counter, false); dispatcher.addHandler(target); @@ -153,7 +153,7 @@ public class FailOverDispatcherTests { @Test public void firstHandlerReturnsTrue() { - FailOverDispatcher dispatcher = new FailOverDispatcher(); + UnicastingDispatcher dispatcher = new UnicastingDispatcher(); final AtomicInteger counter = new AtomicInteger(); MessageHandler target1 = new CountingTestEndpoint(counter, true); MessageHandler target2 = new CountingTestEndpoint(counter, false); @@ -167,7 +167,7 @@ public class FailOverDispatcherTests { @Test public void middleHandlerReturnsTrue() { - FailOverDispatcher dispatcher = new FailOverDispatcher(); + UnicastingDispatcher dispatcher = new UnicastingDispatcher(); final AtomicInteger counter = new AtomicInteger(); MessageHandler target1 = new CountingTestEndpoint(counter, false); MessageHandler target2 = new CountingTestEndpoint(counter, true); @@ -181,7 +181,7 @@ public class FailOverDispatcherTests { @Test public void allHandlersReturnFalse() { - FailOverDispatcher dispatcher = new FailOverDispatcher(); + UnicastingDispatcher dispatcher = new UnicastingDispatcher(); final AtomicInteger counter = new AtomicInteger(); MessageHandler target1 = new CountingTestEndpoint(counter, false); MessageHandler target2 = new CountingTestEndpoint(counter, false); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java index 820d0380e7..578d3791b6 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java @@ -46,7 +46,7 @@ public class RoundRobinDispatcherConcurrentTests { private static final int TOTAL_EXECUTIONS = 40; - private RoundRobinDispatcher dispatcher = new RoundRobinDispatcher(); + private UnicastingDispatcher dispatcher = new UnicastingDispatcher(); private ThreadPoolTaskExecutor scheduler = new ThreadPoolTaskExecutor(); @@ -67,6 +67,7 @@ public class RoundRobinDispatcherConcurrentTests { @Before public void initialize() throws Exception { + dispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); scheduler.setCorePoolSize(10); scheduler.setMaxPoolSize(10); scheduler.initialize(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java index 059878a837..3f4a09c460 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherTests.java @@ -15,8 +15,10 @@ package org.springframework.integration.dispatcher; -import static org.mockito.Mockito.*; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; @@ -32,7 +34,7 @@ import org.springframework.integration.message.MessageHandler; @RunWith(MockitoJUnit44Runner.class) public class RoundRobinDispatcherTests { - private AbstractUnicastDispatcher dispatcher = new RoundRobinDispatcher(); + private UnicastingDispatcher dispatcher = new UnicastingDispatcher(); @Mock private MessageHandler handler; @@ -43,6 +45,12 @@ public class RoundRobinDispatcherTests { @Mock private MessageHandler differentHandler; + @Before + public void setupDispatcher() { + this.dispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); + } + + @Test public void dispatchMessageWithSingleHandler() throws Exception { dispatcher.addHandler(handler);