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