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

Simplified Message dispatching. Removed MessageRetriever and MessageDistributor (and the "default" implementations). Added SimpleDispatcher and PollingDispatcher. Added the SubscribableSource interface and a SubscriptionManager to simplify the coordination of dispatchers within the MessageBus. These changes also lay the groundwork for further refactoring of the various Source and SourceAdapter implementations.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-18 21:26:31 +00:00
parent 72ed283296
commit d30f3230b0
20 changed files with 541 additions and 597 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java
View File

@@ -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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(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]);

54
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java
View File

@@ -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>(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<TestObject>(testObject1), 0);
channel.send(new GenericMessage<TestObject>(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<TestObject>(testObject1), 0);
channel.send(new GenericMessage<TestObject>(testObject2), 0);
dispatcher.dispatch();
dispatcher.run();
String newLine = System.getProperty("line.separator");
assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
}

46
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -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<MessageChannel, SchedulingMessageDispatcher> dispatchers = new ConcurrentHashMap<MessageChannel, SchedulingMessageDispatcher>();
private final Map<MessageChannel, SubscriptionManager> subscriptionManagers = new ConcurrentHashMap<MessageChannel, SubscriptionManager>();
private final List<Lifecycle> lifecycleSourceAdapters = new CopyOnWriteArrayList<Lifecycle>();
@@ -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<SchedulingMessageDispatcher> dispatchers = this.dispatchers.values();
Collection<SubscriptionManager> 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 + "'");
}
}
}

162
spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java Normal file
View File

@@ -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<Schedule, PollingDispatcher> dispatchers = new ConcurrentHashMap<Schedule, PollingDispatcher>();
private final List<Lifecycle> lifecycleTargets = new CopyOnWriteArrayList<Lifecycle>();
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<PollingDispatcher> 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;
}
}
}

69
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java
View File

@@ -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 '<em>maxMessagesPerTask</em>'
* property of the channel's dispatcher policy, and the timeout for each receive
* call is determined by the dispatcher policy's '<em>receiveTimeout</em>'
* 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<Message<?>> retrieveMessages() {
List<Message<?>> messages = new LinkedList<Message<?>>();
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;
}
}

221
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java
View File

@@ -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<Schedule, List<Target>> scheduledTargets = new ConcurrentHashMap<Schedule, List<Target>>();
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<Target> targets = this.scheduledTargets.get(schedule);
if (targets == null) {
targets = this.scheduledTargets.putIfAbsent(schedule, new CopyOnWriteArrayList<Target>());
}
this.scheduledTargets.get(schedule).add(target);
if (targets == null && this.isRunning()) {
this.scheduleDispatcherTask(schedule);
}
}
public boolean removeTarget(Target target) {
boolean removed = false;
Collection<List<Target>> targetLists = this.scheduledTargets.values();
for (List<Target> 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<Target> 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<Target> 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<Message<?>> 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));
}
}
}

8
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java
View File

@@ -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);
}

35
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDistributor.java
View File

@@ -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);
}

69
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java Normal file
View File

@@ -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<Message<?>> messages = this.poller.poll();
for (Message<?> message : messages) {
this.dispatch(message);
}
}
}

35
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SchedulingMessageDispatcher.java
View File

@@ -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);
}

25
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java → spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java
View File

@@ -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<Target> targets = new CopyOnWriteArrayList<Target>();
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<Target> targets = new ArrayList<Target>(this.targets);
List<Target> targetList = new ArrayList<Target>(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<Target> iter = targets.iterator();
Iterator<Target> 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();
}

90
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SourcePoller.java Normal file
View File

@@ -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 '<em>maxMessagesPerTask</em>'
* property, and the timeout for each receive call is determined by the '<em>timeout</em>'
* 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<Message<?>> poll() {
List<Message<?>> messages = new LinkedList<Message<?>>();
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;
}
}

26
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
View File

@@ -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);

20
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageRetriever.java → spring-integration-core/src/main/java/org/springframework/integration/message/SubscribableSource.java
View File

@@ -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<Message<?>> 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);
}

5
spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java
View File

@@ -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;

123
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java → spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java
View File

@@ -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());

33
spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java
View File

@@ -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());

24
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDistributorTests.java → spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java
View File

@@ -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());

28
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java → spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java
View File

@@ -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<Message<?>> results = retriever.retrieveMessages();
SourcePoller poller = new SourcePoller(channel);
Collection<Message<?>> 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<Message<?>> results = retriever.retrieveMessages();
SourcePoller poller = new SourcePoller(channel);
Collection<Message<?>> 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<Message<?>> 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<Message<?>> results = retriever.retrieveMessages();
SourcePoller poller = new SourcePoller(channel);
Collection<Message<?>> 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<Message<?>> iter = results.iterator();
assertEquals("test2", iter.next().getPayload());

4
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java
View File

@@ -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() {