diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 6f25a34de6..8e0534184d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -46,7 +46,7 @@ import org.springframework.util.StringUtils; */ public abstract class AbstractMessageChannel extends IntegrationObjectSupport implements MessageChannel, TrackableComponent { - private final Log logger = LogFactory.getLog(this.getClass()); + protected final Log logger = LogFactory.getLog(this.getClass()); private volatile boolean shouldTrack = false; 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 6086b147af..637fd0d1a3 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 @@ -16,6 +16,7 @@ package org.springframework.integration.channel; +import org.springframework.integration.core.MessageHandler; import org.springframework.integration.dispatcher.LoadBalancingStrategy; import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; import org.springframework.integration.dispatcher.UnicastingDispatcher; @@ -33,6 +34,14 @@ public class DirectChannel extends AbstractSubscribableChannel { private final UnicastingDispatcher dispatcher = new UnicastingDispatcher(); + public boolean subscribe(MessageHandler handler) { + + if (this.getDispatcher().size() > 0){ + this.logger.info("DirectChannel has more then 1 subscriber. " + + "Incoming messages will be dispatched between all subscribers following the load-balancing strategy."); + } + return super.subscribe(handler); + } /** * Create a channel with default {@link RoundRobinLoadBalancingStrategy} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index a06b9d1d2e..a5170dda88 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -77,4 +77,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher { return this.getClass().getSimpleName() + " with handlers: " + this.handlers.toString(); } + public int size(){ + return this.handlers.toArray().length; + } } 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 65f46609a4..28e59d5885 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 @@ -18,11 +18,17 @@ package org.springframework.integration.channel; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import java.lang.reflect.Field; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.apache.commons.logging.Log; import org.junit.Test; +import org.mockito.Mockito; import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.Message; @@ -30,6 +36,9 @@ import org.springframework.integration.core.MessageHandler; import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.ReflectionUtils.FieldCallback; /** * @author Mark Fisher @@ -51,6 +60,26 @@ public class DirectChannelTests { Object loadBalancingStrategy = dispatcherAccessor.getPropertyValue("loadBalancingStrategy"); assertTrue(loadBalancingStrategy instanceof RoundRobinLoadBalancingStrategy); } + + @Test + public void testWithMoreThenOneSubscriber() { + final DirectChannel channel = new DirectChannel(); + final Log logger = mock(Log.class); + ReflectionUtils.doWithFields(AbstractMessageChannel.class, new FieldCallback() { + + public void doWith(Field field) throws IllegalArgumentException, + IllegalAccessException { + if ("logger".equals(field.getName())){ + field.setAccessible(true); + field.set(channel, logger); + } + } + }); + channel.subscribe(mock(MessageHandler.class)); + verify(logger, times(0)).info(Mockito.anyString()); + channel.subscribe(mock(MessageHandler.class)); + verify(logger, times(1)).info(Mockito.anyString()); + } @Test public void testSendInSeparateThread() throws InterruptedException {