From 61631c365b8e8ec1c3ddaccc54723fc504ad17cf Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 7 Apr 2008 13:25:50 +0000 Subject: [PATCH] MessageEndpointAnnotationPostProcessor no longer sets a default Schedule when creating a Subscription (INT-171). --- .../integration/bus/MessageBus.java | 10 +++++-- ...essageEndpointAnnotationPostProcessor.java | 4 +-- .../SynchronousChannelSubscriptionTests.java | 29 ++++++++++++++++++- 3 files changed, 36 insertions(+), 7 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 6b9ef3b1ec..31bc2c4ac6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -303,11 +303,15 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio private void activateEndpoint(MessageEndpoint endpoint) { Subscription subscription = endpoint.getSubscription(); + if (subscription == null) { + throw new MessagingConfigurationException("Unable to register endpoint '" + + endpoint + "'. No subscription information is available."); + } MessageChannel channel = subscription.getChannel(); if (channel == null) { String channelName = subscription.getChannelName(); if (channelName == null) { - throw new MessagingConfigurationException("endpoint '" + endpoint.getName() + + throw new MessagingConfigurationException("endpoint '" + endpoint + "' must provide either 'channel' or 'channelName' in its subscription metadata"); } channel = this.lookupChannel(channelName); @@ -329,7 +333,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio if (outputChannelName != null && this.lookupChannel(outputChannelName) == null) { if (!this.autoCreateChannels) { throw new MessagingConfigurationException("Unknown channel '" + outputChannelName + - "' configured as 'default-output' for endpoint '" + endpoint.getName() + + "' configured as 'default-output' for endpoint '" + endpoint + "'. Consider enabling the 'autoCreateChannels' option for the message bus."); } this.registerChannel(outputChannelName, new SimpleChannel()); @@ -341,7 +345,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio this.registerWithDispatcher(channel, endpoint, subscription.getSchedule()); if (logger.isInfoEnabled()) { logger.info("activated subscription to channel '" + channel.getName() + - "' for endpoint '" + endpoint.getName() + "'"); + "' for endpoint '" + endpoint + "'"); } } 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 78d8d87c46..77579a9320 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 @@ -54,7 +54,6 @@ import org.springframework.integration.router.config.RouterMessageHandlerCreator import org.springframework.integration.router.config.SplitterMessageHandlerCreator; import org.springframework.integration.router.config.AggregatorMessageHandlerCreator; import org.springframework.integration.scheduling.PollingSchedule; -import org.springframework.integration.scheduling.Schedule; import org.springframework.integration.scheduling.Subscription; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; @@ -136,8 +135,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor final DefaultMessageEndpoint endpoint) { String channelName = annotation.input(); if (StringUtils.hasText(channelName)) { - Schedule schedule = new PollingSchedule(annotation.pollPeriod()); - Subscription subscription = new Subscription(channelName, schedule); + Subscription subscription = new Subscription(channelName); endpoint.setSubscription(subscription); } ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java index f9da4243ee..9511c6bd4c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java @@ -21,7 +21,10 @@ import static org.junit.Assert.assertEquals; import org.junit.Before; import org.junit.Test; +import org.springframework.integration.annotation.Handler; +import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor; import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.MessageHandler; @@ -49,7 +52,7 @@ public class SynchronousChannelSubscriptionTests { @Test - public void testSendAndReceive() throws InterruptedException { + public void testSendAndReceiveForRegisteredEndpoint() { DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new TestHandler()); endpoint.setSubscription(new Subscription("sourceChannel")); endpoint.setDefaultOutputChannelName("targetChannel"); @@ -58,6 +61,20 @@ public class SynchronousChannelSubscriptionTests { this.sourceChannel.send(new StringMessage("foo")); Message response = this.targetChannel.receive(); assertEquals("foo!", response.getPayload()); + bus.stop(); + } + + @Test + public void testSendAndReceiveForAnnotatedEndpoint() { + MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(bus); + postProcessor.afterPropertiesSet(); + TestEndpoint endpoint = new TestEndpoint(); + postProcessor.postProcessAfterInitialization(endpoint, "testEndpoint"); + bus.start(); + this.sourceChannel.send(new StringMessage("foo")); + Message response = this.targetChannel.receive(); + assertEquals("foo-from-annotated-endpoint", response.getPayload()); + bus.stop(); } @@ -68,4 +85,14 @@ public class SynchronousChannelSubscriptionTests { } } + + @MessageEndpoint(input="sourceChannel", defaultOutput="targetChannel") + public static class TestEndpoint { + + @Handler + public Message handle(Message message) { + return new StringMessage(message.getPayload() + "-from-annotated-endpoint"); + } + } + }