diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/IntegrationFlowBeanPostProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/IntegrationFlowBeanPostProcessor.java index 999b0f9..582d8fd 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/IntegrationFlowBeanPostProcessor.java @@ -25,6 +25,9 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanFactoryUtils; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.context.ApplicationListener; +import org.springframework.context.event.ApplicationEventMulticaster; +import org.springframework.context.support.AbstractApplicationContext; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.FixedSubscriberChannel; @@ -39,6 +42,7 @@ import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec; import org.springframework.integration.dsl.StandardIntegrationFlow; import org.springframework.integration.dsl.core.ConsumerEndpointSpec; import org.springframework.integration.dsl.support.MessageChannelReference; +import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer; import org.springframework.integration.support.context.NamedComponent; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -177,6 +181,15 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean .values() .contains(component)) { registerComponent(component, generateBeanName(component)); + if (ApplicationEventListeningMessageProducer.class.isAssignableFrom( + AopUtils.getTargetClass(component)) + && this.beanFactory.containsBean( + AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME)) { + ApplicationEventMulticaster multicaster = + (ApplicationEventMulticaster) this.beanFactory.getBean( + AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME); + multicaster.addApplicationListener((ApplicationListener) component); + } } } } diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/event/EventTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/event/EventTests.java new file mode 100644 index 0000000..750021b --- /dev/null +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/event/EventTests.java @@ -0,0 +1,121 @@ +/* + * Copyright 2014 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.dsl.test.event; + +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationListener; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class EventTests { + + @Autowired + private ApplicationContext applicationContext; + + @Autowired + private PollableChannel resultsChannel; + + @Test + public void testRawApplicationEventListeningMessageProducer() { + this.applicationContext.publishEvent(new TestApplicationEvent1()); + Message receive = this.resultsChannel.receive(10000); + assertNotNull(receive); + assertThat(receive.getPayload(), instanceOf(TestApplicationEvent1.class)); + + this.applicationContext.publishEvent(new TestApplicationEvent2()); + receive = this.resultsChannel.receive(10000); + assertNotNull(receive); + assertThat(receive.getPayload(), instanceOf(TestApplicationEvent2.class)); + } + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + + @Bean + @SuppressWarnings("unchecked") + public ApplicationListener applicationListener() { + ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer(); + producer.setEventTypes(TestApplicationEvent1.class); + producer.setOutputChannel(resultsChannel()); + return producer; + } + + + @Bean + public PollableChannel resultsChannel() { + return new QueueChannel(); + } + + @Bean + @SuppressWarnings("unchecked") + public IntegrationFlow eventProducerFlow() { + ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer(); + producer.setEventTypes(TestApplicationEvent2.class); + + return IntegrationFlows.from(producer) + .channel(resultsChannel()) + .get(); + } + + } + + @SuppressWarnings("serial") + private static class TestApplicationEvent1 extends ApplicationEvent { + + public TestApplicationEvent1() { + super("TestApplicationEvent1"); + } + + } + + @SuppressWarnings("serial") + private static class TestApplicationEvent2 extends ApplicationEvent { + + public TestApplicationEvent2() { + super("TestApplicationEvent2"); + } + + } + +}