From cfb0d7e2fd4f927fea11c8bdd8985c2ac7316d8b Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 24 Apr 2008 19:57:27 +0000 Subject: [PATCH] Added 'taskAdviceChain' and 'dispatchAdviceChain' to PollingSourceEndpoint. --- .../stream/ByteStreamTargetAdapterTests.java | 53 ++++---- .../CharacterStreamTargetAdapterTests.java | 49 +++---- spring-integration-core/ivy.xml | 2 +- .../integration/bus/SubscriptionManager.java | 30 +++-- .../dispatcher/DefaultPollingDispatcher.java | 97 ++++++++++++++ .../dispatcher/MessageDispatcher.java | 2 + .../dispatcher/PollingDispatcher.java | 88 +------------ .../dispatcher/PollingDispatcherTask.java | 60 +++++++++ .../endpoint/PollingSourceEndpoint.java | 73 ++++++++++- .../integration/message/Poller.java | 28 ++++ ...ava => DefaultPollingDispatcherTests.java} | 8 +- .../endpoint/PollingSourceEndpointTests.java | 120 ++++++++++++++++++ 12 files changed, 451 insertions(+), 159 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/message/Poller.java rename spring-integration-core/src/test/java/org/springframework/integration/dispatcher/{PollingDispatcherTests.java => DefaultPollingDispatcherTests.java} (92%) diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java index 23fbd3356a..2290fc6978 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java @@ -25,7 +25,8 @@ import org.junit.Test; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.dispatcher.DefaultPollingDispatcher; +import org.springframework.integration.dispatcher.PollingDispatcherTask; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; @@ -63,12 +64,12 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(3); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - dispatcher.run(); + task.run(); byte[] result = stream.toByteArray(); assertEquals(9, result.length); assertEquals(1, result[0]); @@ -82,12 +83,12 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - dispatcher.run(); + task.run(); byte[] result = stream.toByteArray(); assertEquals(6, result.length); assertEquals(1, result[0]); @@ -101,12 +102,12 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(5); dispatcherPolicy.setReceiveTimeout(0); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - dispatcher.run(); + task.run(); byte[] result = stream.toByteArray(); assertEquals(9, result.length); assertEquals(1, result[0]); @@ -120,16 +121,16 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - dispatcher.run(); + task.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); assertEquals(1, result1[0]); - dispatcher.run(); + task.run(); byte[] result2 = stream.toByteArray(); assertEquals(9, result2.length); assertEquals(1, result2[0]); @@ -144,16 +145,16 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(5); dispatcherPolicy.setReceiveTimeout(0); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - dispatcher.run(); + task.run(); byte[] result1 = stream.toByteArray(); assertEquals(9, result1.length); assertEquals(1, result1[0]); - dispatcher.run(); + task.run(); byte[] result2 = stream.toByteArray(); assertEquals(9, result2.length); assertEquals(1, result2[0]); @@ -167,16 +168,16 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - dispatcher.run(); + task.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); stream.reset(); - dispatcher.run(); + task.run(); byte[] result2 = stream.toByteArray(); assertEquals(3, result2.length); assertEquals(7, result2[0]); @@ -190,17 +191,17 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - dispatcher.run(); + task.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); stream.write(new byte[] {123}); stream.flush(); - dispatcher.run(); + task.run(); byte[] result2 = stream.toByteArray(); assertEquals(10, result2.length); assertEquals(1, result2[0]); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java index 3976de3af9..f06a73d32d 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java @@ -25,7 +25,8 @@ import org.junit.Test; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.dispatcher.DefaultPollingDispatcher; +import org.springframework.integration.dispatcher.PollingDispatcherTask; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; @@ -47,13 +48,13 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new QueueChannel(); StringWriter writer = new StringWriter(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - dispatcher.run(); + task.run(); assertEquals("foo", writer.toString()); - dispatcher.run(); + task.run(); assertEquals("foobar", writer.toString()); } @@ -63,14 +64,14 @@ public class CharacterStreamTargetAdapterTests { StringWriter writer = new StringWriter(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer); adapter.setShouldAppendNewLine(true); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - dispatcher.run(); + task.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine, writer.toString()); - dispatcher.run(); + task.run(); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } @@ -81,11 +82,11 @@ public class CharacterStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - dispatcher.run(); + task.run(); assertEquals("foobar", writer.toString()); } @@ -97,12 +98,12 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(10); dispatcherPolicy.setReceiveTimeout(0); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); adapter.setShouldAppendNewLine(true); - dispatcher.subscribe(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - dispatcher.run(); + task.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } @@ -112,11 +113,11 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new QueueChannel(); StringWriter writer = new StringWriter(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); TestObject testObject = new TestObject("foo"); channel.send(new GenericMessage(testObject)); - dispatcher.run(); + task.run(); assertEquals("foo", writer.toString()); } @@ -128,13 +129,13 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); channel.send(new GenericMessage(testObject2), 0); - dispatcher.run(); + task.run(); assertEquals("foobar", writer.toString()); } @@ -147,13 +148,13 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); QueueChannel channel = new QueueChannel(5, dispatcherPolicy); adapter.setShouldAppendNewLine(true); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); - dispatcher.subscribe(adapter); + PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null); + task.getDispatcher().subscribe(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); channel.send(new GenericMessage(testObject2), 0); - dispatcher.run(); + task.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } diff --git a/spring-integration-core/ivy.xml b/spring-integration-core/ivy.xml index 2cbf2238e0..bb502f277e 100644 --- a/spring-integration-core/ivy.xml +++ b/spring-integration-core/ivy.xml @@ -21,8 +21,8 @@ - + \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java index c6ef2cb4f8..689dfc72f7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java @@ -28,7 +28,8 @@ import org.apache.commons.logging.LogFactory; import org.springframework.context.Lifecycle; import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.dispatcher.DefaultPollingDispatcher; +import org.springframework.integration.dispatcher.PollingDispatcherTask; import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.endpoint.TargetEndpoint; import org.springframework.integration.message.MessagingException; @@ -55,7 +56,7 @@ public class SubscriptionManager { private volatile Schedule defaultSchedule = new PollingSchedule(5); - private final ConcurrentMap dispatchers = new ConcurrentHashMap(); + private final ConcurrentMap dispatcherTasks = new ConcurrentHashMap(); private final List lifecycleTargets = new CopyOnWriteArrayList(); @@ -119,21 +120,22 @@ public class SubscriptionManager { } return; } - PollingDispatcher dispatcher = this.dispatchers.get(schedule); - if (dispatcher == null) { - dispatcher = this.dispatchers.putIfAbsent(schedule, new PollingDispatcher(this.channel, schedule)); + PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule); + if (dispatcherTask == null) { + DefaultPollingDispatcher dispatcher = new DefaultPollingDispatcher(this.channel); + dispatcherTask = this.dispatcherTasks.putIfAbsent(schedule, new PollingDispatcherTask(dispatcher, schedule)); } - this.dispatchers.get(schedule).subscribe(target); - if (dispatcher == null && this.isRunning()) { + this.dispatcherTasks.get(schedule).getDispatcher().subscribe(target); + if (dispatcherTask == null && this.isRunning()) { this.scheduleDispatcherTask(schedule); } } public boolean removeTarget(Target target) { boolean removed = false; - Collection dispatcherValues = this.dispatchers.values(); - for (PollingDispatcher dispatcher : dispatcherValues) { - removed = (removed || dispatcher.unsubscribe(target)); + Collection dispatcherTaskValues = this.dispatcherTasks.values(); + for (PollingDispatcherTask dispatcherTask : dispatcherTaskValues) { + removed = (removed || dispatcherTask.getDispatcher().unsubscribe(target)); } return removed; } @@ -156,7 +158,7 @@ public class SubscriptionManager { for (Lifecycle target : lifecycleTargets) { target.start(); } - for (Schedule schedule : this.dispatchers.keySet()) { + for (Schedule schedule : this.dispatcherTasks.keySet()) { this.scheduleDispatcherTask(schedule); } this.running = true; @@ -164,9 +166,9 @@ public class SubscriptionManager { } private void scheduleDispatcherTask(Schedule schedule) { - PollingDispatcher dispatcher = this.dispatchers.get(schedule); - if (dispatcher != null) { - this.scheduler.schedule(dispatcher); + PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule); + if (dispatcherTask != null) { + this.scheduler.schedule(dispatcherTask); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java new file mode 100644 index 0000000000..68606d792f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java @@ -0,0 +1,97 @@ +/* + * 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.dispatcher; + +import java.util.LinkedList; +import java.util.List; + +import org.springframework.integration.channel.DispatcherPolicy; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.BlockingSource; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageDeliveryAware; +import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.integration.message.PollableSource; +import org.springframework.util.Assert; + +/** + * A subclass of {@link SimpleDispatcher} that adds message retrieval + * capabilities by polling a {@link PollableSource} for {@link Message Messages}. + * The number of messages retrieved per poll is limited by the 'maxMessagesPerTask' + * property of the provided {@link DispatcherPolicy}, and the timeout for each + * receive call is determined by the policy's 'receiveTimeout' + * property. In general, it is recommended to use a value of 1 (the default) for + * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise + * the poller may be holding on to available messages while waiting for + * additional messages. Note that the 'timeout' value is only relevant if the + * specified source is an implementation of {@link BlockingSource}. The default + * timeout value is 0 indicating that the method should return immediately + * rather than waiting for a {@link Message} to become available. + * + * @author Mark Fisher + */ +public class DefaultPollingDispatcher extends SimpleDispatcher implements PollingDispatcher { + + private final PollableSource source; + + + public DefaultPollingDispatcher(MessageChannel channel) { + this(channel, channel.getDispatcherPolicy()); + } + + public DefaultPollingDispatcher(PollableSource source, DispatcherPolicy dispatcherPolicy) { + super(dispatcherPolicy); + Assert.notNull(source, "source must not be null"); + this.source = source; + this.dispatcherPolicy.setReceiveTimeout(0); + } + + @Override + public boolean dispatch(Message message) { + boolean sent = super.dispatch(message); + if (this.source instanceof MessageDeliveryAware) { + if (sent) { + ((MessageDeliveryAware) this.source).onSend(message); + } + else { + ((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message")); + } + } + return sent; + } + + public List> poll() { + List> messages = new LinkedList>(); + int limit = this.dispatcherPolicy.getMaxMessagesPerTask(); + while (messages.size() < limit) { + Message message = null; + long timeout = this.dispatcherPolicy.getReceiveTimeout(); + if (this.source instanceof BlockingSource && timeout >= 0) { + message = ((BlockingSource) this.source).receive(timeout); + } + else { + message = this.source.receive(); + } + if (message == null) { + return messages; + } + messages.add(message); + } + return messages; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java index 3b1c70c67d..6bc4b74b13 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java @@ -27,4 +27,6 @@ public interface MessageDispatcher { boolean dispatch(Message message); + void setSendTimeout(long timeout); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java index d718630009..c1848ab70d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java @@ -16,94 +16,12 @@ package org.springframework.integration.dispatcher; -import java.util.LinkedList; -import java.util.List; - -import org.springframework.integration.channel.DispatcherPolicy; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.BlockingSource; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageDeliveryAware; -import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.PollableSource; -import org.springframework.integration.scheduling.MessagingTask; -import org.springframework.integration.scheduling.Schedule; -import org.springframework.util.Assert; +import org.springframework.integration.message.Poller; +import org.springframework.integration.message.SubscribableSource; /** - * A subclass of {@link SimpleDispatcher} that adds message retrieval - * capabilities and may be scheduled to run as a task. It polls a source for - * {@link Message Messages}. The number of messages retrieved per poll is - * limited by the 'maxMessagesPerTask' property of the provided - * {@link DispatcherPolicy}, and the timeout for each receive call is - * determined by the policy's 'receiveTimeout' property. In - * general, it is recommended to use a value of 1 (the default) for - * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise - * the poller may be holding on to available messages while waiting for - * additional messages. Note that the 'timeout' value is only relevant if the - * specified source is an implementation of {@link BlockingSource}. The default - * timeout value is 0 indicating that the method should return immediately - * rather than waiting for a {@link Message} to become available. - * * @author Mark Fisher */ -public class PollingDispatcher extends SimpleDispatcher implements MessagingTask { - - private final PollableSource source; - - private final Schedule schedule; - - - public PollingDispatcher(MessageChannel channel, Schedule schedule) { - this(channel, channel.getDispatcherPolicy(), schedule); - } - - public PollingDispatcher(PollableSource source, DispatcherPolicy dispatcherPolicy, Schedule schedule) { - super(dispatcherPolicy); - Assert.notNull(source, "source must not be null"); - this.source = source; - this.schedule = schedule; - this.dispatcherPolicy.setReceiveTimeout(0); - } - - - public Schedule getSchedule() { - return this.schedule; - } - - public List> poll() { - List> messages = new LinkedList>(); - int limit = this.dispatcherPolicy.getMaxMessagesPerTask(); - while (messages.size() < limit) { - Message message = null; - long timeout = this.dispatcherPolicy.getReceiveTimeout(); - if (this.source instanceof BlockingSource && timeout >= 0) { - message = ((BlockingSource) this.source).receive(timeout); - } - else { - message = this.source.receive(); - } - if (message == null) { - return messages; - } - messages.add(message); - } - return messages; - } - - public void run() { - List> messages = this.poll(); - for (Message message : messages) { - boolean sent = this.dispatch(message); - if (this.source instanceof MessageDeliveryAware) { - if (sent) { - ((MessageDeliveryAware) this.source).onSend(message); - } - else { - ((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message")); - } - } - } - } +public interface PollingDispatcher extends Poller, MessageDispatcher, SubscribableSource { } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java new file mode 100644 index 0000000000..798c5cc140 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java @@ -0,0 +1,60 @@ +/* + * 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.dispatcher; + +import java.util.List; + +import org.springframework.integration.message.Message; +import org.springframework.integration.scheduling.MessagingTask; +import org.springframework.integration.scheduling.Schedule; +import org.springframework.util.Assert; + +/** + * A {@link MessagingTask} that combines polling and dispatching. + * + * @author Mark Fisher + */ +public class PollingDispatcherTask implements MessagingTask { + + private final PollingDispatcher dispatcher; + + private final Schedule schedule; + + + public PollingDispatcherTask(PollingDispatcher dispatcher, Schedule schedule) { + Assert.notNull(dispatcher, "dispatcher must not be null"); + this.dispatcher = dispatcher; + this.schedule = schedule; + } + + + public PollingDispatcher getDispatcher() { + return this.dispatcher; + } + + public Schedule getSchedule() { + return this.schedule; + } + + public void run() { + List> messages = this.dispatcher.poll(); + for (Message message : messages) { + this.dispatcher.dispatch(message); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java index 179afd07d3..f84ceec665 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java @@ -16,9 +16,19 @@ package org.springframework.integration.endpoint; +import java.lang.reflect.Method; +import java.util.List; + +import org.aopalliance.aop.Advice; + +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.aop.support.StaticMethodMatcherPointcutAdvisor; +import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.dispatcher.DefaultPollingDispatcher; import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.dispatcher.PollingDispatcherTask; import org.springframework.integration.message.PollableSource; import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.PollingSchedule; @@ -31,21 +41,60 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class PollingSourceEndpoint extends AbstractSourceEndpoint implements MessagingTask { +public class PollingSourceEndpoint extends AbstractSourceEndpoint implements MessagingTask, InitializingBean { private final DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - private final PollingDispatcher dispatcher; + private volatile PollingDispatcher dispatcher; + + private volatile MessagingTask task; + + private volatile List taskAdviceChain; + + private volatile List dispatchAdviceChain; public PollingSourceEndpoint(PollableSource source, MessageChannel channel, PollingSchedule schedule) { super(source, channel); Assert.notNull(schedule, "schedule must not be null"); - this.dispatcher = new PollingDispatcher(source, this.dispatcherPolicy, schedule); + this.dispatcher = new DefaultPollingDispatcher(source, this.dispatcherPolicy); this.dispatcher.subscribe(this.getChannel()); + this.task = new PollingDispatcherTask(this.dispatcher, schedule); } + public void setTaskAdviceChain(List taskAdviceChain) { + this.taskAdviceChain = taskAdviceChain; + } + + public void setDispatchAdviceChain(List dispatchAdviceChain) { + this.dispatchAdviceChain = dispatchAdviceChain; + } + + public void afterPropertiesSet() { + this.initializeProxies(); + } + + public void initializeProxies() { + if (this.dispatchAdviceChain != null && this.dispatchAdviceChain.size() > 0) { + ProxyFactory proxyFactory = new ProxyFactory(this.dispatcher); + proxyFactory.setInterfaces(new Class[] { PollingDispatcher.class }); + for (Advice advice : this.dispatchAdviceChain) { + proxyFactory.addAdvisor(new DispatchMethodAdvisor(advice)); + } + this.dispatcher = (PollingDispatcher) proxyFactory.getProxy(); + this.task = new PollingDispatcherTask(this.dispatcher, this.task.getSchedule()); + } + if (this.taskAdviceChain != null && this.taskAdviceChain.size() > 0) { + ProxyFactory proxyFactory = new ProxyFactory(this.task); + proxyFactory.setInterfaces(new Class[] { MessagingTask.class }); + for (Advice advice : this.taskAdviceChain) { + proxyFactory.addAdvice(advice); + } + this.task = (MessagingTask) proxyFactory.getProxy(); + } + } + public void setMaxMessagesPerTask(int maxMessagesPerTask) { this.dispatcherPolicy.setMaxMessagesPerTask(maxMessagesPerTask); } @@ -55,11 +104,25 @@ public class PollingSourceEndpoint extends AbstractSourceEndpoint implements Mes } public Schedule getSchedule() { - return this.dispatcher.getSchedule(); + return this.task.getSchedule(); } public void run() { - this.dispatcher.run(); + this.task.run(); + } + + + @SuppressWarnings("serial") + private static class DispatchMethodAdvisor extends StaticMethodMatcherPointcutAdvisor { + + DispatchMethodAdvisor(Advice advice) { + super(advice); + } + + @SuppressWarnings("unchecked") + public boolean matches(Method method, Class targetClass) { + return method.getName().equals("dispatch"); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/Poller.java b/spring-integration-core/src/main/java/org/springframework/integration/message/Poller.java new file mode 100644 index 0000000000..5c7dc44d45 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/Poller.java @@ -0,0 +1,28 @@ +/* + * 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.message; + +import java.util.List; + +/** + * @author Mark Fisher + */ +public interface Poller { + + List> poll(); + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultPollingDispatcherTests.java similarity index 92% rename from spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultPollingDispatcherTests.java index a92106f0fc..496eb542d9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultPollingDispatcherTests.java @@ -33,14 +33,14 @@ import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher */ -public class PollingDispatcherTests { +public class DefaultPollingDispatcherTests { @Test public void testSingleMessagePerRetrieval() { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); MessageChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + DefaultPollingDispatcher dispatcher = new DefaultPollingDispatcher(channel); Collection> results = dispatcher.poll(); assertTrue(results.isEmpty()); channel.send(new StringMessage("test1"), 0); @@ -59,7 +59,7 @@ public class PollingDispatcherTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); MessageChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + DefaultPollingDispatcher dispatcher = new DefaultPollingDispatcher(channel); Collection> results = dispatcher.poll(); assertTrue(results.isEmpty()); channel.send(new StringMessage("test1"), 0); @@ -81,7 +81,7 @@ public class PollingDispatcherTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(1); MessageChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + DefaultPollingDispatcher dispatcher = new DefaultPollingDispatcher(channel); Collection> results = dispatcher.poll(); assertTrue(results.isEmpty()); channel.send(new StringMessage("test1"), 0); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java index b687e221a5..efcafb9f2d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java @@ -20,10 +20,18 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.atomic.AtomicInteger; +import org.aopalliance.aop.Advice; +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; + import org.junit.Test; +import org.springframework.aop.MethodBeforeAdvice; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; @@ -90,6 +98,118 @@ public class PollingSourceEndpointTests { assertNull("message should be null", message4); } + @Test + public void testTaskAdviceChain() { + TestSource source = new TestSource("testing", 3); + QueueChannel channel = new QueueChannel(); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + final StringBuffer buffer = new StringBuffer(); + List taskAdviceChain = new ArrayList(); + taskAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append(1); + } + }); + taskAdviceChain.add(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + buffer.append(2); + Object retval = invocation.proceed(); + buffer.append(4); + return retval; + } + }); + taskAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append(3); + } + }); + endpoint.setTaskAdviceChain(taskAdviceChain); + endpoint.afterPropertiesSet(); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); + assertEquals("1234", buffer.toString()); + } + + @Test + public void testDispatchAdviceChain() { + TestSource source = new TestSource("testing", 2); + QueueChannel channel = new QueueChannel(); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + final StringBuffer buffer = new StringBuffer(); + List dispatchAdviceChain = new ArrayList(); + dispatchAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append("a"); + } + }); + dispatchAdviceChain.add(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + buffer.append("b"); + Object retval = invocation.proceed(); + buffer.append("d"); + return retval; + } + }); + dispatchAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append("c"); + } + }); + endpoint.setDispatchAdviceChain(dispatchAdviceChain); + endpoint.afterPropertiesSet(); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); + assertEquals("abcdabcd", buffer.toString()); + } + + @Test + public void testTaskAndDispatchAdviceChains() { + TestSource source = new TestSource("testing", 3); + QueueChannel channel = new QueueChannel(); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + List dispatchAdviceChain = new ArrayList(); + List taskAdviceChain = new ArrayList(); + final StringBuffer buffer = new StringBuffer(); + dispatchAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append("a"); + } + }); + dispatchAdviceChain.add(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + buffer.append("b"); + Object retval = invocation.proceed(); + buffer.append("c"); + return retval; + } + }); + taskAdviceChain.add(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + buffer.append(1); + Object retval = invocation.proceed(); + buffer.append(3); + return retval; + } + }); + taskAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append(2); + } + }); + endpoint.setTaskAdviceChain(taskAdviceChain); + endpoint.setDispatchAdviceChain(dispatchAdviceChain); + endpoint.afterPropertiesSet(); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); + assertEquals("12abcabcabc3", buffer.toString()); + } + private static class TestSource implements PollableSource {