Extended test coverage

spring-integration-core/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java

@@ -0,0 +1,154 @@
+/*
+ * Copyright 2002-2007 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.config;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import org.springframework.beans.factory.config.RuntimeBeanReference;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.integration.annotation.Subscriber;
+import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.SimpleChannel;
+import org.springframework.integration.message.StringMessage;
+
+/**
+ * @author Mark Fisher
+ */
+public class SubscriberAnnotationPostProcessorTests {
+
+	@Test
+	public void testAnnotatedSubscriber() throws InterruptedException {
+		CountDownLatch latch = new CountDownLatch(1);
+		GenericApplicationContext context = new GenericApplicationContext();
+		context.registerBeanDefinition("testChannel", new RootBeanDefinition(SimpleChannel.class));
+		RootBeanDefinition subscriberDef = new RootBeanDefinition(SubscriberAnnotationTestBean.class);
+		subscriberDef.getConstructorArgumentValues().addGenericArgumentValue(latch);
+		context.registerBeanDefinition("testBean", subscriberDef);
+		String busBeanName = MessageBusParser.MESSAGE_BUS_BEAN_NAME;
+		context.registerBeanDefinition(busBeanName, new RootBeanDefinition(MessageBus.class));
+		RootBeanDefinition postProcessorDef = new RootBeanDefinition(SubscriberAnnotationPostProcessor.class);
+		postProcessorDef.getPropertyValues().addPropertyValue("messageBus", new RuntimeBeanReference(busBeanName));
+		context.registerBeanDefinition("postProcessor", postProcessorDef);
+		context.refresh();
+		context.start();
+		SubscriberAnnotationTestBean testBean = (SubscriberAnnotationTestBean) context.getBean("testBean");
+		assertEquals(1, latch.getCount());
+		assertNull(testBean.getMessageText());
+		MessageChannel testChannel = (MessageChannel) context.getBean("testChannel");
+		testChannel.send(new StringMessage("test-123"));
+		latch.await(1000, TimeUnit.MILLISECONDS);
+		assertEquals(0, latch.getCount());
+		assertEquals("test-123", testBean.getMessageText());
+		context.stop();
+	}
+
+	@Test
+	public void testCustomAnnotation() throws InterruptedException {
+		CountDownLatch latch = new CountDownLatch(1);
+		GenericApplicationContext context = new GenericApplicationContext();
+		context.registerBeanDefinition("testChannel", new RootBeanDefinition(SimpleChannel.class));
+		RootBeanDefinition subscriberDef = new RootBeanDefinition(CustomAnnotationTestBean.class);
+		subscriberDef.getConstructorArgumentValues().addGenericArgumentValue(latch);
+		context.registerBeanDefinition("testBean", subscriberDef);
+		String busBeanName = MessageBusParser.MESSAGE_BUS_BEAN_NAME;
+		context.registerBeanDefinition(busBeanName, new RootBeanDefinition(MessageBus.class));
+		RootBeanDefinition postProcessorDef = new RootBeanDefinition(SubscriberAnnotationPostProcessor.class);
+		postProcessorDef.getPropertyValues().addPropertyValue("messageBus", new RuntimeBeanReference(busBeanName));
+		postProcessorDef.getPropertyValues().addPropertyValue("subscriberAnnotationType", CustomSubscriberAnnotation.class);
+		postProcessorDef.getPropertyValues().addPropertyValue("channelNameAttribute", "subscribeTo");
+		context.registerBeanDefinition("postProcessor", postProcessorDef);
+		context.refresh();
+		context.start();
+		CustomAnnotationTestBean testBean = (CustomAnnotationTestBean) context.getBean("testBean");
+		assertEquals(1, latch.getCount());
+		assertNull(testBean.getMessageText());
+		MessageChannel testChannel = (MessageChannel) context.getBean("testChannel");
+		testChannel.send(new StringMessage("test-456"));
+		latch.await(1000, TimeUnit.MILLISECONDS);
+		assertEquals(0, latch.getCount());
+		assertEquals("test-456", testBean.getMessageText());
+		context.stop();
+	}
+
+
+	public static class AbstractSubscriberAnnotationTestBean {
+
+		protected String messageText;
+
+		private CountDownLatch latch;
+
+		public AbstractSubscriberAnnotationTestBean(CountDownLatch latch) {
+			this.latch = latch;
+		}
+
+		protected void countDown() {
+			this.latch.countDown();
+		}
+
+		public String getMessageText() {
+			return this.messageText;
+		}
+	}
+
+
+	public static class SubscriberAnnotationTestBean extends SubscriberAnnotationPostProcessorTests.AbstractSubscriberAnnotationTestBean {
+
+		public SubscriberAnnotationTestBean(CountDownLatch latch) {
+			super(latch);
+		}
+
+		@Subscriber(channel="testChannel")
+		public void testMethod(String messageText) {
+			this.messageText = messageText;
+			this.countDown();
+		}
+	}
+
+
+	public static class CustomAnnotationTestBean extends SubscriberAnnotationPostProcessorTests.AbstractSubscriberAnnotationTestBean {
+
+		public CustomAnnotationTestBean(CountDownLatch latch) {
+			super(latch);
+		}
+
+		@CustomSubscriberAnnotation(subscribeTo="testChannel")
+		public void testMethod(String messageText) {
+			this.messageText = messageText;
+			this.countDown();
+		}
+	}
+
+
+	@Target(ElementType.METHOD)
+	@Retention(RetentionPolicy.RUNTIME)
+	public static @interface CustomSubscriberAnnotation {
+		String subscribeTo();
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java

@@ -17,12 +17,22 @@
 package org.springframework.integration.endpoint.annotation;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
 import org.springframework.context.support.AbstractApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.annotation.DefaultOutput;
+import org.springframework.integration.annotation.MessageEndpoint;
+import org.springframework.integration.annotation.Polled;
+import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.SimpleChannel;
+import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.StringMessage;
 
@@ -55,4 +65,79 @@ public class MessageEndpointAnnotationPostProcessorTests {
 		context.stop();
 	}
 
+	@Test
+	public void testPolledAnnotation() throws InterruptedException {
+		MessageBus messageBus = new MessageBus();
+		SimpleChannel testChannel = new SimpleChannel();
+		messageBus.registerChannel("testChannel", testChannel);
+		MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor();
+		postProcessor.setMessageBus(messageBus);
+		PolledAnnotationTestBean testBean = new PolledAnnotationTestBean();
+		postProcessor.postProcessAfterInitialization(testBean, "testBean");
+		messageBus.start();
+		Message<?> message = testChannel.receive(1000);
+		assertEquals("test", message.getPayload());
+		messageBus.stop();
+	}
+
+	@Test
+	public void testDefaultOutputAnnotation() throws InterruptedException {
+		MessageBus messageBus = new MessageBus();
+		SimpleChannel testChannel = new SimpleChannel();
+		messageBus.registerChannel("testChannel", testChannel);
+		MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor();
+		postProcessor.setMessageBus(messageBus);
+		CountDownLatch latch = new CountDownLatch(1);
+		DefaultOutputAnnotationTestBean testBean = new DefaultOutputAnnotationTestBean(latch);
+		postProcessor.postProcessAfterInitialization(testBean, "testBean");
+		messageBus.start();
+		testChannel.send(new StringMessage("foo"));
+		latch.await(1000, TimeUnit.MILLISECONDS);
+		assertEquals(0, latch.getCount());
+		assertEquals("foo", testBean.getMessageText());
+		messageBus.stop();
+	}
+
+	@Test
+	public void testPostProcessorWithNoMessageBus() {
+		MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor();
+		PolledAnnotationTestBean testBean = new PolledAnnotationTestBean();
+		Object result = postProcessor.postProcessAfterInitialization(testBean, "testBean");
+		assertSame(testBean, result);
+	}
+
+
+	@MessageEndpoint(defaultOutput="testChannel")
+	private static class PolledAnnotationTestBean {
+
+		@Polled(period=100)
+		public String poller() {
+			return "test";
+		}
+	}
+
+
+	@MessageEndpoint(input="testChannel")
+	private static class DefaultOutputAnnotationTestBean {
+
+		private String messageText;
+
+		private CountDownLatch latch;
+
+
+		public DefaultOutputAnnotationTestBean(CountDownLatch latch) {
+			this.latch = latch;
+		}
+
+		public String getMessageText() {
+			return this.messageText;
+		}
+
+		@DefaultOutput
+		public void countdown(String input) {
+			this.messageText = input;
+			latch.countDown();
+		}
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/router/MultiChannelRouterTests.java

@@ -106,6 +106,7 @@ public class MultiChannelRouterTests {
 		router.afterPropertiesSet();
 	}
 
+	@Test
 	public void testChannelResolutionFailureIgnoredByDefault() {
 		MultiChannelResolver channelResolver = new MultiChannelResolver() {
 			public List<MessageChannel> resolve(Message<?> message) {
@@ -114,7 +115,6 @@ public class MultiChannelRouterTests {
 		};
 		MultiChannelRouter router = new MultiChannelRouter();
 		router.setChannelResolver(channelResolver);
-		router.setResolutionRequired(true);
 		router.afterPropertiesSet();
 		Message<String> message = new StringMessage("123", "test");
 		router.handle(message);
@@ -135,6 +135,7 @@ public class MultiChannelRouterTests {
 		router.handle(message);
 	}
 
+	@Test
 	public void testChannelNameResolutionFailureIgnoredByDefault() {
 		MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
 			public String[] resolve(Message<?> message) {
@@ -154,7 +155,7 @@ public class MultiChannelRouterTests {
 	public void testChannelNameResolutionFailureThrowsExceptionWhenResolutionRequired() {
 		MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
 			public String[] resolve(Message<?> message) {
-				return new String[] {"noSuchChannel"};
+				return null;
 			}
 		};
 		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
@@ -167,4 +168,36 @@ public class MultiChannelRouterTests {
 		router.handle(message);
 	}
 
+	@Test(expected=MessagingConfigurationException.class)
+	public void testChannelRegistryIsRequiredWhenUsingChannelNameResolver() {
+		MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
+			public String[] resolve(Message<?> message) {
+				return new String[] {"notImportant"};
+			}
+		};
+		MultiChannelRouter router = new MultiChannelRouter();
+		router.setChannelNameResolver(channelNameResolver);
+		router.resolveChannels(new StringMessage("this should fail"));
+	}
+
+	@Test(expected=MessagingConfigurationException.class)
+	public void testValidateChannelRegistryIsPresentWhenUsingChannelNameResolver() {
+		MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
+			public String[] resolve(Message<?> message) {
+				return new String[] {"notImportant"};
+			}
+		};
+		MultiChannelRouter router = new MultiChannelRouter();
+		router.setChannelNameResolver(channelNameResolver);
+		router.afterPropertiesSet();
+	}
+
+	@Test(expected=MessagingConfigurationException.class)
+	public void testChannelResolverIsRequired() {
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		MultiChannelRouter router = new MultiChannelRouter();
+		router.setChannelRegistry(channelRegistry);
+		router.afterPropertiesSet();
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/router/RouterMessageHandlerAdapterTests.java

@@ -293,7 +293,44 @@ public class RouterMessageHandlerAdapterTests {
 	}
 
 	@Test
-	public void testMultiChannelInstanceResolutionByPayload() throws Exception {
+	public void testMultiChannelNameArrayResolutionByMessage() throws Exception {
+		SimpleChannel fooChannel = new SimpleChannel();
+		SimpleChannel barChannel = new SimpleChannel();
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		channelRegistry.registerChannel("foo-channel", fooChannel);
+		channelRegistry.registerChannel("bar-channel", barChannel);
+		MultiChannelNameRoutingTestBean testBean = new MultiChannelNameRoutingTestBean();
+		Method routingMethod = testBean.getClass().getMethod("routeMessageToArray", Message.class);
+		Map<String, Object> attribs = new ConcurrentHashMap<String, Object>();
+		RouterMessageHandlerAdapter adapter = new RouterMessageHandlerAdapter(testBean, routingMethod, attribs);
+		Message<String> fooMessage = new StringMessage("foo");
+		Message<String> barMessage = new StringMessage("bar");
+		Message<String> badMessage = new StringMessage("bad");
+		adapter.setChannelRegistry(channelRegistry);
+		adapter.afterPropertiesSet();
+		adapter.handle(fooMessage);
+		Message<?> result1 = fooChannel.receive(0);
+		assertNotNull(result1);
+		assertEquals("foo", result1.getPayload());
+		Message<?> result2 = barChannel.receive(0);
+		assertNotNull(result2);
+		assertEquals("foo", result2.getPayload());
+		adapter.handle(barMessage);
+		Message<?> result3 = fooChannel.receive(0);
+		assertNotNull(result3);
+		assertEquals("bar", result3.getPayload());
+		Message<?> result4 = barChannel.receive(0);
+		assertNotNull(result4);
+		assertEquals("bar", result4.getPayload());
+		adapter.handle(badMessage);
+		Message<?> result5 = fooChannel.receive(0);
+		assertNull(result5);
+		Message<?> result6 = barChannel.receive(0);
+		assertNull(result6);
+	}
+
+	@Test
+	public void testMultiChannelListResolutionByPayload() throws Exception {
 		SimpleChannel fooChannel = new SimpleChannel();
 		SimpleChannel barChannel = new SimpleChannel();
 		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
@@ -330,7 +367,7 @@ public class RouterMessageHandlerAdapterTests {
 	}
 
 	@Test
-	public void testMultiChannelInstanceResolutionByMessage() throws Exception {
+	public void testMultiChannelListResolutionByMessage() throws Exception {
 		SimpleChannel fooChannel = new SimpleChannel();
 		SimpleChannel barChannel = new SimpleChannel();
 		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
@@ -366,6 +403,43 @@ public class RouterMessageHandlerAdapterTests {
 		assertNull(result6);
 	}
 
+	@Test
+	public void testMultiChannelArrayResolutionByMessage() throws Exception {
+		SimpleChannel fooChannel = new SimpleChannel();
+		SimpleChannel barChannel = new SimpleChannel();
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		channelRegistry.registerChannel("foo-channel", fooChannel);
+		channelRegistry.registerChannel("bar-channel", barChannel);
+		MultiChannelInstanceRoutingTestBean testBean = new MultiChannelInstanceRoutingTestBean(channelRegistry);
+		Method routingMethod = testBean.getClass().getMethod("routeMessageToArray", Message.class);
+		Map<String, Object> attribs = new ConcurrentHashMap<String, Object>();
+		RouterMessageHandlerAdapter adapter = new RouterMessageHandlerAdapter(testBean, routingMethod, attribs);
+		Message<String> fooMessage = new StringMessage("foo");
+		Message<String> barMessage = new StringMessage("bar");
+		Message<String> badMessage = new StringMessage("bad");
+		adapter.setChannelRegistry(channelRegistry);
+		adapter.afterPropertiesSet();
+		adapter.handle(fooMessage);
+		Message<?> result1 = fooChannel.receive(0);
+		assertNotNull(result1);
+		assertEquals("foo", result1.getPayload());
+		Message<?> result2 = barChannel.receive(0);
+		assertNotNull(result2);
+		assertEquals("foo", result2.getPayload());
+		adapter.handle(barMessage);
+		Message<?> result3 = fooChannel.receive(0);
+		assertNotNull(result3);
+		assertEquals("bar", result3.getPayload());
+		Message<?> result4 = barChannel.receive(0);
+		assertNotNull(result4);
+		assertEquals("bar", result4.getPayload());
+		adapter.handle(badMessage);
+		Message<?> result5 = fooChannel.receive(0);
+		assertNull(result5);
+		Message<?> result6 = barChannel.receive(0);
+		assertNull(result6);
+	}
+
 
 	public static class SingleChannelNameRoutingTestBean {
 
@@ -404,6 +478,16 @@ public class RouterMessageHandlerAdapterTests {
 			}
 			return results;
 		}
+
+		public String[] routeMessageToArray(Message<?> message) {
+			String[] results = null;
+			if (message.getPayload().equals("foo") || message.getPayload().equals("bar")) {
+				results = new String[2];
+				results[0] = "foo-channel";
+				results[1] = "bar-channel";
+			}
+			return results;
+		}
 	}
 
 
@@ -456,6 +540,16 @@ public class RouterMessageHandlerAdapterTests {
 			}
 			return results;
 		}
+
+		public MessageChannel[] routeMessageToArray(Message<?> message) {
+			MessageChannel[] results = null;
+			if (message.getPayload().equals("foo") || message.getPayload().equals("bar")) {
+				results = new MessageChannel[2];
+				results[0] = registry.lookupChannel("foo-channel");
+				results[1] = registry.lookupChannel("bar-channel");
+			}
+			return results;
+		}
 	}
 
 }

spring-integration-core/src/test/java/org/springframework/integration/router/SingleChannelRouterTests.java

@@ -92,6 +92,7 @@ public class SingleChannelRouterTests {
 		router.afterPropertiesSet();
 	}
 
+	@Test
 	public void testChannelResolutionFailureIgnoredByDefault() {
 		ChannelResolver channelResolver = new ChannelResolver() {
 			public MessageChannel resolve(Message<?> message) {
@@ -100,7 +101,6 @@ public class SingleChannelRouterTests {
 		};
 		SingleChannelRouter router = new SingleChannelRouter();
 		router.setChannelResolver(channelResolver);
-		router.setResolutionRequired(true);
 		router.afterPropertiesSet();
 		Message<String> message = new StringMessage("123", "test");
 		router.handle(message);
@@ -121,6 +121,7 @@ public class SingleChannelRouterTests {
 		router.handle(message);
 	}
 
+	@Test
 	public void testChannelNameResolutionFailureIgnoredByDefault() {
 		ChannelNameResolver channelNameResolver = new ChannelNameResolver() {
 			public String resolve(Message<?> message) {
@@ -153,4 +154,36 @@ public class SingleChannelRouterTests {
 		router.handle(message);
 	}
 
+	@Test(expected=MessagingConfigurationException.class)
+	public void testChannelRegistryIsRequiredWhenUsingChannelNameResolver() {
+		ChannelNameResolver channelNameResolver = new ChannelNameResolver() {
+			public String resolve(Message<?> message) {
+				return "notImportant";
+			}
+		};
+		SingleChannelRouter router = new SingleChannelRouter();
+		router.setChannelNameResolver(channelNameResolver);
+		router.resolveChannels(new StringMessage("this should fail"));
+	}
+
+	@Test(expected=MessagingConfigurationException.class)
+	public void testValidateChannelRegistryIsPresentWhenUsingChannelNameResolver() {
+		ChannelNameResolver channelNameResolver = new ChannelNameResolver() {
+			public String resolve(Message<?> message) {
+				return "notImportant";
+			}
+		};
+		SingleChannelRouter router = new SingleChannelRouter();
+		router.setChannelNameResolver(channelNameResolver);
+		router.afterPropertiesSet();
+	}
+
+	@Test(expected=MessagingConfigurationException.class)
+	public void testChannelResolverIsRequired() {
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		SingleChannelRouter router = new SingleChannelRouter();
+		router.setChannelRegistry(channelRegistry);
+		router.afterPropertiesSet();
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java

@@ -0,0 +1,242 @@
+/*
+ * Copyright 2002-2007 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.router;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.junit.Test;
+
+import org.springframework.integration.MessagingConfigurationException;
+import org.springframework.integration.channel.ChannelRegistry;
+import org.springframework.integration.channel.DefaultChannelRegistry;
+import org.springframework.integration.channel.SimpleChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.StringMessage;
+
+/**
+ * @author Mark Fisher
+ */
+public class SplitterMessageHandlerAdapterTests {
+
+	private SimpleChannel testChannel = new SimpleChannel();
+
+	private ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+
+	private SplitterTestBean testBean = new SplitterTestBean();
+
+	private Map<String, Object> attribs = new ConcurrentHashMap<String, Object>();
+
+
+	public SplitterMessageHandlerAdapterTests() {
+		this.channelRegistry.registerChannel("testChannel", testChannel);
+		this.attribs.put(SplitterMessageHandlerAdapter.CHANNEL_KEY, "testChannel");
+	}
+
+
+	@Test
+	public void testSplitPayloadToStringArray() throws Exception {
+		StringMessage message = new StringMessage("foo.bar");
+		SplitterMessageHandlerAdapter adapter = this.getAdapter("stringToStringArray");
+		adapter.handle(message);
+		Message<?> reply1 = testChannel.receive(0);
+		assertNotNull(reply1);
+		assertEquals("foo", reply1.getPayload());
+		Message<?> reply2 = testChannel.receive(0);
+		assertNotNull(reply2);
+		assertEquals("bar", reply2.getPayload());
+	}
+
+	@Test
+	public void testSplitPayloadToStringList() throws Exception {
+		StringMessage message = new StringMessage("foo.bar");
+		SplitterMessageHandlerAdapter adapter = this.getAdapter("stringToStringList");
+		adapter.handle(message);
+		Message<?> reply1 = testChannel.receive(0);
+		assertNotNull(reply1);
+		assertEquals("foo", reply1.getPayload());
+		Message<?> reply2 = testChannel.receive(0);
+		assertNotNull(reply2);
+		assertEquals("bar", reply2.getPayload());
+	}
+
+	@Test
+	public void testSplitMessageToStringArray() throws Exception {
+		StringMessage message = new StringMessage("foo.bar");
+		SplitterMessageHandlerAdapter adapter = this.getAdapter("messageToStringArray");
+		adapter.handle(message);
+		Message<?> reply1 = testChannel.receive(0);
+		assertNotNull(reply1);
+		assertEquals("foo", reply1.getPayload());
+		Message<?> reply2 = testChannel.receive(0);
+		assertNotNull(reply2);
+		assertEquals("bar", reply2.getPayload());
+	}
+
+	@Test
+	public void testSplitMessageToStringList() throws Exception {
+		StringMessage message = new StringMessage("foo.bar");
+		SplitterMessageHandlerAdapter adapter = this.getAdapter("messageToStringList");
+		adapter.handle(message);
+		Message<?> reply1 = testChannel.receive(0);
+		assertNotNull(reply1);
+		assertEquals("foo", reply1.getPayload());
+		Message<?> reply2 = testChannel.receive(0);
+		assertNotNull(reply2);
+		assertEquals("bar", reply2.getPayload());
+	}
+
+	@Test
+	public void testSplitMessageToMessageArray() throws Exception {
+		StringMessage message = new StringMessage("foo.bar");
+		SplitterMessageHandlerAdapter adapter = this.getAdapter("messageToMessageArray");
+		adapter.handle(message);
+		Message<?> reply1 = testChannel.receive(0);
+		assertNotNull(reply1);
+		assertEquals("foo", reply1.getPayload());
+		Message<?> reply2 = testChannel.receive(0);
+		assertNotNull(reply2);
+		assertEquals("bar", reply2.getPayload());
+	}
+
+	@Test
+	public void testSplitMessageToMessageList() throws Exception {
+		StringMessage message = new StringMessage("foo.bar");
+		SplitterMessageHandlerAdapter adapter = this.getAdapter("messageToMessageList");
+		adapter.handle(message);
+		Message<?> reply1 = testChannel.receive(0);
+		assertNotNull(reply1);
+		assertEquals("foo", reply1.getPayload());
+		Message<?> reply2 = testChannel.receive(0);
+		assertNotNull(reply2);
+		assertEquals("bar", reply2.getPayload());
+	}
+
+	@Test
+	public void testSplitStringToMessageArray() throws Exception {
+		StringMessage message = new StringMessage("foo.bar");
+		SplitterMessageHandlerAdapter adapter = this.getAdapter("stringToMessageArray");
+		adapter.handle(message);
+		Message<?> reply1 = testChannel.receive(0);
+		assertNotNull(reply1);
+		assertEquals("foo", reply1.getPayload());
+		Message<?> reply2 = testChannel.receive(0);
+		assertNotNull(reply2);
+		assertEquals("bar", reply2.getPayload());
+	}
+
+	@Test
+	public void testSplitStringToMessageList() throws Exception {
+		StringMessage message = new StringMessage("foo.bar");
+		SplitterMessageHandlerAdapter adapter = this.getAdapter("stringToMessageList");
+		adapter.handle(message);
+		Message<?> reply1 = testChannel.receive(0);
+		assertNotNull(reply1);
+		assertEquals("foo", reply1.getPayload());
+		Message<?> reply2 = testChannel.receive(0);
+		assertNotNull(reply2);
+		assertEquals("bar", reply2.getPayload());
+	}
+
+	@Test(expected=MessagingConfigurationException.class)
+	public void testInvalidReturnType() throws Exception {
+		Method splittingMethod = this.testBean.getClass().getMethod("invalidParameterCount", String.class, String.class);
+		SplitterMessageHandlerAdapter adapter = new SplitterMessageHandlerAdapter(testBean, splittingMethod, attribs);
+		adapter.setChannelRegistry(channelRegistry);
+		adapter.afterPropertiesSet();
+		StringMessage message = new StringMessage("foo.bar");
+		adapter.handle(message);
+	}
+
+
+	private SplitterMessageHandlerAdapter getAdapter(String methodName) throws Exception {
+		Class<?> paramType = methodName.startsWith("message") ? Message.class : String.class;
+		Method splittingMethod = this.testBean.getClass().getMethod(methodName, paramType);
+		SplitterMessageHandlerAdapter adapter = new SplitterMessageHandlerAdapter(testBean, splittingMethod, attribs);
+		adapter.setChannelRegistry(channelRegistry);
+		adapter.afterPropertiesSet();
+		return adapter;
+	}
+
+
+	public static class SplitterTestBean {
+
+		public String[] stringToStringArray(String input) {
+			return input.split("\\.");
+		}
+
+		public List<String> stringToStringList(String input) {
+			return Arrays.asList(input.split("\\."));
+		}
+
+		public String[] messageToStringArray(Message<?> input) {
+			return input.getPayload().toString().split("\\.");
+		}
+
+		public List<String> messageToStringList(Message<?> input) {
+			return Arrays.asList(input.getPayload().toString().split("\\."));
+		}
+
+		public Message<String>[] messageToMessageArray(Message<?> input) {
+			String[] strings = input.getPayload().toString().split("\\.");
+			Message<String>[] messages = new StringMessage[strings.length];
+			for (int i = 0; i < strings.length; i++) {
+				messages[i] = new StringMessage(strings[i]);
+			}
+			return messages;
+		}
+
+		public List<Message<String>> messageToMessageList(Message<?> input) {
+			String[] strings = input.getPayload().toString().split("\\.");
+			List<Message<String>> messages = new ArrayList<Message<String>>();
+			for (String s : strings) {
+				messages.add(new StringMessage(s));
+			}
+			return messages;
+		}
+
+		public Message<String>[] stringToMessageArray(String input) {
+			String[] strings = input.split("\\.");
+			Message<String>[] messages = new StringMessage[strings.length];
+			for (int i = 0; i < strings.length; i++) {
+				messages[i] = new StringMessage(strings[i]);
+			}
+			return messages;
+		}
+
+		public List<Message<String>> stringToMessageList(String input) {
+			String[] strings = input.split("\\.");
+			List<Message<String>> messages = new ArrayList<Message<String>>();
+			for (String s : strings) {
+				messages.add(new StringMessage(s));
+			}
+			return messages;
+		}
+
+		public String[] invalidParameterCount(String param1, String param2) {
+			return null;
+		}
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/transformer/MessageFilterTests.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2002-2007 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.transformer;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Test;
+
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.handler.MessageHandlerChain;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.integration.message.selector.MessageSelector;
+
+/**
+ * @author Mark Fisher
+ */
+public class MessageFilterTests {
+
+	@Test
+	public void testFilterAcceptsMessage() {
+		final AtomicBoolean secondHandlerReceived = new AtomicBoolean();
+		MessageHandlerChain chain = new MessageHandlerChain();
+		MessageFilter filter = new MessageFilter(new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				return true;
+			}
+		});
+		chain.add(filter);
+		chain.add(new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				secondHandlerReceived.set(true);
+				return null;
+			}
+		});
+		chain.handle(new StringMessage("test"));
+		assertTrue(secondHandlerReceived.get());
+	}
+
+	@Test
+	public void testFilterRejectsMessage() {
+		final AtomicBoolean secondHandlerReceived = new AtomicBoolean();
+		MessageHandlerChain chain = new MessageHandlerChain();
+		MessageFilter filter = new MessageFilter(new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				return false;
+			}
+		});
+		chain.add(filter);
+		chain.add(new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				secondHandlerReceived.set(true);
+				return null;
+			}
+		});
+		chain.handle(new StringMessage("test"));
+		assertFalse(secondHandlerReceived.get());
+	}
+
+}