From 46e76afa536917cd4f7896640f6f2aac259c5cd9 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 8 Oct 2008 19:43:08 +0000 Subject: [PATCH] MessagingAnnotationPostProcessor now registers endpoints with the MessageBus so that they may be activated even if the bus has already detected endpoint beans from the context (INT-386). For an example, see: AnnotatedEndpointActivationTests. Although it does not impact this particular scenario, a related change is that the message bus no longer registers endpoint beans after receiving the application context refresh event. Instead it will retrieve all beans from the context when start() is invoked. It maintains a set of endpoints that also includes those registered only with the bus itself (not in the application context). --- .../integration/bus/DefaultMessageBus.java | 18 ++- .../MessagingAnnotationPostProcessor.java | 1 + ...notatedEndpointActivationTests-context.xml | 18 +++ .../AnnotatedEndpointActivationTests.java | 109 ++++++++++++++++++ 4 files changed, 135 insertions(+), 11 deletions(-) create mode 100644 org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests-context.xml create mode 100644 org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java index 8dce414879..9b4af0c914 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java @@ -66,6 +66,8 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A private final Map channels = new ConcurrentHashMap(); + private final Set endpoints = new CopyOnWriteArraySet(); + private final MessageBusInterceptorsList interceptors = new MessageBusInterceptorsList(); private final Set lifecycleGateways = new CopyOnWriteArraySet(); @@ -138,15 +140,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A } } - @SuppressWarnings("unchecked") - private void registerEndpoints(ApplicationContext context) { - Map endpointBeans = (Map) context - .getBeansOfType(MessageEndpoint.class); - for (Map.Entry entry : endpointBeans.entrySet()) { - this.registerEndpoint(entry.getValue()); - } - } - @SuppressWarnings("unchecked") private void registerGateways(ApplicationContext context) { Map gatewayBeans = (Map) context @@ -200,6 +193,9 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A public void registerEndpoint(MessageEndpoint endpoint) { Assert.notNull(endpoint, "'endpoint' must not be null"); + if (!this.endpoints.contains(endpoint)) { + this.endpoints.add(endpoint); + } if (this.isRunning()) { this.activateEndpoint(endpoint); } @@ -220,7 +216,8 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A private Collection getEndpoints() { GenericBeanFactoryAccessor accessor = new GenericBeanFactoryAccessor(this.applicationContext); - return accessor.getBeansOfType(MessageEndpoint.class).values(); + this.endpoints.addAll(accessor.getBeansOfType(MessageEndpoint.class).values()); + return this.endpoints; } private void activateEndpoints() { @@ -341,7 +338,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A if (event instanceof ContextRefreshedEvent) { ApplicationContext context = ((ContextRefreshedEvent) event).getApplicationContext(); this.registerChannels(context); - this.registerEndpoints(context); this.registerGateways(context); if (this.configureAsyncEventMulticaster) { this.initialize(); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java index bf4aa49ab5..4a9933e9dd 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java @@ -112,6 +112,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean ((BeanNameAware) result).setBeanName(endpointBeanName); } beanFactory.registerSingleton(endpointBeanName, result); + messageBus.registerEndpoint((MessageEndpoint) result); } } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests-context.xml new file mode 100644 index 0000000000..a1d47fde15 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests-context.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java new file mode 100644 index 0000000000..dbeacb7323 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java @@ -0,0 +1,109 @@ +/* + * Copyright 2002-2008 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.annotation; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + * @author Mark Fisher + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@MessageEndpoint +public class AnnotatedEndpointActivationTests { + + @Autowired + @Qualifier("input") + private MessageChannel input; + + @Autowired + @Qualifier("output") + private PollableChannel output; + + @Autowired + private MessageBus messageBus; + + // This has to be static because the MessageBus registers the handler + // more than once (every time a test instance is created), but only one of + // them will get the message. + private static volatile int count = 0; + + + @ServiceActivator(inputChannel = "input", outputChannel = "output") + public String process(String message) { + count++; + String result = message + ": " + count; + return result; + } + + @Before + public void resetCount() { + count = 0; + } + + @Test + public void configCheck() { + assertTrue(true); + } + + @Test + public void sendAndReceive() { + this.input.send(new GenericMessage("foo")); + Message message = this.output.receive(100); + assertNotNull(message); + assertEquals("foo: 1", message.getPayload()); + assertEquals(1, count); + } + + @Test(expected = MessageDeliveryException.class) + public void stopMessageBus() { + messageBus.stop(); + this.input.send(new GenericMessage("foo")); + } + + @Test + public void stopAndRestartMessageBus() { + messageBus.stop(); + messageBus.start(); + this.input.send(new GenericMessage("foo")); + Message message = this.output.receive(100); + assertNotNull(message); + assertEquals("foo: 1", message.getPayload()); + assertEquals(1, count); + } + +}