Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1582 if the payload is an ApplicationEvent it is passed as-is (no wrapping in MessagingException). Also, general polishing

Browse Source
This commit is contained in:
Mark Fisher
2010-11-03 12:39:11 -04:00
parent d1babad811
commit 6cc758eefc
15 changed files with 238 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
docs/src/reference/docbook/event.xml
View File

@@ -13,21 +13,20 @@
<title>Receiving Spring ApplicationEvents</title>
<para>
To receive events and send them to a channel, simply define an instance of Spring Integration's
<classname>ApplicationEventListeningChannelAdapter</classname>. This class is an implementation of
<classname>ApplicationEventListeningMessageProducer</classname>. This class is an implementation of
Spring's <interfacename>ApplicationListener</interfacename> interface. By default it will pass all
received events as Spring Integration Messages. To limit based on the type of event, configure the
list of event types that you want to receive with the 'eventTypes' property.
</para>
<para>
For convenience namespace support was provided to configure <classname>ApplicationEventListeningChannelAdapter</classname> via <emphasis>inbound-channel-adapter</emphasis>
<programlisting language="xml"><![CDATA[<int-event:inbound-channel-adapter channel="input" event-types="foo.bar.FooApplicationEvent, foo.bar.BarApplicationEvent"/>
For convenience namespace support was provided to configure <classname>ApplicationEventListeningMessageProducer</classname> via <emphasis>inbound-channel-adapter</emphasis>
<programlisting language="xml"><![CDATA[<int-event:inbound-channel-adapter channel="eventChannel" event-types="example.FooEvent, example.BarEvent"/>
<int:publish-subscribe-channel id="sampleEventChannel"/>]]></programlisting>
In the above sample, all Application Context events that are of type specified by the 'event-types' (optional) attribute will be
delivered as Spring Integration Messages to 'sampleEventChannel'.
<int:publish-subscribe-channel id="eventChannel"/>]]></programlisting>
In the above example, all Application Context events that match one of the types specified by the 'event-types' (optional) attribute will be
delivered as Spring Integration Messages to 'eventChannel'.
</para>
</section>
<section id="applicationevent-outbound">
@@ -41,21 +40,21 @@ delivered as Spring Integration Messages to 'sampleEventChannel'.
</para>
<para>
For convenience namespace support was provided to configure <classname>ApplicationEventPublishingMessageHandler</classname> via <emphasis>outbound-channel-adapter</emphasis> element
<programlisting language="xml"><![CDATA[<int:channel id="input"/>
<programlisting language="xml"><![CDATA[<int:channel id="eventChannel"/>
<int-event:outbound-channel-adapter channel="input"/>]]></programlisting>
If you are using PollableChannel (e.g., Queue), you can also provide <emphasis>poller</emphasis> as sub-element of <emphasis>outbound-channel-adapter</emphasis>, optionally providing <emphasis>task-executor</emphasis>
<programlisting language="xml"><![CDATA[<int:channel id="input">
<int-event:outbound-channel-adapter channel="eventChannel"/>]]></programlisting>
If you are using a PollableChannel (e.g., Queue), you can also provide <emphasis>poller</emphasis> as a sub-element of the <emphasis>outbound-channel-adapter</emphasis> element. You can also optionally provide a <emphasis>task-executor</emphasis> reference for that poller.
<programlisting language="xml"><![CDATA[<int:channel id="eventChannel">
<int:queue/>
</int:channel>
<int-event:outbound-channel-adapter channel="input">
<int-event:outbound-channel-adapter channel="eventChannel">
<int:poller max-messages-per-poll="1" task-executor="executor" fixed-rate="100"/>
</int-event:outbound-channel-adapter>
<task:executor id="executor" pool-size="5"/>]]></programlisting>
In the above sample, all messages sent to an 'input' channel will be published as ApplicationEvents to Spring Application sContext
In the above example, all messages sent to the 'eventChannel' channel will be published as ApplicationEvents to any relevant ApplicationListeners within the Spring ApplicationContext. If the payload of the Message is an ApplicationEvent, it will be passed as-is. Otherwise the Message itself will be wrapped in a MessagingEvent instance.
</para>
</section>

16
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

@@ -57,6 +57,22 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
Assert.notNull(this.outputChannel, "outputChannel is required");
}
/**
* Takes no action by default. Subclasses may override this if they
* need lifecycle-managed behavior.
*/
@Override
protected void doStart() {
}
/**
* Takes no action by default. Subclasses may override this if they
* need lifecycle-managed behavior.
*/
@Override
protected void doStop() {
}
protected void sendMessage(Message<?> message) {
if (message == null) {
throw new MessagingException("cannot send a null message");

10
spring-integration-event/src/main/java/org/springframework/integration/event/config/EventInboundChannelAdapterParser.java
View File

@@ -16,23 +16,25 @@
package org.springframework.integration.event.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.event.ApplicationEventInboundChannelAdapter;
import org.w3c.dom.Element;
/**
* @author Oleg Zhurakousky
* @author Mark Fisher
* @since 2.0
*/
public class EventInboundChannelAdapterParser extends AbstractChannelAdapterParser{
public class EventInboundChannelAdapterParser extends AbstractChannelAdapterParser {
@Override
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.rootBeanDefinition(ApplicationEventInboundChannelAdapter.class);
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.rootBeanDefinition(
"org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(adapterBuilder, element, "channel", "outputChannel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "event-types");
IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "payload-expression");

2
spring-integration-event/src/main/java/org/springframework/integration/event/config/EventNamespaceHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 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.

31
spring-integration-event/src/main/java/org/springframework/integration/event/config/EventOutboundChannelAdapterParser.java
View File

@@ -13,44 +13,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.event.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.event.ApplicationEventPublishingMessageHandler;
import org.w3c.dom.Element;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
public class EventOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser{
@Override
protected AbstractBeanDefinition parseConsumer(Element element,
ParserContext parserContext) {
BeanDefinitionBuilder invokerBuilder = BeanDefinitionBuilder.genericBeanDefinition(ApplicationEventPublishingMessageHandler.class);
// BeanComponentDefinition innerHandlerDefinition =
// IntegrationNamespaceUtils.parseInnerHandlerDefinition(element, parserContext);
// if (innerHandlerDefinition == null){
// Assert.hasText(element.getAttribute(IntegrationNamespaceUtils.REF_ATTRIBUTE),
// "You must provide 'ref' attribute or register inner bean for " +
// "Outbound Channel consumer.");
// invokerBuilder.addConstructorArgReference(element.getAttribute(IntegrationNamespaceUtils.REF_ATTRIBUTE));
// } else {
// invokerBuilder.addConstructorArgValue(innerHandlerDefinition);
// }
// invokerBuilder.addConstructorArgValue(element.getAttribute(IntegrationNamespaceUtils.METHOD_ATTRIBUTE));
// String order = element.getAttribute(IntegrationNamespaceUtils.ORDER);
// if (StringUtils.hasText(order)) {
// invokerBuilder.addPropertyValue(IntegrationNamespaceUtils.ORDER, order);
// }
return invokerBuilder.getBeanDefinition();
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler");
return builder.getBeanDefinition();
}
}

2
spring-integration-event/src/main/java/org/springframework/integration/event/MessagingEvent.java → spring-integration-event/src/main/java/org/springframework/integration/event/core/MessagingEvent.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.event;
package org.springframework.integration.event.core;
import org.springframework.context.ApplicationEvent;
import org.springframework.integration.Message;

41
spring-integration-event/src/main/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapter.java → spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java
View File

@@ -14,13 +14,14 @@
* limitations under the License.
*/
package org.springframework.integration.event;
package org.springframework.integration.event.inbound;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ApplicationContextEvent;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.endpoint.MessageProducerSupport;
@@ -31,16 +32,18 @@ import org.springframework.util.CollectionUtils;
/**
* An inbound Channel Adapter that passes Spring {@link ApplicationEvent ApplicationEvents} within messages.
* If a {@link #setPayloadExpression(String) payloadExpression} is provided, it will be evaluated against
* the ApplicationEvent instance to create the Message payload.
* the ApplicationEvent instance to create the Message payload. Otherwise, the event itself will be the payload.
*
* @author Mark Fisher
*/
public class ApplicationEventInboundChannelAdapter extends MessageProducerSupport implements ApplicationListener<ApplicationEvent> {
public class ApplicationEventListeningMessageProducer extends MessageProducerSupport implements ApplicationListener<ApplicationEvent> {
private final Set<Class<? extends ApplicationEvent>> eventTypes = new CopyOnWriteArraySet<Class<? extends ApplicationEvent>>();
private volatile Expression payloadExpression;
private volatile boolean active;
private final SpelExpressionParser parser = new SpelExpressionParser();
@@ -77,29 +80,33 @@ public class ApplicationEventInboundChannelAdapter extends MessageProducerSuppor
}
public void onApplicationEvent(ApplicationEvent event) {
if (CollectionUtils.isEmpty(this.eventTypes)) {
this.sendEventAsMessage(event);
return;
}
for (Class<? extends ApplicationEvent> eventType : this.eventTypes) {
if (eventType.isAssignableFrom(event.getClass())) {
if (this.active || event instanceof ApplicationContextEvent) {
if (CollectionUtils.isEmpty(this.eventTypes)) {
this.sendEventAsMessage(event);
return;
}
for (Class<? extends ApplicationEvent> eventType : this.eventTypes) {
if (eventType.isAssignableFrom(event.getClass())) {
this.sendEventAsMessage(event);
return;
}
}
}
}
@Override
protected void doStart() {
this.active = true;
}
@Override
protected void doStop() {
this.active = false;
}
private void sendEventAsMessage(ApplicationEvent event) {
Object payload = (this.payloadExpression != null) ? this.payloadExpression.getValue(event) : event;
this.sendMessage(MessageBuilder.withPayload(payload).build());
}
@Override
protected void doStart() {
}
@Override
protected void doStop() {
}
}

12
spring-integration-event/src/main/java/org/springframework/integration/event/ApplicationEventPublishingMessageHandler.java → spring-integration-event/src/main/java/org/springframework/integration/event/outbound/ApplicationEventPublishingMessageHandler.java
View File

@@ -14,12 +14,13 @@
* limitations under the License.
*/
package org.springframework.integration.event;
package org.springframework.integration.event.outbound;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.integration.Message;
import org.springframework.integration.event.core.MessagingEvent;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.util.Assert;
@@ -31,7 +32,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class ApplicationEventPublishingMessageHandler<T> extends AbstractMessageHandler implements ApplicationEventPublisherAware {
public class ApplicationEventPublishingMessageHandler extends AbstractMessageHandler implements ApplicationEventPublisherAware {
private ApplicationEventPublisher applicationEventPublisher;
@@ -43,7 +44,12 @@ public class ApplicationEventPublishingMessageHandler<T> extends AbstractMessage
@Override
protected void handleMessageInternal(Message<?> message) {
Assert.notNull(this.applicationEventPublisher, "applicationEventPublisher is required");
this.applicationEventPublisher.publishEvent(new MessagingEvent(message));
if (message.getPayload() instanceof ApplicationEvent) {
this.applicationEventPublisher.publishEvent((ApplicationEvent) message.getPayload());
}
else {
this.applicationEventPublisher.publishEvent(new MessagingEvent(message));
}
}
}

172
spring-integration-event/src/main/resources/org/springframework/integration/event/config/spring-integration-event-2.0.xsd
View File

@@ -1,104 +1,98 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/event"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/event"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/event"
elementFormDefault="qualified" attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-2.0.xsd"/>
<xsd:import namespace="http://www.springframework.org/schema/beans" />
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-2.0.xsd" />
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for Spring Integration Event Adapters.
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for Spring Integration Event Adapters.
]]></xsd:documentation>
</xsd:annotation>
</xsd:annotation>
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Configures an inbound Channel Adapter which listens for an Application Context events, converts them to
Messages and sends them to a 'channel'
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID" use="optional" />
<xsd:attribute name="channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Identifies inbound 'channel' which accepts Messages generated from Application Context events.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="event-types" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
Comma delimited list of event types (classes that extend ApplicationEvent) that
this adapter should send to the message channel. By default, all event
types will be sent [OPTIONAL]
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="payload-expression">
<xsd:annotation>
<xsd:documentation><![CDATA[
SpEL expression to be evaluated against the ApplicationEvent to create the payload instance.
If not provided, the ApplicationEvent itself will be used as the payload.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-channel-adapter">
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines a Channel Adapter that receives from a MessageChannel and passes to
a method-invoking
MessageHandler.
Configures an inbound Channel Adapter which listens for Application Context
events, converts them to Messages and sends them to a Message Channel.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID" use="optional" />
<xsd:attribute name="channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.core.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
The channel to which Messages generated from Application Context events will be sent.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="event-types" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
Comma delimited list of event types (classes that extend ApplicationEvent) that this adapter
should send to the message channel. By default, all event types will be sent [OPTIONAL]
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="payload-expression">
<xsd:annotation>
<xsd:documentation><![CDATA[
SpEL expression to be evaluated against the ApplicationEvent to create the payload instance.
If not provided, the ApplicationEvent itself will be the payload.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:string" default="true" />
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines a Channel Adapter that receives Messages from a MessageChannel and then publishes
MessagingEvents containing those Messages.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="channelAdapterType">
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a
SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
<xsd:all>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
</xsd:all>
<xsd:attribute name="id" type="xsd:ID" />
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.core.MessageChannel" />
</tool:annotation>
<xsd:documentation>
The Message Channel from which this adapter receives Messages.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:string" default="true" />
</xsd:complexType>
</xsd:element>
<xsd:complexType name="channelAdapterType">
<xsd:all>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
</xsd:all>
<xsd:attribute name="id" type="xsd:ID" />
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:string" default="true" />
</xsd:complexType>
</xsd:schema>

10
spring-integration-event/src/test/java/org/springframework/integration/event/config/EventInboundChannelAdapterParserTests.java
View File

@@ -36,7 +36,7 @@ import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.expression.Expression;
import org.springframework.integration.Message;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.event.ApplicationEventInboundChannelAdapter;
import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
@@ -59,7 +59,7 @@ public class EventInboundChannelAdapterParserTests {
public void validateEventParser() {
Object adapter = context.getBean("eventAdapterSimple");
Assert.assertNotNull(adapter);
Assert.assertTrue(adapter instanceof ApplicationEventInboundChannelAdapter);
Assert.assertTrue(adapter instanceof ApplicationEventListeningMessageProducer);
DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
Assert.assertEquals(context.getBean("input"), adapterAccessor.getPropertyValue("outputChannel"));
}
@@ -69,7 +69,7 @@ public class EventInboundChannelAdapterParserTests {
public void validateEventParserWithEventTypes() {
Object adapter = context.getBean("eventAdapterFiltered");
Assert.assertNotNull(adapter);
Assert.assertTrue(adapter instanceof ApplicationEventInboundChannelAdapter);
Assert.assertTrue(adapter instanceof ApplicationEventListeningMessageProducer);
DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
Assert.assertEquals(context.getBean("inputFiltered"), adapterAccessor.getPropertyValue("outputChannel"));
Set<Class<? extends ApplicationEvent>> eventTypes = (Set<Class<? extends ApplicationEvent>>) adapterAccessor.getPropertyValue("eventTypes");
@@ -84,7 +84,7 @@ public class EventInboundChannelAdapterParserTests {
public void validateEventParserWithEventTypesAndPlaceholder() {
Object adapter = context.getBean("eventAdapterFilteredPlaceHolder");
Assert.assertNotNull(adapter);
Assert.assertTrue(adapter instanceof ApplicationEventInboundChannelAdapter);
Assert.assertTrue(adapter instanceof ApplicationEventListeningMessageProducer);
DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
Assert.assertEquals(context.getBean("inputFilteredPlaceHolder"), adapterAccessor.getPropertyValue("outputChannel"));
Set<Class<? extends ApplicationEvent>> eventTypes = (Set<Class<? extends ApplicationEvent>>) adapterAccessor.getPropertyValue("eventTypes");
@@ -113,7 +113,7 @@ public class EventInboundChannelAdapterParserTests {
public void validatePayloadExpression() {
Object adapter = context.getBean("eventAdapterSpel");
Assert.assertNotNull(adapter);
Assert.assertTrue(adapter instanceof ApplicationEventInboundChannelAdapter);
Assert.assertTrue(adapter instanceof ApplicationEventListeningMessageProducer);
DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
Expression expression = (Expression) adapterAccessor.getPropertyValue("payloadExpression");
Assert.assertEquals("source + '-test'", expression.getExpressionString());

57
spring-integration-event/src/test/java/org/springframework/integration/event/config/EventOutboundChannelAdapterParserTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.event.config;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import junit.framework.Assert;
@@ -34,7 +35,7 @@ import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.event.ApplicationEventPublishingMessageHandler;
import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
import org.springframework.integration.message.GenericMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -46,13 +47,15 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class EventOutboundChannelAdapterParserTests {
@Autowired
private ConfigurableApplicationContext context;
private boolean recievedEvent;
private volatile ConfigurableApplicationContext context;
private volatile boolean receivedEvent;
@Test
public void validateEventParser(){
public void validateEventParser() {
EventDrivenConsumer adapter = context.getBean("eventAdapter", EventDrivenConsumer.class);
Assert.assertNotNull(adapter);
DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
@@ -60,16 +63,16 @@ public class EventOutboundChannelAdapterParserTests {
Assert.assertTrue(handler instanceof ApplicationEventPublishingMessageHandler);
Assert.assertEquals(context.getBean("input"), adapterAccessor.getPropertyValue("inputChannel"));
}
@Test
public void validateUsage(){
ApplicationListener listener = new ApplicationListener<ApplicationEvent>() {
public void validateUsage() {
ApplicationListener<?> listener = new ApplicationListener<ApplicationEvent>() {
public void onApplicationEvent(ApplicationEvent event) {
Object source = event.getSource();
if (source instanceof Message){
String payload = (String) ((Message)source).getPayload();
if (payload.equals("hello")){
recievedEvent = true;
String payload = (String) ((Message<?>) source).getPayload();
if (payload.equals("hello")) {
receivedEvent = true;
}
}
}
@@ -77,34 +80,38 @@ public class EventOutboundChannelAdapterParserTests {
context.addApplicationListener(listener);
DirectChannel channel = context.getBean("input", DirectChannel.class);
channel.send(new GenericMessage<String>("hello"));
Assert.assertTrue(recievedEvent);
Assert.assertTrue(receivedEvent);
}
@Test(timeout=2000)
public void validateUsageWithPollableChannel() throws Exception {
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("EventOutboundChannelAdapterParserTestsWithPollable-context.xml", EventOutboundChannelAdapterParserTests.class);
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("EventOutboundChannelAdapterParserTestsWithPollable-context.xml", EventOutboundChannelAdapterParserTests.class);
final CyclicBarrier barier = new CyclicBarrier(2);
ApplicationListener listener = new ApplicationListener<ApplicationEvent>() {
ApplicationListener<?> listener = new ApplicationListener<ApplicationEvent>() {
public void onApplicationEvent(ApplicationEvent event) {
Object source = event.getSource();
if (source instanceof Message){
String payload = (String) ((Message)source).getPayload();
String payload = (String) ((Message<?>) source).getPayload();
if (payload.equals("hello")){
recievedEvent = true;
receivedEvent = true;
try {
barier.await();
} catch (Exception e) {
e.printStackTrace();
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
catch (BrokenBarrierException e) {
throw new IllegalStateException("broken barrier", e);
}
}
}
}
};
ac.addApplicationListener(listener);
QueueChannel channel = ac.getBean("input", QueueChannel.class);
context.addApplicationListener(listener);
QueueChannel channel = context.getBean("input", QueueChannel.class);
channel.send(new GenericMessage<String>("hello"));
barier.await();
Assert.assertTrue(recievedEvent);
Assert.assertTrue(receivedEvent);
}
}

4
spring-integration-event/src/test/java/org/springframework/integration/event/config/EventOutboundChannelAdapterParserTestsWithPollable-context.xml
View File

@@ -15,9 +15,7 @@
</int:channel>
<int-event:outbound-channel-adapter id="eventAdapter" channel="input">
<int:poller max-messages-per-poll="1" task-executor="executor">
<int:interval-trigger interval="100" time-unit="MILLISECONDS"/>
</int:poller>
<int:poller max-messages-per-poll="1" task-executor="executor" fixed-delay="100"/>
</int-event:outbound-channel-adapter>
<task:executor id="executor" pool-size="5"/>

25
spring-integration-event/src/test/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapterTests.java → spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.java
View File

@@ -14,9 +14,14 @@
* limitations under the License.
*/
package org.springframework.integration.event;
package org.springframework.integration.event.inbound;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import org.junit.Test;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.ContextRefreshedEvent;
@@ -26,21 +31,19 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.PollableChannel;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
/**
* @author Mark Fisher
*/
public class ApplicationEventInboundChannelAdapterTests {
public class ApplicationEventListeningMessageProducerTests {
@Test
public void anyApplicationEventSentByDefault() {
QueueChannel channel = new QueueChannel();
ApplicationEventInboundChannelAdapter adapter = new ApplicationEventInboundChannelAdapter();
ApplicationEventListeningMessageProducer adapter = new ApplicationEventListeningMessageProducer();
adapter.setOutputChannel(channel);
adapter.start();
Message<?> message1 = channel.receive(0);
assertNull(message1);
adapter.onApplicationEvent(new TestApplicationEvent1());
@@ -57,9 +60,10 @@ public class ApplicationEventInboundChannelAdapterTests {
@SuppressWarnings("unchecked")
public void onlyConfiguredEventTypesAreSent() {
QueueChannel channel = new QueueChannel();
ApplicationEventInboundChannelAdapter adapter = new ApplicationEventInboundChannelAdapter();
ApplicationEventListeningMessageProducer adapter = new ApplicationEventListeningMessageProducer();
adapter.setOutputChannel(channel);
adapter.setEventTypes(new Class[]{TestApplicationEvent1.class});
adapter.start();
Message<?> message1 = channel.receive(0);
assertNull(message1);
adapter.onApplicationEvent(new TestApplicationEvent1());
@@ -96,9 +100,10 @@ public class ApplicationEventInboundChannelAdapterTests {
@Test
public void payloadExpressionEvaluatedAgainstApplicationEvent() {
QueueChannel channel = new QueueChannel();
ApplicationEventInboundChannelAdapter adapter = new ApplicationEventInboundChannelAdapter();
ApplicationEventListeningMessageProducer adapter = new ApplicationEventListeningMessageProducer();
adapter.setPayloadExpression("'received: ' + source");
adapter.setOutputChannel(channel);
adapter.start();
Message<?> message1 = channel.receive(0);
assertNull(message1);
adapter.onApplicationEvent(new TestApplicationEvent1());
@@ -118,8 +123,6 @@ public class ApplicationEventInboundChannelAdapterTests {
public TestApplicationEvent1() {
super("event1");
}
}

2
spring-integration-event/src/test/java/org/springframework/integration/event/applicationEventInboundChannelAdapterTests.xml → spring-integration-event/src/test/java/org/springframework/integration/event/inbound/applicationEventInboundChannelAdapterTests.xml
View File

@@ -9,7 +9,7 @@
<int:queue capacity="5"/>
</int:channel>
<bean id="adapter" class="org.springframework.integration.event.ApplicationEventInboundChannelAdapter">
<bean id="adapter" class="org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer">
<property name="outputChannel" ref="channel"/>
</bean>

29
spring-integration-event/src/test/java/org/springframework/integration/event/ApplicationEventPublishingMessageHandlerTests.java → spring-integration-event/src/test/java/org/springframework/integration/event/outbound/ApplicationEventPublishingMessageHandlerTests.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.event;
package org.springframework.integration.event.outbound;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
@@ -24,6 +24,8 @@ import org.junit.Test;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.Message;
import org.springframework.integration.event.core.MessagingEvent;
import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
import org.springframework.integration.message.GenericMessage;
/**
@@ -32,8 +34,7 @@ import org.springframework.integration.message.GenericMessage;
public class ApplicationEventPublishingMessageHandlerTests {
@Test
@SuppressWarnings("unchecked")
public void testSendingEvent() throws InterruptedException {
public void messagingEvent() throws InterruptedException {
TestApplicationEventPublisher publisher = new TestApplicationEventPublisher();
ApplicationEventPublishingMessageHandler handler = new ApplicationEventPublishingMessageHandler();
handler.setApplicationEventPublisher(publisher);
@@ -45,6 +46,19 @@ public class ApplicationEventPublishingMessageHandlerTests {
assertEquals(message, ((MessagingEvent) event).getMessage());
}
@Test
public void payloadAsEvent() {
TestApplicationEventPublisher publisher = new TestApplicationEventPublisher();
ApplicationEventPublishingMessageHandler handler = new ApplicationEventPublishingMessageHandler();
handler.setApplicationEventPublisher(publisher);
assertNull(publisher.getLastEvent());
Message<?> message = new GenericMessage<TestEvent>(new TestEvent("foo"));
handler.handleMessage(message);
ApplicationEvent event = publisher.getLastEvent();
assertEquals(TestEvent.class, event.getClass());
assertEquals("foo", ((TestEvent) event).getSource());
}
private static class TestApplicationEventPublisher implements ApplicationEventPublisher {
@@ -59,4 +73,13 @@ public class ApplicationEventPublishingMessageHandlerTests {
}
}
@SuppressWarnings("serial")
private static class TestEvent extends ApplicationEvent {
public TestEvent(String text) {
super(text);
}
}
}