From 27948795f0aa1e338b1f5bf5cbb8a48a43f4314c Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 3 Jan 2008 16:10:33 +0000 Subject: [PATCH] Factored out common source adapter behavior into AbstractSourceAdapter from PollingSourceAdapter and implemented ApplicationEventSourceAdapter. --- .../adapter/AbstractSourceAdapter.java | 65 ++++++++++ .../adapter/PollingSourceAdapter.java | 40 +----- .../integration/adapter/SourceAdapter.java | 3 - .../event/ApplicationEventSourceAdapter.java | 63 ++++++++++ .../integration/bus/MessageBus.java | 5 +- .../integration/bus/MessageDispatcher.java | 2 + .../bus/UnicastMessageDispatcher.java | 4 + .../ApplicationEventSourceAdapterTests.java | 115 ++++++++++++++++++ .../applicationEventSourceAdapterTests.xml | 15 +++ 9 files changed, 270 insertions(+), 42 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java new file mode 100644 index 0000000000..7c18432b23 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java @@ -0,0 +1,65 @@ +/* + * Copyright 2002-2007 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.adapter; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageMapper; +import org.springframework.integration.message.SimplePayloadMessageMapper; +import org.springframework.util.Assert; + +/** + * A base class providing common behavior for source adapters. + * + * @author Mark Fisher + */ +public class AbstractSourceAdapter implements SourceAdapter { + + private MessageChannel channel; + + private MessageMapper mapper = new SimplePayloadMessageMapper(); + + private long sendTimeout = -1; + + + public void setChannel(MessageChannel channel) { + Assert.notNull(channel, "'channel' must not be null"); + this.channel = channel; + } + + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + + public void setMessageMapper(MessageMapper mapper) { + Assert.notNull(mapper, "'mapper' must not be null"); + this.mapper = mapper; + } + + protected MessageMapper getMessageMapper() { + return this.mapper; + } + + protected boolean sendToChannel(T object) { + Message message = this.mapper.toMessage(object); + if (this.sendTimeout < 0) { + return this.channel.send(message); + } + return this.channel.send(message, this.sendTimeout); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index 3e6ce5f962..6578eec675 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -22,9 +22,7 @@ import org.springframework.integration.MessageHandlingException; import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.bus.MessageDispatcher; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageMapper; -import org.springframework.integration.message.SimplePayloadMessageMapper; import org.springframework.util.Assert; /** @@ -34,49 +32,25 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class PollingSourceAdapter implements SourceAdapter, MessageDispatcher { +public class PollingSourceAdapter extends AbstractSourceAdapter implements MessageDispatcher { private static int DEFAULT_PERIOD = 1000; private PollableSource source; - private MessageChannel channel; - - private MessageMapper mapper = new SimplePayloadMessageMapper(); - private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(DEFAULT_PERIOD); - private long sendTimeout = -1; - public PollingSourceAdapter(PollableSource source) { Assert.notNull(source, "'source' must not be null"); this.source = source; } - public void setChannel(MessageChannel channel) { - Assert.notNull(channel, "'channel' must not be null"); - this.channel = channel; - } - public void setPeriod(int period) { Assert.isTrue(period > 0, "'period' must be a positive value"); this.policy.setPeriod(period); } - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; - } - - public void setMessageMapper(MessageMapper mapper) { - Assert.notNull(mapper, "'mapper' must not be null"); - this.mapper = mapper; - } - - protected MessageMapper getMessageMapper() { - return this.mapper; - } - public void setMaxMessagesPerTask(int maxMessagesPerTask) { Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be a positive value"); this.policy.setMaxMessagesPerTask(maxMessagesPerTask); @@ -95,16 +69,8 @@ public class PollingSourceAdapter implements SourceAdapter, MessageDispatcher throw new MessageHandlingException("source returned too many results, the limit is " + limit); } for (T next : results) { - Message message = this.mapper.toMessage(next); - if (this.sendTimeout < 0) { - if (this.channel.send(message)) { - messagesProcessed++; - } - } - else { - if (this.channel.send(message, this.sendTimeout)) { - messagesProcessed++; - } + if (this.sendToChannel(next)) { + messagesProcessed++; } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java index 8d1ec65df7..5fe72ad111 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java @@ -16,7 +16,6 @@ package org.springframework.integration.adapter; -import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.MessageChannel; /** @@ -28,6 +27,4 @@ public interface SourceAdapter { void setChannel(MessageChannel channel); - ConsumerPolicy getConsumerPolicy(); - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java new file mode 100644 index 0000000000..78167c383a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java @@ -0,0 +1,63 @@ +/* + * Copyright 2002-2007 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.adapter.event; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationListener; +import org.springframework.integration.adapter.AbstractSourceAdapter; +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + +/** + * A source adapter for passing Spring + * {@link ApplicationEvent ApplicationEvents} within messages. + * + * @author Mark Fisher + */ +public class ApplicationEventSourceAdapter extends AbstractSourceAdapter implements + ApplicationListener { + + private List> eventTypes = new ArrayList>(); + + + /** + * Set the 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. + */ + public void setEventTypes(List> eventTypes) { + Assert.notEmpty(eventTypes, "at least one event type is required"); + this.eventTypes = eventTypes; + } + + public void onApplicationEvent(ApplicationEvent event) { + if (CollectionUtils.isEmpty(this.eventTypes)) { + this.sendToChannel(event); + return; + } + for (Class eventType : this.eventTypes) { + if (eventType.isAssignableFrom(event.getClass())) { + this.sendToChannel(event); + return; + } + } + } + +} 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 533bb1c541..ef069b4ddf 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 @@ -179,8 +179,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif public void registerSourceAdapter(String name, SourceAdapter adapter) { // TODO: use the name if (adapter instanceof MessageDispatcher) { - ConsumerPolicy policy = adapter.getConsumerPolicy(); - DispatcherTask dispatcherTask = new DispatcherTask((MessageDispatcher) adapter, policy); + MessageDispatcher dispatcher = (MessageDispatcher) adapter; + ConsumerPolicy policy = dispatcher.getConsumerPolicy(); + DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy); this.addDispatcherTask(dispatcherTask); if (logger.isInfoEnabled()) { logger.info("registered source adapter '" + name + "'"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java index 7e070d6399..e350f83aa8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java @@ -23,6 +23,8 @@ package org.springframework.integration.bus; */ public interface MessageDispatcher { + ConsumerPolicy getConsumerPolicy(); + int dispatch(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java index fd38e5a3ac..e5a721f310 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java @@ -38,6 +38,10 @@ public class UnicastMessageDispatcher extends AbstractMessageDispatcher { } + public ConsumerPolicy getConsumerPolicy() { + return this.policy; + } + @Override protected boolean dispatchMessage(Message message) { int attempts = 0; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java new file mode 100644 index 0000000000..b1d20b25fe --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java @@ -0,0 +1,115 @@ +/* + * Copyright 2002-2007 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.adapter.event; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Test; + +import org.springframework.context.ApplicationEvent; +import org.springframework.context.event.ContextClosedEvent; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.context.event.ContextStartedEvent; +import org.springframework.context.event.ContextStoppedEvent; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PointToPointChannel; +import org.springframework.integration.message.Message; + +/** + * @author Mark Fisher + */ +public class ApplicationEventSourceAdapterTests { + + @Test + public void testAnyApplicationEventSentByDefault() { + MessageChannel channel = new PointToPointChannel(); + ApplicationEventSourceAdapter adapter = new ApplicationEventSourceAdapter(); + adapter.setChannel(channel); + Message message1 = channel.receive(0); + assertNull(message1); + adapter.onApplicationEvent(new TestApplicationEvent1()); + adapter.onApplicationEvent(new TestApplicationEvent2()); + Message message2 = channel.receive(20); + assertNotNull(message2); + assertEquals("event1", ((ApplicationEvent) message2.getPayload()).getSource()); + Message message3 = channel.receive(20); + assertNotNull(message3); + assertEquals("event2", ((ApplicationEvent) message3.getPayload()).getSource()); + } + + @Test + public void testOnlyConfiguredEventTypesAreSent() { + MessageChannel channel = new PointToPointChannel(); + ApplicationEventSourceAdapter adapter = new ApplicationEventSourceAdapter(); + List> eventTypes = new ArrayList>(); + eventTypes.add(TestApplicationEvent1.class); + adapter.setEventTypes(eventTypes); + adapter.setChannel(channel); + Message message1 = channel.receive(0); + assertNull(message1); + adapter.onApplicationEvent(new TestApplicationEvent1()); + adapter.onApplicationEvent(new TestApplicationEvent2()); + Message message2 = channel.receive(20); + assertNotNull(message2); + assertEquals("event1", ((ApplicationEvent) message2.getPayload()).getSource()); + Message message3 = channel.receive(0); + assertNull(message3); + } + + @Test + public void testApplicationContextEvents() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("applicationEventSourceAdapterTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("channel"); + Message refreshedEventMessage = channel.receive(0); + assertNotNull(refreshedEventMessage); + assertEquals(ContextRefreshedEvent.class, refreshedEventMessage.getPayload().getClass()); + context.start(); + Message startedEventMessage = channel.receive(0); + assertNotNull(startedEventMessage); + assertEquals(ContextStartedEvent.class, startedEventMessage.getPayload().getClass()); + context.close(); + Message closedEventMessage = channel.receive(0); + assertNotNull(closedEventMessage); + assertEquals(ContextClosedEvent.class, closedEventMessage.getPayload().getClass()); + Message stoppedEventMessage = channel.receive(0); + assertNotNull(stoppedEventMessage); + assertEquals(ContextStoppedEvent.class, stoppedEventMessage.getPayload().getClass()); + } + + + private static class TestApplicationEvent1 extends ApplicationEvent { + + public TestApplicationEvent1() { + super("event1"); + } + } + + + private static class TestApplicationEvent2 extends ApplicationEvent { + + public TestApplicationEvent2() { + super("event2"); + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml new file mode 100644 index 0000000000..26f6ce8098 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml @@ -0,0 +1,15 @@ + + + + + + + + + + + +