Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1814 refactored to ensure that the log message appears for ANY P2P channel, but not for PUB-SUB channel

Browse Source
This commit is contained in:
Oleg Zhurakousky
2011-03-01 14:38:12 -05:00
parent 7932a6592a
commit b73e73f069
4 changed files with 124 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java
View File

@@ -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) {

10
spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java
View File

@@ -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}
*/

29
spring-integration-core/src/test/java/org/springframework/integration/channel/DirectChannelTests.java
View File

@@ -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 {

113
spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java Normal file
View File

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