diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java index e6b13508ad..7d40249cb1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java @@ -21,6 +21,7 @@ import org.springframework.integration.MessageChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.util.Assert; /** @@ -28,11 +29,20 @@ import org.springframework.util.Assert; * {@link MessageHandler handler(s)} by delegating to a {@link MessageDispatcher}. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel implements SubscribableChannel { public boolean subscribe(MessageHandler handler) { - return this.getRequiredDispatcher().addHandler(handler); + MessageDispatcher dispatcher = this.getRequiredDispatcher(); + if (dispatcher instanceof UnicastingDispatcher){ + if (((UnicastingDispatcher) dispatcher).size() > 0){ + String message = "Point-to-Point channel '" + this.getComponentName() + "' has more then 1 subscriber. " + + "If load balancing strategy is provided, messages will be dispatched following its rules."; + this.logger.info(message); + } + } + return dispatcher.addHandler(handler); } public boolean unsubscribe(MessageHandler handle) { 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 637fd0d1a3..51a53bd01b 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,7 +16,6 @@ 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; @@ -34,15 +33,6 @@ 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/test/java/org/springframework/integration/channel/DirectChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java index 28e59d5885..65f46609a4 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,17 +18,11 @@ 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; @@ -36,9 +30,6 @@ 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 @@ -60,26 +51,6 @@ 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 { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java new file mode 100644 index 0000000000..1497dc8428 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java @@ -0,0 +1,113 @@ +/* + * Copyright 2002-2011 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.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import java.lang.reflect.Field; +import java.util.concurrent.Executor; + +import org.apache.commons.logging.Log; +import org.junit.Test; + +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.core.MessageHandler; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.ReflectionUtils.FieldCallback; + +/** + * @author Oleg Zhurakousky + * + */ +public class P2pChannelTests { + + @Test + public void testDirectChannelLoggingWithMoreThenOneSubscriber() { + final DirectChannel channel = new DirectChannel(); + channel.setBeanName("directChannel"); + + String logMessage = "Point-to-Point channel '" + channel.getComponentName() + "' has more then 1 subscriber. " + + "If load balancing strategy is provided, messages will be dispatched following its rules."; + + 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(logMessage); + channel.subscribe(mock(MessageHandler.class)); + verify(logger, times(1)).info(logMessage); + } + + @Test + public void testExecutorChannelLoggingWithMoreThenOneSubscriber() { + final ExecutorChannel channel = new ExecutorChannel(mock(Executor.class)); + channel.setBeanName("executorChannel"); + + String logMessage = "Point-to-Point channel '" + channel.getComponentName() + "' has more then 1 subscriber. " + + "If load balancing strategy is provided, messages will be dispatched following its rules."; + + 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(logMessage); + channel.subscribe(mock(MessageHandler.class)); + verify(logger, times(1)).info(logMessage); + } + + @Test + public void testPubSubChannelLoggingWithMoreThenOneSubscriber() { + final PublishSubscribeChannel channel = new PublishSubscribeChannel(); + channel.setBeanName("pubSubChannel"); + + String logMessage = "Point-to-Point channel '" + channel.getComponentName() + "' has more then 1 subscriber. " + + "If load balancing strategy is provided, messages will be dispatched following its rules."; + + 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(logMessage); + channel.subscribe(mock(MessageHandler.class)); + verify(logger, times(0)).info(logMessage); + } +}