diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index 5e7fe56c0e..093beb9aaf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -105,12 +105,18 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor if (endpointAnnotation == null) { return bean; } + if (this.messageBus == null) { + if (logger.isWarnEnabled()) { + logger.warn(this.getClass().getSimpleName() + " is disabled since no 'messageBus' was provided"); + } + return bean; + } DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); this.configureInput(bean, beanName, endpointAnnotation, endpoint); MessageHandlerChain handlerChain = this.createHandlerChain(bean); endpoint.setHandler(handlerChain); this.configureDefaultOutput(bean, beanName, endpointAnnotation, endpoint); - this.messageBus.registerEndpoint(bean + "-endpoint", endpoint); + this.messageBus.registerEndpoint(beanName + "-endpoint", endpoint); return bean; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java index 05d53036ba..2fe0200553 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java @@ -76,6 +76,12 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor { if (targetClass == null) { return bean; } + if (this.messageBus == null) { + if (logger.isWarnEnabled()) { + logger.warn(this.getClass().getSimpleName() + " is disabled since no 'messageBus' was provided"); + } + return bean; + } ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() { public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { Annotation annotation = method.getAnnotation(subscriberAnnotationType); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java index 0d6434970e..cffeeb38e0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java @@ -47,10 +47,6 @@ public class ChannelPollingMessageRetriever implements MessageRetriever { } - public MessageChannel getChannel() { - return this.channel; - } - public Collection> retrieveMessages() { List> messages = new LinkedList>(); while (messages.size() < this.channel.getDispatcherPolicy().getMaxMessagesPerTask()) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java index 270492cf10..aa8c4e00df 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java @@ -29,6 +29,7 @@ import org.springframework.integration.handler.AbstractMessageHandlerAdapter; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.util.SimpleMethodInvoker; +import org.springframework.util.Assert; /** * MessageHandler adapter for methods annotated with {@link Splitter @Splitter}. @@ -37,7 +38,8 @@ import org.springframework.integration.util.SimpleMethodInvoker; */ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter implements ChannelRegistryAware { - private static final String CHANNEL_KEY = "channel"; + public static final String CHANNEL_KEY = "channel"; + private Map attributes; @@ -47,7 +49,12 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter public SplitterMessageHandlerAdapter(Object object, Method method, Map attributes) { + Assert.notNull(object, "'object' must not be null"); + Assert.notNull(method, "'method' must not be null"); + Assert.isTrue(attributes != null && attributes.get(CHANNEL_KEY) != null, + "the 'channel' attribute is required"); this.setObject(object); + this.setMethodName(method.getName()); this.method = method; this.attributes = attributes; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java new file mode 100644 index 0000000000..329d2b0a89 --- /dev/null +++ b/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(); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java index 81175ecf69..614f066373 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java +++ b/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(); + } + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/MultiChannelRouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/MultiChannelRouterTests.java index c2030e9a2b..19489653b5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/MultiChannelRouterTests.java +++ b/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 resolve(Message message) { @@ -114,7 +115,6 @@ public class MultiChannelRouterTests { }; MultiChannelRouter router = new MultiChannelRouter(); router.setChannelResolver(channelResolver); - router.setResolutionRequired(true); router.afterPropertiesSet(); Message 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(); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/RouterMessageHandlerAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/RouterMessageHandlerAdapterTests.java index c179686079..395c88ff10 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/RouterMessageHandlerAdapterTests.java +++ b/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 attribs = new ConcurrentHashMap(); + RouterMessageHandlerAdapter adapter = new RouterMessageHandlerAdapter(testBean, routingMethod, attribs); + Message fooMessage = new StringMessage("foo"); + Message barMessage = new StringMessage("bar"); + Message 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 attribs = new ConcurrentHashMap(); + RouterMessageHandlerAdapter adapter = new RouterMessageHandlerAdapter(testBean, routingMethod, attribs); + Message fooMessage = new StringMessage("foo"); + Message barMessage = new StringMessage("bar"); + Message 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; + } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/SingleChannelRouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/SingleChannelRouterTests.java index 0467739795..f6869e9310 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/SingleChannelRouterTests.java +++ b/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 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(); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java new file mode 100644 index 0000000000..ae8ad6e9ef --- /dev/null +++ b/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 attribs = new ConcurrentHashMap(); + + + 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 stringToStringList(String input) { + return Arrays.asList(input.split("\\.")); + } + + public String[] messageToStringArray(Message input) { + return input.getPayload().toString().split("\\."); + } + + public List messageToStringList(Message input) { + return Arrays.asList(input.getPayload().toString().split("\\.")); + } + + public Message[] messageToMessageArray(Message input) { + String[] strings = input.getPayload().toString().split("\\."); + Message[] messages = new StringMessage[strings.length]; + for (int i = 0; i < strings.length; i++) { + messages[i] = new StringMessage(strings[i]); + } + return messages; + } + + public List> messageToMessageList(Message input) { + String[] strings = input.getPayload().toString().split("\\."); + List> messages = new ArrayList>(); + for (String s : strings) { + messages.add(new StringMessage(s)); + } + return messages; + } + + public Message[] stringToMessageArray(String input) { + String[] strings = input.split("\\."); + Message[] messages = new StringMessage[strings.length]; + for (int i = 0; i < strings.length; i++) { + messages[i] = new StringMessage(strings[i]); + } + return messages; + } + + public List> stringToMessageList(String input) { + String[] strings = input.split("\\."); + List> messages = new ArrayList>(); + for (String s : strings) { + messages.add(new StringMessage(s)); + } + return messages; + } + + public String[] invalidParameterCount(String param1, String param2) { + return null; + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/transformer/MessageFilterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/transformer/MessageFilterTests.java new file mode 100644 index 0000000000..9163c1780d --- /dev/null +++ b/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()); + } + +}