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 1962739665..1a90dc0108 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,11 +25,9 @@ import org.junit.Test; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.dispatcher.DefaultMessageDispatcher; -import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.dispatcher.PollingDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; -import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; /** * @author Mark Fisher @@ -65,13 +63,12 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(3); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.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); - assertEquals(3, dispatcher.dispatch()); + dispatcher.run(); byte[] result = stream.toByteArray(); assertEquals(9, result.length); assertEquals(1, result[0]); @@ -85,13 +82,12 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.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); - assertEquals(2, dispatcher.dispatch()); + dispatcher.run(); byte[] result = stream.toByteArray(); assertEquals(6, result.length); assertEquals(1, result[0]); @@ -105,13 +101,12 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(5); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.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); - assertEquals(3, dispatcher.dispatch()); + dispatcher.run(); byte[] result = stream.toByteArray(); assertEquals(9, result.length); assertEquals(1, result[0]); @@ -125,17 +120,16 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.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); - assertEquals(2, dispatcher.dispatch()); + dispatcher.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); assertEquals(1, result1[0]); - assertEquals(1, dispatcher.dispatch()); + dispatcher.run(); byte[] result2 = stream.toByteArray(); assertEquals(9, result2.length); assertEquals(1, result2[0]); @@ -150,17 +144,16 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(5); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.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); - assertEquals(3, dispatcher.dispatch()); + dispatcher.run(); byte[] result1 = stream.toByteArray(); assertEquals(9, result1.length); assertEquals(1, result1[0]); - assertEquals(0, dispatcher.dispatch()); + dispatcher.run(); byte[] result2 = stream.toByteArray(); assertEquals(9, result2.length); assertEquals(1, result2[0]); @@ -174,17 +167,16 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.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); - assertEquals(2, dispatcher.dispatch()); + dispatcher.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); stream.reset(); - assertEquals(1, dispatcher.dispatch()); + dispatcher.run(); byte[] result2 = stream.toByteArray(); assertEquals(3, result2.length); assertEquals(7, result2[0]); @@ -198,18 +190,17 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.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); - assertEquals(2, dispatcher.dispatch()); + dispatcher.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); stream.write(new byte[] {123}); stream.flush(); - assertEquals(1, dispatcher.dispatch()); + dispatcher.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 d0f6d4a633..89e928154b 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,20 +25,15 @@ import org.junit.Test; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.dispatcher.DefaultMessageDispatcher; -import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.dispatcher.PollingDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; -import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; /** * @author Mark Fisher */ public class CharacterStreamTargetAdapterTests { - private SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - - @Test public void testSingleString() { StringWriter writer = new StringWriter(); @@ -52,13 +47,13 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new SimpleChannel(); StringWriter writer = new StringWriter(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.subscribe(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - assertEquals(1, dispatcher.dispatch()); + dispatcher.run(); assertEquals("foo", writer.toString()); - assertEquals(1, dispatcher.dispatch()); + dispatcher.run(); assertEquals("foobar", writer.toString()); } @@ -68,14 +63,14 @@ public class CharacterStreamTargetAdapterTests { StringWriter writer = new StringWriter(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer); adapter.setShouldAppendNewLine(true); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.subscribe(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - assertEquals(1, dispatcher.dispatch()); + dispatcher.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine, writer.toString()); - assertEquals(1, dispatcher.dispatch()); + dispatcher.run(); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } @@ -86,11 +81,11 @@ public class CharacterStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.subscribe(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - assertEquals(2, dispatcher.dispatch()); + dispatcher.run(); assertEquals("foobar", writer.toString()); } @@ -102,12 +97,12 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(10); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); adapter.setShouldAppendNewLine(true); - dispatcher.addTarget(adapter); + dispatcher.subscribe(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - assertEquals(2, dispatcher.dispatch()); + dispatcher.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } @@ -117,12 +112,11 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new SimpleChannel(); StringWriter writer = new StringWriter(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.subscribe(adapter); TestObject testObject = new TestObject("foo"); channel.send(new GenericMessage(testObject)); - int count = dispatcher.dispatch(); - assertEquals(1, count); + dispatcher.run(); assertEquals("foo", writer.toString()); } @@ -134,13 +128,13 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.subscribe(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); channel.send(new GenericMessage(testObject2), 0); - assertEquals(2, dispatcher.dispatch()); + dispatcher.run(); assertEquals("foobar", writer.toString()); } @@ -152,14 +146,14 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); adapter.setShouldAppendNewLine(true); - dispatcher.addTarget(adapter); + PollingDispatcher dispatcher = new PollingDispatcher(channel, null); + dispatcher.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.dispatch(); + dispatcher.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } 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 99d7e3fcd1..7973e3cbeb 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 @@ -39,8 +39,6 @@ import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.dispatcher.DefaultMessageDispatcher; -import org.springframework.integration.dispatcher.SchedulingMessageDispatcher; import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultEndpointRegistry; @@ -79,7 +77,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio private final EndpointRegistry endpointRegistry = new DefaultEndpointRegistry(); - private final Map dispatchers = new ConcurrentHashMap(); + private final Map subscriptionManagers = new ConcurrentHashMap(); private final List lifecycleSourceAdapters = new CopyOnWriteArrayList(); @@ -213,8 +211,8 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio this.initialize(); } channel.setName(name); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, this.taskScheduler); - this.dispatchers.put(channel, dispatcher); + SubscriptionManager manager = new SubscriptionManager(channel, this.taskScheduler); + this.subscriptionManagers.put(channel, manager); this.channelRegistry.registerChannel(name, channel); if (logger.isInfoEnabled()) { logger.info("registered channel '" + name + "'"); @@ -224,9 +222,9 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio public MessageChannel unregisterChannel(String name) { MessageChannel removedChannel = this.channelRegistry.unregisterChannel(name); if (removedChannel != null) { - SchedulingMessageDispatcher removedDispatcher = this.dispatchers.remove(removedChannel); - if (removedDispatcher != null && removedDispatcher.isRunning()) { - removedDispatcher.stop(); + SubscriptionManager manager = this.subscriptionManagers.remove(removedChannel); + if (manager != null && manager.isRunning()) { + manager.stop(); } } return removedChannel; @@ -287,10 +285,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio if (endpoint == null) { return null; } - Collection dispatchers = this.dispatchers.values(); + Collection managers = this.subscriptionManagers.values(); boolean removed = false; - for (SchedulingMessageDispatcher dispatcher : dispatchers) { - removed = (removed || dispatcher.removeTarget(endpoint)); + for (SubscriptionManager manager : managers) { + removed = (removed || manager.removeTarget(endpoint)); } if (removed) { return endpoint; @@ -387,7 +385,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio private void registerWithDispatcher(MessageChannel channel, Target target, Schedule schedule) { if (schedule == null && (channel instanceof SynchronousChannel)) { - ((SynchronousChannel) channel).addTarget(target); + ((SynchronousChannel) channel).subscribe(target); if (target instanceof Lifecycle) { ((Lifecycle) target).start(); } @@ -403,16 +401,16 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio } return; } - SchedulingMessageDispatcher dispatcher = dispatchers.get(channel); - if (dispatcher == null) { + SubscriptionManager manager = subscriptionManagers.get(channel); + if (manager == null) { if (logger.isWarnEnabled()) { - logger.warn("no dispatcher available for channel '" + channel.getName() + "', be sure to register the channel"); + logger.warn("no subscription manager available for channel '" + channel.getName() + "', be sure to register the channel"); } return; } - dispatcher.addTarget(target, schedule); - if (this.isRunning() && !dispatcher.isRunning()) { - dispatcher.start(); + manager.addTarget(target, schedule); + if (this.isRunning() && !manager.isRunning()) { + manager.start(); } } @@ -433,10 +431,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio synchronized (this.lifecycleMonitor) { this.activateEndpoints(); this.taskScheduler.start(); - for (SchedulingMessageDispatcher dispatcher : this.dispatchers.values()) { - dispatcher.start(); + for (SubscriptionManager manager : this.subscriptionManagers.values()) { + manager.start(); if (logger.isInfoEnabled()) { - logger.info("started dispatcher '" + dispatcher + "'"); + logger.info("started subscription manager '" + manager + "'"); } } for (Lifecycle adapter : this.lifecycleSourceAdapters) { @@ -466,10 +464,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio logger.info("stopped source adapter '" + adapter + "'"); } } - for (SchedulingMessageDispatcher dispatcher : this.dispatchers.values()) { - dispatcher.stop(); + for (SubscriptionManager manager : this.subscriptionManagers.values()) { + manager.stop(); if (logger.isInfoEnabled()) { - logger.info("stopped dispatcher '" + dispatcher + "'"); + logger.info("stopped subscription manager '" + manager + "'"); } } } 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 new file mode 100644 index 0000000000..ff3949f066 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java @@ -0,0 +1,162 @@ +/* + * 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.bus; + +import java.util.Collection; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.commons.logging.Log; +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.message.Target; +import org.springframework.integration.scheduling.MessagingTaskScheduler; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Schedule; +import org.springframework.util.Assert; + +/** + * Manages subscriptions for {@link Target Targets} to a {@link MessageChannel} + * including the creation, scheduling, and lifecycle management of dispatchers. + * + * @author Mark Fisher + */ +public class SubscriptionManager { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final MessageChannel channel; + + private final MessagingTaskScheduler scheduler; + + private volatile Schedule defaultSchedule = new PollingSchedule(5); + + private final ConcurrentMap dispatchers = new ConcurrentHashMap(); + + private final List lifecycleTargets = new CopyOnWriteArrayList(); + + private volatile boolean running; + + private final Object lifecycleMonitor = new Object(); + + + public SubscriptionManager(MessageChannel channel, MessagingTaskScheduler scheduler) { + Assert.notNull(channel, "channel must not be null"); + Assert.notNull(scheduler, "scheduler must not be null"); + this.channel = channel; + this.scheduler = scheduler; + } + + + public void setDefaultSchedule(Schedule defaultSchedule) { + Assert.notNull(defaultSchedule, "'defaultSchedule' must not be null"); + this.defaultSchedule = defaultSchedule; + } + + public void addTarget(Target target) { + this.addTarget(target, null); + } + + public void addTarget(Target target, Schedule schedule) { + Assert.notNull(target, "'target' must not be null"); + if (schedule == null) { + schedule = this.defaultSchedule; + } + else if (this.channel.getDispatcherPolicy().isPublishSubscribe()) { + if (logger.isInfoEnabled()) { + logger.info("This dispatcher broadcasts messages for a publish-subscribe channel. " + + "Therefore all targets are scheduled with its 'defaultSchedule', " + + "and the provided schedule will be ignored."); + } + schedule = this.defaultSchedule; + } + if (target instanceof Lifecycle) { + this.lifecycleTargets.add((Lifecycle) target); + if (this.isRunning()) { + ((Lifecycle) target).start(); + } + } + PollingDispatcher dispatcher = this.dispatchers.get(schedule); + if (dispatcher == null) { + dispatcher = this.dispatchers.putIfAbsent(schedule, new PollingDispatcher(this.channel, schedule)); + } + this.dispatchers.get(schedule).subscribe(target); + if (dispatcher == 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)); + } + return removed; + } + + public boolean isRunning() { + return this.running; + } + + public void start() { + synchronized (this.lifecycleMonitor) { + if (this.running) { + return; + } + if (this.scheduler == null) { + throw new ConfigurationException("scheduler is required"); + } + if (!this.scheduler.isRunning()) { + this.scheduler.start(); + } + for (Lifecycle target : lifecycleTargets) { + target.start(); + } + for (Schedule schedule : this.dispatchers.keySet()) { + this.scheduleDispatcherTask(schedule); + } + this.running = true; + } + } + + private void scheduleDispatcherTask(Schedule schedule) { + PollingDispatcher dispatcher = this.dispatchers.get(schedule); + if (dispatcher != null) { + this.scheduler.schedule(dispatcher); + } + } + + public void stop() { + synchronized (this.lifecycleMonitor) { + if (!this.running) { + return; + } + for (Lifecycle target : lifecycleTargets) { + target.stop(); + } + this.running = false; + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java deleted file mode 100644 index 1bc8eb503d..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java +++ /dev/null @@ -1,69 +0,0 @@ -/* - * 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.Collection; -import java.util.LinkedList; -import java.util.List; - -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.Message; -import org.springframework.util.Assert; - -/** - * Message retriever that polls a {@link MessageChannel}. The number of - * messages retrieved per poll is limited by the 'maxMessagesPerTask' - * property of the channel's dispatcher policy, and the timeout for each receive - * call is determined by the dispatcher policy's 'receiveTimeout' - * property. In general, it is recommended to use a value of 1 for - * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise the - * retriever may be holding on to available messages while waiting for - * additional messages. - * - * @author Mark Fisher - */ -public class ChannelPollingMessageRetriever implements MessageRetriever { - - private MessageChannel channel; - - - public ChannelPollingMessageRetriever(MessageChannel channel) { - Assert.notNull(channel, "'channel' must not be null"); - this.channel = channel; - } - - - public Collection> retrieveMessages() { - List> messages = new LinkedList>(); - while (messages.size() < this.channel.getDispatcherPolicy().getMaxMessagesPerTask()) { - Message message = null; - long timeout = this.channel.getDispatcherPolicy().getReceiveTimeout(); - if (timeout < 0) { - message = this.channel.receive(); - } - else { - message = this.channel.receive(timeout); - } - if (message == null) { - return messages; - } - messages.add(message); - } - return messages; - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java deleted file mode 100644 index 8f8312cc36..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java +++ /dev/null @@ -1,221 +0,0 @@ -/* - * 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.Collection; -import java.util.List; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.atomic.AtomicLong; - -import org.apache.commons.logging.Log; -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.message.Message; -import org.springframework.integration.message.Target; -import org.springframework.integration.scheduling.MessagingTask; -import org.springframework.integration.scheduling.MessagingTaskScheduler; -import org.springframework.integration.scheduling.PollingSchedule; -import org.springframework.integration.scheduling.Schedule; -import org.springframework.util.Assert; - -/** - * Default implementation of {@link MessageDispatcher}. For a non-broadcasting - * {@link MessageChannel} (point-to-point), each handler can be registered with - * a {@link Schedule}. If the channel is broadcasting (publish-subscribe), the - * handlers will all be scheduled together according to the dispatcher's - * {@link #defaultSchedule}. - * - * @author Mark Fisher - */ -public class DefaultMessageDispatcher implements SchedulingMessageDispatcher { - - protected final Log logger = LogFactory.getLog(this.getClass()); - - private final MessageChannel channel; - - private final MessageRetriever retriever; - - private final MessagingTaskScheduler scheduler; - - private volatile Schedule defaultSchedule = new PollingSchedule(5); - - private final ConcurrentMap> scheduledTargets = new ConcurrentHashMap>(); - - private final AtomicLong totalMessagesProcessed = new AtomicLong(); - - private volatile boolean running; - - private final Object lifecycleMonitor = new Object(); - - - public DefaultMessageDispatcher(MessageChannel channel, MessagingTaskScheduler scheduler) { - Assert.notNull(channel, "'channel' must not be null"); - Assert.notNull(scheduler, "'scheduler' must not be null"); - this.channel = channel; - this.scheduler = scheduler; - this.retriever = new ChannelPollingMessageRetriever(this.channel); - } - - - public void setDefaultSchedule(Schedule defaultSchedule) { - Assert.notNull(defaultSchedule, "'defaultSchedule' must not be null"); - this.defaultSchedule = defaultSchedule; - } - - public void addTarget(Target target) { - this.addTarget(target, null); - } - - public void addTarget(Target target, Schedule schedule) { - Assert.notNull(target, "'target' must not be null"); - if (schedule == null) { - schedule = this.defaultSchedule; - } - else if (this.channel.getDispatcherPolicy().isPublishSubscribe()) { - if (logger.isInfoEnabled()) { - logger.info("This dispatcher broadcasts messages for a publish-subscribe channel. " + - "Therefore all targets are scheduled with its 'defaultSchedule', " + - "and the provided schedule will be ignored."); - } - schedule = this.defaultSchedule; - } - if (this.isRunning() && target instanceof Lifecycle) { - ((Lifecycle) target).start(); - } - List targets = this.scheduledTargets.get(schedule); - if (targets == null) { - targets = this.scheduledTargets.putIfAbsent(schedule, new CopyOnWriteArrayList()); - } - this.scheduledTargets.get(schedule).add(target); - if (targets == null && this.isRunning()) { - this.scheduleDispatcherTask(schedule); - } - } - - public boolean removeTarget(Target target) { - boolean removed = false; - Collection> targetLists = this.scheduledTargets.values(); - for (List targets : targetLists) { - removed = (removed || targets.remove(target)); - } - return removed; - } - - public boolean isRunning() { - return this.running; - } - - public void start() { - synchronized (this.lifecycleMonitor) { - if (this.running) { - return; - } - if (this.scheduler == null) { - throw new ConfigurationException("'scheduler' is required"); - } - if (!this.scheduler.isRunning()) { - this.scheduler.start(); - } - for (Schedule schedule : this.scheduledTargets.keySet()) { - scheduleDispatcherTask(schedule); - } - this.running = true; - } - } - - private void scheduleDispatcherTask(Schedule schedule) { - List targets = this.scheduledTargets.get(schedule); - for (Target target : targets) { - if (target instanceof Lifecycle) { - ((Lifecycle) target).start(); - } - } - this.scheduler.schedule(new DispatcherTask(schedule)); - } - - public void stop() { - if (!this.running) { - return; - } - synchronized (this.lifecycleMonitor) { - for (List targetList : this.scheduledTargets.values()) { - for (Target target : targetList) { - if (target instanceof Lifecycle) { - ((Lifecycle) target).stop(); - } - } - } - this.running = false; - } - } - - public int dispatch() { - MessageDistributor distributor = this.getDistributor(this.defaultSchedule); - return this.doDispatch(distributor); - } - - private int doDispatch(MessageDistributor distributor) { - int messagesProcessed = 0; - Collection> messages = this.retriever.retrieveMessages(); - if (messages == null) { - return 0; - } - for (Message message : messages) { - if (distributor.distribute(message)) { - messagesProcessed++; - } - } - totalMessagesProcessed.addAndGet(messagesProcessed); - return messagesProcessed; - } - - private MessageDistributor getDistributor(Schedule schedule) { - if (schedule == null) { - schedule = this.defaultSchedule; - } - MessageDistributor distributor = new DefaultMessageDistributor(this.channel.getDispatcherPolicy()); - for (Target target : this.scheduledTargets.get(schedule)) { - distributor.addTarget(target); - } - return distributor; - } - - - private class DispatcherTask implements MessagingTask { - - private Schedule schedule; - - - public DispatcherTask(Schedule schedule) { - this.schedule = (schedule != null) ? schedule : defaultSchedule; - } - - public Schedule getSchedule() { - return this.schedule; - } - - public void run() { - doDispatch(getDistributor(this.schedule)); - } - } - -} 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 c077696c79..3b1c70c67d 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 @@ -16,7 +16,7 @@ package org.springframework.integration.dispatcher; -import org.springframework.integration.message.Target; +import org.springframework.integration.message.Message; /** * Strategy interface for dispatching messages. @@ -25,10 +25,6 @@ import org.springframework.integration.message.Target; */ public interface MessageDispatcher { - void addTarget(Target target); - - boolean removeTarget(Target target); - - int dispatch(); + boolean dispatch(Message message); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDistributor.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDistributor.java deleted file mode 100644 index b196987378..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDistributor.java +++ /dev/null @@ -1,35 +0,0 @@ -/* - * 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 org.springframework.integration.message.Message; -import org.springframework.integration.message.Target; - -/** - * Strategy interface for distributing a {@link Message} to one or more {@link Target targets}. - * - * @author Mark Fisher - */ -public interface MessageDistributor { - - void addTarget(Target target); - - boolean removeTarget(Target target); - - boolean distribute(Message message); - -} 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 new file mode 100644 index 0000000000..8b55f3509c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java @@ -0,0 +1,69 @@ +/* + * 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.channel.DispatcherPolicy; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.PollableSource; +import org.springframework.integration.scheduling.MessagingTask; +import org.springframework.integration.scheduling.Schedule; +import org.springframework.util.Assert; + +/** + * A subclass of {@link SimpleDispatcher} that adds message retrieval + * capabilities and may be scheduled to run as a task. + * + * @author Mark Fisher + */ +public class PollingDispatcher extends SimpleDispatcher implements MessagingTask { + + private final SourcePoller poller; + + 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.poller = new SourcePoller(source); + if (!(source instanceof MessageChannel)) { + this.poller.setMaxMessagesPerTask(dispatcherPolicy.getMaxMessagesPerTask()); + this.poller.setTimeout(dispatcherPolicy.getReceiveTimeout()); + } + this.schedule = schedule; + } + + + public Schedule getSchedule() { + return this.schedule; + } + + public void run() { + List> messages = this.poller.poll(); + for (Message message : messages) { + this.dispatch(message); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SchedulingMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SchedulingMessageDispatcher.java deleted file mode 100644 index edf486dd96..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SchedulingMessageDispatcher.java +++ /dev/null @@ -1,35 +0,0 @@ -/* - * 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 org.springframework.context.Lifecycle; -import org.springframework.integration.message.Target; -import org.springframework.integration.scheduling.Schedule; - -/** - * An extension to the {@link MessageDispatcher} strategy for targets that may - * be scheduled. - * - * @author Mark Fisher - */ -public interface SchedulingMessageDispatcher extends MessageDispatcher, Lifecycle { - - void setDefaultSchedule(Schedule defaultSchedule); - - void addTarget(Target target, Schedule schedule); - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java similarity index 81% rename from spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java rename to spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java index c29e40cbfb..953c9342ff 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java @@ -29,40 +29,39 @@ import org.springframework.integration.handler.MessageHandlerNotRunningException import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.integration.message.SubscribableSource; import org.springframework.integration.message.Target; -import org.springframework.util.Assert; /** - * Default implementation of the {@link MessageDistributor} interface. + * Basic implementation of {@link MessageDispatcher}. * * @author Mark Fisher */ -public class DefaultMessageDistributor implements MessageDistributor { +public class SimpleDispatcher implements MessageDispatcher, SubscribableSource { - private final Log logger = LogFactory.getLog(this.getClass()); + protected final Log logger = LogFactory.getLog(this.getClass()); private final List targets = new CopyOnWriteArrayList(); private final DispatcherPolicy dispatcherPolicy; - public DefaultMessageDistributor(DispatcherPolicy dispatcherPolicy) { - Assert.notNull(dispatcherPolicy, "'dispatcherPolicy' must not be null"); + public SimpleDispatcher(DispatcherPolicy dispatcherPolicy) { this.dispatcherPolicy = dispatcherPolicy; } - public void addTarget(Target target) { - this.targets.add(target); + public boolean subscribe(Target target) { + return this.targets.add(target); } - public boolean removeTarget(Target target) { + public boolean unsubscribe(Target target) { return this.targets.remove(target); } - public boolean distribute(Message message) { + public boolean dispatch(Message message) { int attempts = 0; - List targets = new ArrayList(this.targets); + List targetList = new ArrayList(this.targets); while (attempts < this.dispatcherPolicy.getRejectionLimit()) { if (attempts > 0) { if (logger.isDebugEnabled()) { @@ -78,7 +77,7 @@ public class DefaultMessageDistributor implements MessageDistributor { return false; } } - Iterator iter = targets.iterator(); + Iterator iter = targetList.iterator(); if (!iter.hasNext()) { if (logger.isWarnEnabled()) { logger.warn("no active targets"); @@ -94,7 +93,7 @@ public class DefaultMessageDistributor implements MessageDistributor { return true; } if (!sent && logger.isDebugEnabled()) { - logger.debug("endpoint rejected message, continuing with other targets if available"); + logger.debug("target rejected message, continuing with other targets if available"); } iter.remove(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SourcePoller.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SourcePoller.java new file mode 100644 index 0000000000..3f3f9d701e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SourcePoller.java @@ -0,0 +1,90 @@ +/* + * 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.PollableSource; +import org.springframework.util.Assert; + +/** + * Polls a source for {@link Message Messages}. The number of messages + * retrieved per poll is limited by the 'maxMessagesPerTask' + * property, and the timeout for each receive call is determined by the 'timeout' + * 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 SourcePoller { + + private final PollableSource source; + + private volatile int maxMessagesPerTask = 1; + + private volatile long timeout = 0; + + + public SourcePoller(PollableSource source) { + Assert.notNull(source, "source must not be null"); + if (source instanceof MessageChannel) { + DispatcherPolicy dispatcherPolicy = ((MessageChannel) source).getDispatcherPolicy(); + this.setMaxMessagesPerTask(dispatcherPolicy.getMaxMessagesPerTask()); + this.setTimeout(dispatcherPolicy.getReceiveTimeout()); + } + this.source = source; + } + + + public void setMaxMessagesPerTask(int maxMessagesPerTask) { + Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be a positive value"); + this.maxMessagesPerTask = maxMessagesPerTask; + } + + public void setTimeout(long timeout) { + this.timeout = timeout; + } + + public List> poll() { + List> messages = new LinkedList>(); + while (messages.size() < this.maxMessagesPerTask) { + Message message = null; + if (this.source instanceof BlockingSource && this.timeout >= 0) { + message = ((BlockingSource) this.source).receive(this.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/SynchronousChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java index c2eda45b01..a99067891a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java @@ -27,6 +27,7 @@ import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.SubscribableSource; import org.springframework.integration.message.Target; import org.springframework.integration.message.selector.MessageSelector; @@ -44,14 +45,14 @@ import org.springframework.integration.message.selector.MessageSelector; * @author Dave Syer * @author Mark Fisher */ -public class SynchronousChannel extends AbstractMessageChannel { +public class SynchronousChannel extends AbstractMessageChannel implements SubscribableSource { private static final ThreadLocalMessageHolder messageHolder = new ThreadLocalMessageHolder(); private volatile PollableSource source; - private final MessageDistributor distributor; + private final SimpleDispatcher dispatcher; private final AtomicInteger handlerCount = new AtomicInteger(); @@ -63,7 +64,7 @@ public class SynchronousChannel extends AbstractMessageChannel { public SynchronousChannel(PollableSource source) { super(defaultDispatcherPolicy()); this.source = source; - this.distributor = new DefaultMessageDistributor(this.getDispatcherPolicy()); + this.dispatcher = new SimpleDispatcher(this.getDispatcherPolicy()); } @@ -71,17 +72,20 @@ public class SynchronousChannel extends AbstractMessageChannel { this.source = source; } - public void addTarget(Target target) { - this.distributor.addTarget(target); - this.handlerCount.incrementAndGet(); + public boolean subscribe(Target target) { + boolean added = this.dispatcher.subscribe(target); + if (added) { + this.handlerCount.incrementAndGet(); + } + return added; } - public boolean removeTarget(Target target) { - if (this.distributor.removeTarget(target)) { + public boolean unsubscribe(Target target) { + boolean removed = this.dispatcher.unsubscribe(target); + if (removed) { this.handlerCount.decrementAndGet(); - return true; } - return false; + return removed; } @@ -102,7 +106,7 @@ public class SynchronousChannel extends AbstractMessageChannel { return false; } if (this.handlerCount.get() > 0) { - return this.distributor.distribute(message); + return this.dispatcher.dispatch(message); } else if (this.source == null) { return messageHolder.get().add(message); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageRetriever.java b/spring-integration-core/src/main/java/org/springframework/integration/message/SubscribableSource.java similarity index 63% rename from spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageRetriever.java rename to spring-integration-core/src/main/java/org/springframework/integration/message/SubscribableSource.java index 65761f32a2..dee66a53cc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageRetriever.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/SubscribableSource.java @@ -14,19 +14,23 @@ * limitations under the License. */ -package org.springframework.integration.dispatcher; - -import java.util.Collection; - -import org.springframework.integration.message.Message; +package org.springframework.integration.message; /** - * Strategy interface for retrieving messages. + * Interface for any message source that accepts subscribers. * * @author Mark Fisher */ -public interface MessageRetriever { +public interface SubscribableSource extends Source { - Collection> retrieveMessages(); + /** + * Register a {@link Target} as a subscriber to this source. + */ + boolean subscribe(Target target); + + /** + * Remove a {@link Target} from the subscribers of this source. + */ + boolean unsubscribe(Target target); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java index db06b79583..be6bec2e07 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java @@ -20,7 +20,6 @@ import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; @@ -50,10 +49,6 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler private final Object lifecycleMonitor = new Object(); - public SimpleMessagingTaskScheduler(int corePoolSize) { - this(new ScheduledThreadPoolExecutor(corePoolSize)); - } - public SimpleMessagingTaskScheduler(ScheduledExecutorService executor) { Assert.notNull(executor, "'executor' must not be null"); this.executor = executor; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java similarity index 85% rename from spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java index 7cad05f5e6..a6c2ea6ebd 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dispatcher; +package org.springframework.integration.bus; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; +import org.springframework.integration.bus.SubscriptionManager; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; @@ -47,7 +48,7 @@ import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; /** * @author Mark Fisher */ -public class DefaultMessageDispatcherTests { +public class SubscriptionManagerTests { private SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(new ScheduledThreadPoolExecutor(10)); @@ -61,10 +62,10 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(createEndpoint(handler1, true)); - dispatcher.addTarget(createEndpoint(handler2, true)); - dispatcher.start(); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); + manager.addTarget(createEndpoint(handler1, true)); + manager.addTarget(createEndpoint(handler2, true)); + manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("exactly one handler should have received message", 1, counter1.get() + counter2.get()); @@ -79,10 +80,10 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(createEndpoint(handler1, true)); - dispatcher.addTarget(createEndpoint(handler2, true)); - dispatcher.start(); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); + manager.addTarget(createEndpoint(handler1, true)); + manager.addTarget(createEndpoint(handler2, true)); + manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("both handlers should have received message", 2, counter1.get() + counter2.get()); @@ -98,12 +99,12 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); SimpleChannel channel = new SimpleChannel(); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); MessageEndpoint inactiveEndpoint = createEndpoint(handler1, true); - dispatcher.addTarget(inactiveEndpoint); - dispatcher.addTarget(createEndpoint(handler2, true)); - dispatcher.addTarget(createEndpoint(handler3, true)); - dispatcher.start(); + manager.addTarget(inactiveEndpoint); + manager.addTarget(createEndpoint(handler2, true)); + manager.addTarget(createEndpoint(handler3, true)); + manager.start(); inactiveEndpoint.stop(); channel.send(new StringMessage(1, "test")); latch.await(2000, TimeUnit.MILLISECONDS); @@ -122,12 +123,12 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); MessageEndpoint inactiveEndpoint = createEndpoint(handler2, true); - dispatcher.addTarget(createEndpoint(handler1, true)); - dispatcher.addTarget(inactiveEndpoint); - dispatcher.addTarget(createEndpoint(handler3, true)); - dispatcher.start(); + manager.addTarget(createEndpoint(handler1, true)); + manager.addTarget(inactiveEndpoint); + manager.addTarget(createEndpoint(handler3, true)); + manager.start(); inactiveEndpoint.stop(); channel.send(new StringMessage(1, "test")); latch.await(2000, TimeUnit.MILLISECONDS); @@ -140,8 +141,8 @@ public class DefaultMessageDispatcherTests { public void testDispatcherWithNoExecutorsDoesNotFail() { SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.start(); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); + manager.start(); } @Test @@ -155,17 +156,17 @@ public class DefaultMessageDispatcherTests { channel.getDispatcherPolicy().setRejectionLimit(2); channel.getDispatcherPolicy().setRetryInterval(3); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(createEndpoint(handler1, true)); - dispatcher.addTarget(new Target() { + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); + manager.addTarget(createEndpoint(handler1, true)); + manager.addTarget(new Target() { public boolean send(Message message) { throw new MessageHandlerRejectedExecutionException(message); } }); - dispatcher.addTarget(createEndpoint(handler3, true)); + manager.addTarget(createEndpoint(handler3, true)); SimpleChannel errorChannel = new SimpleChannel(); scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); - dispatcher.start(); + manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); Message errorMessage = errorChannel.receive(1000); @@ -186,16 +187,16 @@ public class DefaultMessageDispatcherTests { channel.getDispatcherPolicy().setRetryInterval(3); channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(createEndpoint(handler1, false)); - dispatcher.addTarget(createEndpoint(new MessageHandler() { + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); + manager.addTarget(createEndpoint(handler1, false)); + manager.addTarget(createEndpoint(new MessageHandler() { public Message handle(Message message) { latch.countDown(); throw new MessageHandlerRejectedExecutionException(message); } }, false)); - dispatcher.addTarget(createEndpoint(handler2, false)); - dispatcher.start(); + manager.addTarget(createEndpoint(handler2, false)); + manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("both non-rejecting handlers should have received message", 2, counter1.get() + counter2.get()); @@ -208,14 +209,14 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.rejectingCountDownHandler(latch); SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(false)); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); channel.getDispatcherPolicy().setRejectionLimit(2); channel.getDispatcherPolicy().setRetryInterval(3); - dispatcher.addTarget(createEndpoint(handler1, false)); - dispatcher.addTarget(createEndpoint(handler2, false)); + manager.addTarget(createEndpoint(handler1, false)); + manager.addTarget(createEndpoint(handler2, false)); SimpleChannel errorChannel = new SimpleChannel(); scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); - dispatcher.start(); + manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); Message errorMessage = errorChannel.receive(500); @@ -236,22 +237,22 @@ public class DefaultMessageDispatcherTests { channel.getDispatcherPolicy().setRetryInterval(3); channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(createEndpoint(new MessageHandler() { + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); + manager.addTarget(createEndpoint(new MessageHandler() { public Message handle(Message message) { rejectedCounter1.incrementAndGet(); latch.countDown(); throw new MessageHandlerRejectedExecutionException(message); } }, false)); - dispatcher.addTarget(createEndpoint(new MessageHandler() { + manager.addTarget(createEndpoint(new MessageHandler() { public Message handle(Message message) { rejectedCounter2.incrementAndGet(); latch.countDown(); throw new MessageHandlerRejectedExecutionException(message); } }, false)); - dispatcher.start(); + manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("latch should have counted down within allotted time", 0, latch.getCount()); @@ -276,15 +277,15 @@ public class DefaultMessageDispatcherTests { dispatcherPolicy.setShouldFailOnRejectionLimit(false); SimpleChannel channel = new SimpleChannel(25, dispatcherPolicy); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(createEndpoint(new MessageHandler() { + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); + manager.addTarget(createEndpoint(new MessageHandler() { public Message handle(Message message) { rejectedCounter1.incrementAndGet(); latch.countDown(); throw new MessageHandlerRejectedExecutionException(message); } }, false)); - dispatcher.addTarget(createEndpoint(new MessageHandler() { + manager.addTarget(createEndpoint(new MessageHandler() { public Message handle(Message message) { if (rejectedCounter2.get() == 1) { return handler2.handle(message); @@ -294,14 +295,14 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(message); } }, false)); - dispatcher.addTarget(createEndpoint(new MessageHandler() { + manager.addTarget(createEndpoint(new MessageHandler() { public Message handle(Message message) { rejectedCounter3.incrementAndGet(); latch.countDown(); throw new MessageHandlerRejectedExecutionException(message); } }, false)); - dispatcher.start(); + manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("handler1 should not have received message", 0, counter1.get()); @@ -327,8 +328,8 @@ public class DefaultMessageDispatcherTests { dispatcherPolicy.setShouldFailOnRejectionLimit(false); SimpleChannel channel = new SimpleChannel(25, dispatcherPolicy); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); - dispatcher.addTarget(createEndpoint(new MessageHandler() { + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); + manager.addTarget(createEndpoint(new MessageHandler() { public Message handle(Message message) { if (rejectedCounter1.get() == 2) { return handler1.handle(message); @@ -338,7 +339,7 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(message); } }, false)); - dispatcher.addTarget(createEndpoint(new MessageHandler() { + manager.addTarget(createEndpoint(new MessageHandler() { public Message handle(Message message) { if (rejectedCounter2.get() == 4) { return handler2.handle(message); @@ -348,7 +349,7 @@ public class DefaultMessageDispatcherTests { throw new MessageHandlerRejectedExecutionException(message); } }, false)); - dispatcher.start(); + manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("handler1 should have received one message", 1, counter1.get()); @@ -366,14 +367,14 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class)); endpoint2.addMessageSelector(new PayloadTypeSelector(String.class)); - dispatcher.addTarget(endpoint1); - dispatcher.addTarget(endpoint2); - dispatcher.start(); + manager.addTarget(endpoint1); + manager.addTarget(endpoint2); + manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("handler1 should not have accepted the message", 0, counter1.get()); @@ -392,7 +393,7 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, handlerLatch); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); final HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); final HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class) { @@ -411,9 +412,9 @@ public class DefaultMessageDispatcherTests { return super.accept(message); } }); - dispatcher.addTarget(endpoint1); - dispatcher.addTarget(endpoint2); - dispatcher.start(); + manager.addTarget(endpoint1); + manager.addTarget(endpoint2); + manager.start(); selectorLatch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, selectorLatch.getCount()); assertEquals("handler1 should not have accepted the message", 0, counter1.get()); @@ -432,16 +433,16 @@ public class DefaultMessageDispatcherTests { MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); endpoint1.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); endpoint2.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class)); endpoint2.addMessageSelector(new PayloadTypeSelector(String.class)); - dispatcher.addTarget(endpoint1); - dispatcher.addTarget(endpoint2); - dispatcher.start(); + manager.addTarget(endpoint1); + manager.addTarget(endpoint2); + manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java index 2f3e2e03f7..75b298d6f5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java @@ -21,6 +21,7 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -29,9 +30,9 @@ import org.junit.Test; import org.springframework.beans.FatalBeanException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.bus.SubscriptionManager; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.dispatcher.DefaultMessageDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; @@ -68,15 +69,15 @@ public class ChannelParserTests { "channelParserTests.xml", this.getClass()); MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelByDefault"); channel.send(new StringMessage("test")); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(new ScheduledThreadPoolExecutor(1)); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); AtomicInteger counter = new AtomicInteger(); CountDownLatch latch = new CountDownLatch(1); TestTarget target1 = new TestTarget(counter, latch); TestTarget target2 = new TestTarget(counter, latch); - dispatcher.addTarget(target1); - dispatcher.addTarget(target2); - dispatcher.start(); + manager.addTarget(target1); + manager.addTarget(target2); + manager.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); assertEquals(1, counter.get()); @@ -88,15 +89,15 @@ public class ChannelParserTests { "channelParserTests.xml", this.getClass()); MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelExplicitlyConfigured"); channel.send(new StringMessage("test")); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(new ScheduledThreadPoolExecutor(1)); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); AtomicInteger counter = new AtomicInteger(); CountDownLatch latch = new CountDownLatch(1); TestTarget target1 = new TestTarget(counter, latch); TestTarget target2 = new TestTarget(counter, latch); - dispatcher.addTarget(target1); - dispatcher.addTarget(target2); - dispatcher.start(); + manager.addTarget(target1); + manager.addTarget(target2); + manager.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); assertEquals(1, counter.get()); @@ -108,15 +109,15 @@ public class ChannelParserTests { "channelParserTests.xml", this.getClass()); MessageChannel channel = (MessageChannel) context.getBean("publishSubscribeChannel"); channel.send(new StringMessage("test")); - SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1); - DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(new ScheduledThreadPoolExecutor(1)); + SubscriptionManager manager = new SubscriptionManager(channel, scheduler); AtomicInteger counter = new AtomicInteger(); CountDownLatch latch = new CountDownLatch(2); TestTarget target1 = new TestTarget(counter, latch); TestTarget target2 = new TestTarget(counter, latch); - dispatcher.addTarget(target1); - dispatcher.addTarget(target2); - dispatcher.start(); + manager.addTarget(target1); + manager.addTarget(target2); + manager.start(); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); assertEquals(2, counter.get()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDistributorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java similarity index 72% rename from spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDistributorTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java index 8caa9d97ef..41ec4b6142 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDistributorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java @@ -34,27 +34,27 @@ import org.springframework.integration.message.Target; /** * @author Mark Fisher */ -public class DefaultMessageDistributorTests { +public class SimpleDispatcherTests { @Test public void testSingleMessage() throws InterruptedException { - MessageDistributor distributor = new DefaultMessageDistributor(new DispatcherPolicy()); + SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy()); final CountDownLatch latch = new CountDownLatch(1); - distributor.addTarget(createEndpoint(TestHandlers.countDownHandler(latch))); - distributor.distribute(new StringMessage("test")); + dispatcher.subscribe(createEndpoint(TestHandlers.countDownHandler(latch))); + dispatcher.dispatch(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); } @Test public void testPointToPoint() throws InterruptedException { - MessageDistributor distributor = new DefaultMessageDistributor(new DispatcherPolicy(false)); + SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy(false)); final CountDownLatch latch = new CountDownLatch(1); final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); - distributor.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch))); - distributor.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch))); - distributor.distribute(new StringMessage("test")); + dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch))); + dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch))); + dispatcher.dispatch(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); assertEquals("only 1 handler should have received the message", 1, counter1.get() + counter2.get()); @@ -62,13 +62,13 @@ public class DefaultMessageDistributorTests { @Test public void testPublishSubscribe() throws InterruptedException { - MessageDistributor distributor = new DefaultMessageDistributor(new DispatcherPolicy(true)); + SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy(true)); final CountDownLatch latch = new CountDownLatch(2); final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); - distributor.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch))); - distributor.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch))); - distributor.distribute(new StringMessage("test")); + dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch))); + dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch))); + dispatcher.dispatch(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); assertEquals(1, counter1.get()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java similarity index 79% rename from spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java index d0257d71ae..31bce8e325 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java @@ -33,22 +33,22 @@ import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher */ -public class ChannelPollingMessageRetrieverTests { +public class SourcePollerTests { @Test public void testSingleMessagePerRetrieval() { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); MessageChannel channel = new SimpleChannel(5, dispatcherPolicy); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - Collection> results = retriever.retrieveMessages(); + SourcePoller poller = new SourcePoller(channel); + Collection> results = poller.poll(); assertTrue(results.isEmpty()); channel.send(new StringMessage("test1"), 0); channel.send(new StringMessage("test2"), 0); - results = retriever.retrieveMessages(); + results = poller.poll(); assertEquals(1, results.size()); assertEquals("test1", results.iterator().next().getPayload()); - results = retriever.retrieveMessages(); + results = poller.poll(); assertEquals(1, results.size()); assertEquals("test2", results.iterator().next().getPayload()); } @@ -59,18 +59,18 @@ public class ChannelPollingMessageRetrieverTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); MessageChannel channel = new SimpleChannel(5, dispatcherPolicy); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - Collection> results = retriever.retrieveMessages(); + SourcePoller poller = new SourcePoller(channel); + Collection> results = poller.poll(); assertTrue(results.isEmpty()); channel.send(new StringMessage("test1"), 0); channel.send(new StringMessage("test2"), 0); channel.send(new StringMessage("test3"), 0); - results = retriever.retrieveMessages(); + results = poller.poll(); assertEquals(2, results.size()); Iterator> iter = results.iterator(); assertEquals("test1", iter.next().getPayload()); assertEquals("test2", iter.next().getPayload()); - results = retriever.retrieveMessages(); + results = poller.poll(); assertEquals(1, results.size()); assertEquals("test3", results.iterator().next().getPayload()); } @@ -81,17 +81,17 @@ public class ChannelPollingMessageRetrieverTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(1); MessageChannel channel = new SimpleChannel(5, dispatcherPolicy); - ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); - Collection> results = retriever.retrieveMessages(); + SourcePoller poller = new SourcePoller(channel); + Collection> results = poller.poll(); assertTrue(results.isEmpty()); channel.send(new StringMessage("test1"), 0); channel.send(new StringMessage("test2"), 0); channel.send(new StringMessage("test3"), 0); - results = retriever.retrieveMessages(); + results = poller.poll(); assertEquals(1, results.size()); assertEquals("test1", results.iterator().next().getPayload()); - channel.getDispatcherPolicy().setMaxMessagesPerTask(5); - results = retriever.retrieveMessages(); + poller.setMaxMessagesPerTask(5); + results = poller.poll(); assertEquals(2, results.size()); Iterator> iter = results.iterator(); assertEquals("test2", iter.next().getPayload()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java index 26b7b87796..7e530ea095 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java @@ -44,7 +44,7 @@ public class SynchronousChannelTests { @Test public void testSend() { SynchronousChannel channel = new SynchronousChannel(); - channel.addTarget(new ThreadNameSettingTestTarget()); + channel.subscribe(new ThreadNameSettingTestTarget()); StringMessage message = new StringMessage("test"); assertTrue(channel.send(message)); String handlerThreadName = message.getHeader().getProperty(HANDLER_THREAD); @@ -82,7 +82,7 @@ public class SynchronousChannelTests { public void testSendInSeparateThread() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); final SynchronousChannel channel = new SynchronousChannel(); - channel.addTarget(new ThreadNameSettingTestTarget(latch)); + channel.subscribe(new ThreadNameSettingTestTarget(latch)); final StringMessage message = new StringMessage("test"); new Thread(new Runnable() { public void run() {