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

Major refactoring of endpoint subscription, scheduling, and activation. This includes significant changes to the dispatcher implementations and the addition of endpoint "triggers" to drive polling and dispatching of messages. Also introduces an explicit PublishSubscribeChannel implementation and thereby removes support for the "publish-subscribe" attribute from channel elements in the namespace.

Browse Source
This commit is contained in:
Mark Fisher
2008-07-04 18:48:33 +00:00
parent 218da1aacf
commit 0aa00b969b
34 changed files with 782 additions and 1222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

186
org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -16,12 +16,11 @@
package org.springframework.integration.bus;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import org.apache.commons.logging.Log;
@@ -29,7 +28,6 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
@@ -45,7 +43,6 @@ import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.DefaultChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.factory.ChannelFactory;
import org.springframework.integration.channel.factory.QueueChannelFactory;
import org.springframework.integration.endpoint.DefaultEndpointRegistry;
@@ -53,16 +50,13 @@ import org.springframework.integration.endpoint.EndpointRegistry;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.MessagingGateway;
import org.springframework.integration.endpoint.MessageProducingEndpoint;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.endpoint.MessageConsumingEndpoint;
import org.springframework.integration.endpoint.TargetEndpoint;
import org.springframework.integration.endpoint.EndpointTrigger;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.CommandMessage;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollCommand;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
import org.springframework.integration.scheduling.SchedulableTask;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.SimpleTaskScheduler;
@@ -91,12 +85,14 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
private final EndpointRegistry endpointRegistry = new DefaultEndpointRegistry();
private final Map<MessageChannel, SubscriptionManager> subscriptionManagers = new ConcurrentHashMap<MessageChannel, SubscriptionManager>();
private final Set<EndpointTrigger> endpointTriggers = new CopyOnWriteArraySet<EndpointTrigger>();
private final List<Lifecycle> lifecycleEndpoints = new CopyOnWriteArrayList<Lifecycle>();
private final MessageBusInterceptorsList interceptors = new MessageBusInterceptorsList();
private volatile Schedule defaultPollerSchedule = new PollingSchedule(0);
private volatile TaskScheduler taskScheduler;
private volatile boolean configureAsyncEventMulticaster = false;
@@ -232,8 +228,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
this.initialize();
}
channel.setName(name);
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 + "'");
@@ -241,14 +235,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
}
public MessageChannel unregisterChannel(String name) {
MessageChannel removedChannel = this.channelRegistry.unregisterChannel(name);
if (removedChannel != null) {
SubscriptionManager manager = this.subscriptionManagers.remove(removedChannel);
if (manager != null && manager.isRunning()) {
manager.stop();
}
}
return removedChannel;
return this.channelRegistry.unregisterChannel(name);
}
public void registerHandler(String name, MessageHandler handler, Subscription subscription) {
@@ -276,12 +263,9 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
if (endpoint instanceof ChannelRegistryAware) {
((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry);
}
if (endpoint instanceof SourceEndpoint) {
this.registerSourceEndpoint(name, (SourceEndpoint) endpoint);
}
this.endpointRegistry.registerEndpoint(name, endpoint);
if (this.isRunning()) {
activateEndpoint(endpoint);
this.activateEndpoint(endpoint);
}
if (logger.isInfoEnabled()) {
logger.info("registered endpoint '" + name + "'");
@@ -293,17 +277,8 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
if (endpoint == null) {
return null;
}
if (endpoint instanceof TargetEndpoint) {
Collection<SubscriptionManager> managers = this.subscriptionManagers.values();
boolean removed = false;
for (SubscriptionManager manager : managers) {
removed = (removed || manager.removeTarget((TargetEndpoint) endpoint));
}
if (removed) {
return endpoint;
}
}
return null;
this.deactivateEndpoint(endpoint);
return endpoint;
}
public MessageEndpoint lookupEndpoint(String endpointName) {
@@ -325,84 +300,63 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
}
private void activateEndpoint(MessageEndpoint endpoint) {
if (endpoint instanceof MessageProducingEndpoint) {
String channelName = ((MessageProducingEndpoint) endpoint).getOutputChannelName();
if (channelName != null && this.lookupChannel(channelName) == null) {
if (!this.autoCreateChannels) {
throw new ConfigurationException("Unknown channel '" + channelName
+ "' configured as output channel for endpoint '" + endpoint
+ "'. Consider enabling the 'autoCreateChannels' option for the message bus.");
}
this.registerChannel(channelName, new QueueChannel());
}
Assert.notNull(endpoint, "'endpoint' must not be null");
if (endpoint.getOutputChannel() == null) {
this.lookupOrCreateChannel(endpoint.getOutputChannelName());
}
if (endpoint instanceof InitializingBean) {
try {
((InitializingBean) endpoint).afterPropertiesSet();
}
catch (Exception e) {
throw new ConfigurationException("failed to initialize endpoint", e);
}
try {
endpoint.afterPropertiesSet();
}
if (endpoint instanceof MessageConsumingEndpoint && endpoint instanceof MessageTarget) {
this.activateSubscriber((MessageConsumingEndpoint) endpoint);
catch (Exception e) {
throw new ConfigurationException("failed to initialize endpoint", e);
}
}
private void activateSubscriber(MessageConsumingEndpoint subscriber) {
Subscription subscription = subscriber.getSubscription();
if (subscription == null) {
throw new ConfigurationException("Unable to register endpoint '" + subscriber
+ "'. No subscription information is available.");
}
MessageChannel channel = subscription.getChannel();
if (channel == null) {
String channelName = subscription.getChannelName();
if (channelName == null) {
throw new ConfigurationException("endpoint '" + subscriber
+ "' must provide either 'channel' or 'channelName' in its subscription metadata");
}
channel = this.lookupChannel(channelName);
Schedule schedule = null;
Subscription subscription = endpoint.getSubscription();
if (subscription != null) {
schedule = subscription.getSchedule();
MessageChannel channel = subscription.getChannel();
if (channel == null) {
if (!this.autoCreateChannels) {
throw new ConfigurationException("Cannot activate subscription, unknown channel '" + channelName
+ "'. Consider enabling the 'autoCreateChannels' option for the message bus.");
channel = this.lookupOrCreateChannel(subscription.getChannelName());
}
if (channel != null && channel instanceof Subscribable) {
((Subscribable) channel).subscribe(endpoint);
if (logger.isInfoEnabled()) {
logger.info("activated subscription to channel '"
+ channel.getName() + "' for endpoint '" + endpoint + "'");
}
if (this.logger.isInfoEnabled()) {
logger.info("auto-creating channel '" + channelName + "'");
}
channel = channelFactory.getChannel(null, null);
this.registerChannel(channelName, channel);
return;
}
}
this.activateSubscription(channel, (MessageTarget) subscriber, subscription.getSchedule());
if (logger.isInfoEnabled()) {
logger.info("activated subscription to channel '" + channel.getName()
+ "' for endpoint '" + subscriber + "' of type '" + subscriber.getClass() + "'");
if (schedule == null) {
schedule = endpoint.getSchedule();
}
EndpointTrigger trigger = endpoint.getTrigger();
if (trigger == null) {
trigger = new EndpointTrigger(schedule != null ? schedule : this.defaultPollerSchedule);
}
trigger.addTarget(endpoint);
if (this.endpointTriggers.add(trigger)) {
this.taskScheduler.schedule(trigger);
}
}
private void registerSourceEndpoint(String name, final SourceEndpoint endpoint) {
if (!this.initialized) {
this.initialize();
private MessageChannel lookupOrCreateChannel(String channelName) {
if (channelName == null) {
return null;
}
final Schedule schedule = endpoint.getSchedule();
if (schedule != null) {
this.taskScheduler.schedule(new SchedulableTask() {
public Schedule getSchedule() {
return schedule;
}
public void run() {
endpoint.invoke(new CommandMessage(new PollCommand()));
}
});
}
if (endpoint instanceof Lifecycle) {
this.lifecycleEndpoints.add((Lifecycle) endpoint);
if (this.isRunning()) {
((Lifecycle) endpoint).start();
MessageChannel channel = this.lookupChannel(channelName);
if (channel == null) {
if (!this.autoCreateChannels) {
throw new ConfigurationException("Cannot activate endpoint, unknown channel '" + channelName
+ "'. Consider enabling the 'autoCreateChannels' option for the message bus.");
}
if (this.logger.isInfoEnabled()) {
logger.info("auto-creating channel '" + channelName + "'");
}
channel = channelFactory.getChannel(null, null);
this.registerChannel(channelName, channel);
}
return channel;
}
private void registerGateway(String name, MessagingGateway gateway) {
@@ -417,18 +371,16 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
}
}
private void activateSubscription(MessageChannel channel, MessageTarget target, Schedule schedule) {
SubscriptionManager manager = this.subscriptionManagers.get(channel);
if (manager == null) {
if (logger.isWarnEnabled()) {
logger.warn("no subscription manager available for channel '" + channel
+ "', be sure to register the channel");
public void deactivateEndpoint(MessageEndpoint endpoint) {
Assert.notNull(endpoint, "'endpoint' must not be null");
for (EndpointTrigger trigger : this.endpointTriggers) {
boolean removed = trigger.removeTarget(endpoint);
if (removed && this.logger.isInfoEnabled()) {
logger.info("removed endpoint '" + endpoint + "' from dispatcher");
}
return;
}
manager.addTarget(target, schedule);
if (this.isRunning() && !manager.isRunning()) {
manager.start();
if (endpoint instanceof Lifecycle) {
((Lifecycle) endpoint).stop();
}
}
@@ -451,12 +403,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
this.activateEndpoints();
this.taskScheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
this.taskScheduler.start();
for (SubscriptionManager manager : this.subscriptionManagers.values()) {
manager.start();
if (logger.isInfoEnabled()) {
logger.info("started subscription manager '" + manager + "'");
}
}
for (Lifecycle endpoint : this.lifecycleEndpoints) {
endpoint.start();
if (logger.isInfoEnabled()) {
@@ -486,12 +432,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
logger.info("stopped endpoint '" + endpoint + "'");
}
}
for (SubscriptionManager manager : this.subscriptionManagers.values()) {
manager.stop();
if (logger.isInfoEnabled()) {
logger.info("stopped subscription manager '" + manager + "'");
}
}
}
this.interceptors.postStop();
if (logger.isInfoEnabled()) {

172
org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java
View File

@@ -1,172 +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.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.DirectChannel;
import org.springframework.integration.dispatcher.PollingDispatcherTask;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
* Manages subscriptions for {@link MessageTarget 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 TaskScheduler scheduler;
private volatile Schedule defaultSchedule = new PollingSchedule(5);
private final ConcurrentMap<Schedule, PollingDispatcherTask> dispatcherTasks = new ConcurrentHashMap<Schedule, PollingDispatcherTask>();
private final List<Lifecycle> lifecycleTargets = new CopyOnWriteArrayList<Lifecycle>();
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
public SubscriptionManager(MessageChannel channel, TaskScheduler 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(MessageTarget target) {
this.addTarget(target, null);
}
public void addTarget(MessageTarget target, Schedule schedule) {
Assert.notNull(target, "'target' must not be null");
if (schedule == null) {
schedule = this.defaultSchedule;
}
else if (this.channel instanceof DirectChannel) {
if (logger.isInfoEnabled()) {
logger.info("Subscribing to a DirectChannel. The provided schedule will be ignored.");
}
}
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();
}
}
if (this.channel instanceof DirectChannel) {
((DirectChannel) this.channel).subscribe(target);
return;
}
PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule);
if (dispatcherTask == null) {
dispatcherTask = this.dispatcherTasks.putIfAbsent(schedule, new PollingDispatcherTask(this.channel, schedule));
}
this.dispatcherTasks.get(schedule).subscribe(target);
if (dispatcherTask == null && this.isRunning()) {
this.scheduleDispatcherTask(schedule);
}
}
public boolean removeTarget(MessageTarget target) {
boolean removed = false;
Collection<PollingDispatcherTask> dispatcherTaskValues = this.dispatcherTasks.values();
for (PollingDispatcherTask dispatcherTask : dispatcherTaskValues) {
removed = (removed || dispatcherTask.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.dispatcherTasks.keySet()) {
this.scheduleDispatcherTask(schedule);
}
this.running = true;
}
}
private void scheduleDispatcherTask(Schedule schedule) {
PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule);
if (dispatcherTask != null) {
this.scheduler.schedule(dispatcherTask);
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (!this.running) {
return;
}
for (Lifecycle target : lifecycleTargets) {
target.stop();
}
this.running = false;
}
}
}

78
org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java Normal file
View File

@@ -0,0 +1,78 @@
/*
* 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.channel;
import java.util.List;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.message.selector.MessageSelector;
/**
* @author Mark Fisher
*/
public class PublishSubscribeChannel extends AbstractMessageChannel implements Subscribable {
private final BroadcastingDispatcher dispatcher = new BroadcastingDispatcher();
public PublishSubscribeChannel() {
super(new DispatcherPolicy(true));
}
/**
* Create a PublishSubscribeChannel that will use a {@link TaskExecutor}
* to publish its Messages.
*/
public PublishSubscribeChannel(TaskExecutor taskExecutor) {
this();
if (taskExecutor != null) {
this.dispatcher.setTaskExecutor(taskExecutor);
}
}
public boolean subscribe(MessageTarget target) {
return this.dispatcher.addTarget(target);
}
public boolean unsubscribe(MessageTarget target) {
return this.dispatcher.removeTarget(target);
}
@Override
protected boolean doSend(Message<?> message, long timeout) {
return this.dispatcher.send(message);
}
@Override
protected Message<?> doReceive(long timeout) {
return null;
}
public List<Message<?>> clear() {
return null;
}
public List<Message<?>> purge(MessageSelector selector) {
return null;
}
}

46
org.springframework.integration/src/main/java/org/springframework/integration/channel/config/PublishSubscribeChannelParser.java Normal file
View File

@@ -0,0 +1,46 @@
/*
* 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.channel.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;publish-subscribe-channel&gt; element.
*
* @author Mark Fisher
*/
public class PublishSubscribeChannelParser extends AbstractChannelParser {
@Override
protected Class<?> getBeanClass(Element element) {
return PublishSubscribeChannel.class;
}
@Override
protected void configureConstructorArgs(BeanDefinitionBuilder builder, Element element, DispatcherPolicy dispatcherPolicy) {
String taskExecutorRef = element.getAttribute("task-executor");
if (StringUtils.hasText(taskExecutorRef)) {
builder.addConstructorArgReference(taskExecutorRef);
}
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java
View File

@@ -29,6 +29,7 @@ import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
import org.springframework.core.io.support.PropertiesLoaderUtils;
import org.springframework.integration.channel.config.PriorityChannelParser;
import org.springframework.integration.channel.config.PublishSubscribeChannelParser;
import org.springframework.integration.channel.config.QueueChannelParser;
import org.springframework.integration.channel.config.RendezvousChannelParser;
import org.springframework.integration.channel.config.ThreadLocalChannelParser;
@@ -58,6 +59,7 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {
registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenParser());
registerBeanDefinitionParser("channel", new DefaultChannelParser());
registerBeanDefinitionParser("queue-channel", new QueueChannelParser());
registerBeanDefinitionParser("publish-subscribe-channel", new PublishSubscribeChannelParser());
registerBeanDefinitionParser("direct-channel", new DirectChannelParser());
registerBeanDefinitionParser("priority-channel", new PriorityChannelParser());
registerBeanDefinitionParser("rendezvous-channel", new RendezvousChannelParser());

16
org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -63,6 +63,21 @@
</xsd:annotation>
</xsd:element>
<xsd:element name="publish-subscribe-channel">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a publish-subscribe channel that broadcasts to its targets.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="channelType">
<xsd:attribute name="task-executor" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="direct-channel">
<xsd:complexType>
<xsd:annotation>
@@ -121,7 +136,6 @@
<xsd:element ref="interceptor" minOccurs="0" maxOccurs="unbounded"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID" use="required"/>
<xsd:attribute name="publish-subscribe" type="xsd:boolean" default="false"/>
<xsd:attribute name="datatype" type="xsd:string"/>
</xsd:complexType>

81
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java Normal file
View File

@@ -0,0 +1,81 @@
/*
* 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 java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
/**
* Base class for {@link MessageDispatcher} implementations.
*
* @author Mark Fisher
*/
public abstract class AbstractDispatcher implements MessageDispatcher {
protected final Log logger = LogFactory.getLog(this.getClass());
protected final List<MessageTarget> targets = new CopyOnWriteArrayList<MessageTarget>();
private volatile long timeout = 0;
private volatile TaskExecutor taskExecutor;
public void setTimeout(long timeout) {
this.timeout = timeout;
}
public boolean addTarget(MessageTarget target) {
return this.targets.add(target);
}
public boolean removeTarget(MessageTarget target) {
return this.targets.remove(target);
}
/**
* Specify a {@link TaskExecutor} for invoking the target endpoints.
* If none is provided, the invocation will occur in the thread
* that runs this polling dispatcher.
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
protected TaskExecutor getTaskExecutor() {
return this.taskExecutor;
}
protected final boolean sendMessageToTarget(Message<?> message, MessageTarget target) {
boolean sent = (target instanceof BlockingTarget && this.timeout >= 0)
? ((BlockingTarget) target).send(message, this.timeout)
: target.send(message);
if (!sent && logger.isDebugEnabled()) {
logger.debug("dispatcher failed to send message to target '" + target + "'");
}
return sent;
}
}

50
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* 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.core.task.TaskExecutor;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
/**
* A broadcasting dispatcher implementation. It makes a best effort to
* send the message to each of its targets. If it fails to send to any
* one target, it will log a warn-level message but continue to send
* to the other targets.
*
* @author Mark Fisher
*/
public class BroadcastingDispatcher extends AbstractDispatcher {
public boolean send(final Message<?> message) {
for (final MessageTarget target : this.targets) {
TaskExecutor executor = this.getTaskExecutor();
if (executor != null) {
executor.execute(new Runnable() {
public void run() {
sendMessageToTarget(message, target);
}
});
}
else {
this.sendMessageToTarget(message, target);
}
}
return true;
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/DirectChannel.java
View File

@@ -59,7 +59,7 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl
public boolean subscribe(MessageTarget target) {
boolean added = this.dispatcher.subscribe(target);
boolean added = this.dispatcher.addTarget(target);
if (added) {
this.handlerCount.incrementAndGet();
}
@@ -67,7 +67,7 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl
}
public boolean unsubscribe(MessageTarget target) {
boolean removed = this.dispatcher.unsubscribe(target);
boolean removed = this.dispatcher.removeTarget(target);
if (removed) {
this.handlerCount.decrementAndGet();
}

11
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java
View File

@@ -28,6 +28,15 @@ public interface MessageDispatcher extends MessageTarget {
boolean send(Message<?> message);
void setSendTimeout(long timeout);
/**
* Specify the timeout for sending to a target (in milliseconds).
* Note that this value will only be applicable for blocking targets.
* The default value is 0.
*/
void setTimeout(long timeout);
boolean addTarget(MessageTarget target);
boolean removeTarget(MessageTarget target);
}

119
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java Normal file
View File

@@ -0,0 +1,119 @@
/*
* 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.BlockingSource;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.SchedulableTask;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
*/
public class PollingDispatcher implements SchedulableTask {
private final MessageSource<?> source;
private final MessageDispatcher dispatcher;
private final Schedule schedule;
private volatile long receiveTimeout = 5000;
private volatile int maxMessagesPerTask = 1;
/**
* Create a PollingDispatcher for the provided {@link MessageSource}.
* It can be scheduled according to the specified {@link Schedule}.
*/
public PollingDispatcher(MessageSource<?> source, MessageDispatcher dispatcher, Schedule schedule) {
Assert.notNull(source, "source must not be null");
Assert.notNull(dispatcher, "dispatcher must not be null");
this.source = source;
this.dispatcher = dispatcher;
this.schedule = schedule;
}
/**
* Specify the timeout to use when receiving from the source (in milliseconds).
* Note that this value will only be applicable if the source is an instance
* of {@link BlockingSource}.
* <p/>
* A negative value indicates that receive calls should block indefinitely.
* The default value is 5000 (5 seconds).
*/
public void setReceiveTimeout(long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
/**
* Specify the timeout to use when sending to a target (in milliseconds).
* Note that this value will only be applicable if the target is an instance
* of {@link BlockingTarget}. The default value is 0.
*/
public void setSendTimeout(long sendTimeout) {
this.dispatcher.setTimeout(sendTimeout);
}
/**
* Set the maximum number of messages to receive for each poll.
* A non-positive value indicates that polling should repeat as long
* as non-null messages are being received and successfully sent.
*/
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
this.maxMessagesPerTask = maxMessagesPerTask;
}
public boolean addTarget(MessageTarget target) {
return this.dispatcher.addTarget(target);
}
public boolean removeTarget(MessageTarget target) {
return this.dispatcher.removeTarget(target);
}
public Schedule getSchedule() {
return this.schedule;
}
public void run() {
int count = 0;
while (this.maxMessagesPerTask <= 0 || count < this.maxMessagesPerTask) {
if (!this.dispatch()) {
return;
}
count++;
}
}
private boolean dispatch() {
final Message<?> message = (this.source instanceof BlockingSource && this.receiveTimeout >= 0)
? ((BlockingSource<?>) this.source).receive(this.receiveTimeout)
: this.source.receive();
if (message == null) {
return false;
}
return this.dispatcher.send(message);
}
}

94
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java
View File

@@ -1,94 +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.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.SchedulableTask;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
* A {@link SchedulableTask} that combines polling and dispatching.
*
* @author Mark Fisher
*/
public class PollingDispatcherTask implements SchedulableTask, Subscribable {
private final MessageChannel channel;
private final Schedule schedule;
private final SimpleDispatcher dispatcher;
private volatile long receiveTimeout = 5000;
private volatile int maxMessagesPerTask = 1;
public PollingDispatcherTask(MessageChannel channel, Schedule schedule) {
Assert.notNull(channel, "channel must not be null");
this.channel = channel;
this.schedule = schedule;
this.dispatcher = new SimpleDispatcher(this.channel.getDispatcherPolicy());
}
/**
* Set the maximum amount of time in milliseconds to wait for a message to be available.
* A negative value indicates that receive calls should block indefinitely.
*/
public void setReceiveTimeout(long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
/**
* Set the maximum number of messages for each retrieval attempt.
*/
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagePerTask' must be at least 1");
this.maxMessagesPerTask = maxMessagesPerTask;
}
public boolean subscribe(MessageTarget target) {
return this.dispatcher.subscribe(target);
}
public boolean unsubscribe(MessageTarget target) {
return this.dispatcher.unsubscribe(target);
}
public Schedule getSchedule() {
return this.schedule;
}
public void run() {
int count = 0;
while (count < this.maxMessagesPerTask) {
Message<?> message = (this.receiveTimeout < 0) ?
this.channel.receive() : this.channel.receive(this.receiveTimeout);
if (message == null) {
return;
}
this.dispatcher.send(message);
count++;
}
}
}

31
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java
View File

@@ -19,15 +19,10 @@ package org.springframework.integration.dispatcher;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageTarget;
@@ -37,34 +32,15 @@ import org.springframework.integration.message.MessageTarget;
*
* @author Mark Fisher
*/
public class SimpleDispatcher implements MessageDispatcher {
protected final Log logger = LogFactory.getLog(this.getClass());
private final List<MessageTarget> targets = new CopyOnWriteArrayList<MessageTarget>();
public class SimpleDispatcher extends AbstractDispatcher {
protected final DispatcherPolicy dispatcherPolicy;
private volatile long sendTimeout;
public SimpleDispatcher(DispatcherPolicy dispatcherPolicy) {
this.dispatcherPolicy = (dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy();
}
public void setSendTimeout(long sendTimeout) {
this.sendTimeout = sendTimeout;
}
public boolean subscribe(MessageTarget target) {
return this.targets.add(target);
}
public boolean unsubscribe(MessageTarget target) {
return this.targets.remove(target);
}
public boolean send(Message<?> message) {
int attempts = 0;
List<MessageTarget> targetList = new ArrayList<MessageTarget>(this.targets);
@@ -94,8 +70,7 @@ public class SimpleDispatcher implements MessageDispatcher {
while (iter.hasNext()) {
MessageTarget target = iter.next();
try {
boolean sent = (target instanceof BlockingTarget && this.sendTimeout >= 0) ?
((BlockingTarget) target).send(message, this.sendTimeout) : target.send(message);
boolean sent = this.sendMessageToTarget(message, target);
if (!this.dispatcherPolicy.isPublishSubscribe() && sent) {
return true;
}
@@ -112,7 +87,7 @@ public class SimpleDispatcher implements MessageDispatcher {
catch (MessageHandlerRejectedExecutionException e) {
rejected = true;
if (logger.isDebugEnabled()) {
logger.debug("target is busy, continuing with other targets if available", e);
logger.debug("target '" + target + "' is busy, continuing with other targets if available", e);
}
}
}

73
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
View File

@@ -25,19 +25,22 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Subscription;
/**
* Base class for {@link MessageEndpoint} implementations.
*
* @author Mark Fisher
*/
public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware, Lifecycle, InitializingBean {
public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware, Lifecycle {
protected final Log logger = LogFactory.getLog(this.getClass());
@@ -49,6 +52,16 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
private volatile boolean running;
private volatile Schedule schedule;
private volatile EndpointTrigger trigger;
private volatile Subscription subscription;
private volatile String outputChannelName;
private volatile ChannelRegistry channelRegistry;
private final Object lifecycleMonitor = new Object();
@@ -64,6 +77,60 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
this.setName(beanName);
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
}
public Schedule getSchedule() {
return this.schedule;
}
public void setTrigger(EndpointTrigger trigger) {
this.trigger = trigger;
}
public EndpointTrigger getTrigger() {
return this.trigger;
}
public Subscription getSubscription() {
return this.subscription;
}
public void setSubscription(Subscription subscription) {
this.subscription = subscription;
}
/**
* Set the name of the channel to which this endpoint should send reply
* messages.
*/
public void setOutputChannelName(String outputChannelName) {
this.outputChannelName = outputChannelName;
}
public MessageChannel getOutputChannel() {
if (this.outputChannelName != null && this.channelRegistry != null) {
return this.channelRegistry.lookupChannel(this.outputChannelName);
}
return null;
}
public String getOutputChannelName() {
return this.outputChannelName;
}
/**
* Set the channel registry to use for looking up channels by name.
*/
public void setChannelRegistry(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
}
protected ChannelRegistry getChannelRegistry() {
return this.channelRegistry;
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
@@ -131,7 +198,7 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
}
}
public final boolean invoke(Message<?> message) {
public final boolean send(Message<?> message) {
if (message == null) {
throw new IllegalArgumentException("Message must not be null.");
}

59
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java Normal file
View File

@@ -0,0 +1,59 @@
/*
* 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.endpoint;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.PollCommand;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
/**
* A {@link PollingDispatcher} implementation that sends poll command
* trigger messages to endpoints.
*
* @author Mark Fisher
*/
public class EndpointTrigger extends PollingDispatcher {
/**
* Create a trigger with the specified {@link Schedule}.
*/
public EndpointTrigger(Schedule schedule) {
super(new PollCommandMessageSource(), new BroadcastingDispatcher(), schedule);
}
/**
* Create a trigger. A {@link PollingSchedule} will be created
* with the specified interval.
*/
public EndpointTrigger(long interval) {
this(new PollingSchedule(interval));
}
private static class PollCommandMessageSource implements MessageSource<PollCommand> {
public Message<PollCommand> receive() {
return new GenericMessage<PollCommand>(new PollCommand());
}
}
}

24
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java
View File

@@ -35,7 +35,7 @@ import org.springframework.util.StringUtils;
*
* @author Mark Fisher
*/
public class HandlerEndpoint extends TargetEndpoint implements MessageProducingEndpoint {
public class HandlerEndpoint extends TargetEndpoint {
private volatile MessageHandler handler;
@@ -43,8 +43,6 @@ public class HandlerEndpoint extends TargetEndpoint implements MessageProducingE
private volatile long replyTimeout = 1000;
private volatile String outputChannelName;
private volatile boolean returnAddressOverrides = false;
@@ -73,14 +71,6 @@ public class HandlerEndpoint extends TargetEndpoint implements MessageProducingE
this.replyTimeout = replyTimeout;
}
/**
* Set the name of the channel to which this endpoint should send reply
* messages.
*/
public void setOutputChannelName(String outputChannelName) {
this.outputChannelName = outputChannelName;
}
public void setReturnAddressOverrides(boolean returnAddressOverrides) {
this.returnAddressOverrides = returnAddressOverrides;
}
@@ -128,18 +118,6 @@ public class HandlerEndpoint extends TargetEndpoint implements MessageProducingE
return null;
}
public MessageChannel getOutputChannel() {
ChannelRegistry registry = this.getChannelRegistry();
if (this.outputChannelName != null && registry != null) {
return registry.lookupChannel(this.outputChannelName);
}
return null;
}
public String getOutputChannelName() {
return this.outputChannelName;
}
private static class HandlerInvokingTarget implements MessageTarget {

28
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageConsumingEndpoint.java
View File

@@ -1,28 +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.endpoint;
import org.springframework.integration.scheduling.Subscription;
/**
* @author Mark Fisher
*/
public interface MessageConsumingEndpoint {
Subscription getSubscription();
}

19
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
View File

@@ -16,17 +16,30 @@
package org.springframework.integration.endpoint;
import org.springframework.integration.message.Message;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Subscription;
/**
* Base interface for message endpoints.
*
* @author Mark Fisher
*/
public interface MessageEndpoint {
public interface MessageEndpoint extends MessageTarget, ChannelRegistryAware, InitializingBean {
String getName();
boolean invoke(Message<?> message);
Schedule getSchedule();
EndpointTrigger getTrigger();
Subscription getSubscription();
String getOutputChannelName();
MessageChannel getOutputChannel();
}

30
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducingEndpoint.java
View File

@@ -1,30 +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.endpoint;
import org.springframework.integration.channel.MessageChannel;
/**
* @author Mark Fisher
*/
public interface MessageProducingEndpoint {
String getOutputChannelName();
MessageChannel getOutputChannel();
}

13
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java
View File

@@ -24,7 +24,6 @@ import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.PollCommand;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
@@ -39,25 +38,15 @@ public class SourceEndpoint extends AbstractEndpoint {
private final SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy());
private volatile Schedule schedule;
public SourceEndpoint(MessageSource<?> source, MessageChannel channel) {
Assert.notNull(source, "source must not be null");
Assert.notNull(channel, "channel must not be null");
this.source = source;
this.dispatcher.subscribe(channel);
this.dispatcher.addTarget(channel);
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
}
public Schedule getSchedule() {
return this.schedule;
}
protected boolean supports(Message<?> message) {
return (message.getPayload() instanceof PollCommand);
}

44
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java
View File

@@ -16,10 +16,11 @@
package org.springframework.integration.endpoint;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollCommand;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.Assert;
@@ -29,8 +30,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class TargetEndpoint extends AbstractEndpoint
implements MessageTarget, MessageConsumingEndpoint, ChannelRegistryAware {
public class TargetEndpoint extends AbstractEndpoint {
private volatile MessageTarget target;
@@ -38,8 +38,6 @@ public class TargetEndpoint extends AbstractEndpoint
private volatile MessageSelector selector;
private volatile ChannelRegistry channelRegistry;
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
@@ -75,24 +73,13 @@ public class TargetEndpoint extends AbstractEndpoint
this.subscription = subscription;
}
/**
* Set the channel registry to use for looking up channels by name.
*/
public void setChannelRegistry(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
}
protected ChannelRegistry getChannelRegistry() {
return this.channelRegistry;
}
protected void initialize() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
if (this.target instanceof ChannelRegistryAware && this.channelRegistry != null) {
((ChannelRegistryAware) this.target).setChannelRegistry(this.channelRegistry);
if (this.target instanceof ChannelRegistryAware && this.getChannelRegistry() != null) {
((ChannelRegistryAware) this.target).setChannelRegistry(this.getChannelRegistry());
}
this.initialized = true;
}
@@ -100,6 +87,23 @@ public class TargetEndpoint extends AbstractEndpoint
@Override
protected final boolean doInvoke(Message<?> message) {
if (message.getPayload() instanceof PollCommand) {
MessageChannel channel = this.getSubscription().getChannel();
if (channel == null && this.getSubscription().getChannelName() != null) {
channel = this.getChannelRegistry().lookupChannel(this.getSubscription().getChannelName());
}
if (channel != null) {
Message<?> receivedMessage = channel.receive(5000);
if (receivedMessage != null) {
return this.doInvoke(receivedMessage);
}
}
else if (logger.isDebugEnabled()) {
logger.debug("TargetEndpoint unable to resolve channel '"
+ this.getSubscription().getChannelName() + "'");
}
return false;
}
if (this.selector != null && !this.selector.accept(message)) {
return false;
}
@@ -111,8 +115,4 @@ public class TargetEndpoint extends AbstractEndpoint
return true;
}
public boolean send(Message<?> message) {
return this.invoke(message);
}
}

4
org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java
View File

@@ -30,6 +30,7 @@ import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.handler.MessageHandler;
@@ -129,8 +130,7 @@ public class MessageBusTests {
@Test
public void testBothHandlersReceivePublishSubscribeMessage() throws InterruptedException {
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true);
QueueChannel inputChannel = new QueueChannel(10, dispatcherPolicy);
PublishSubscribeChannel inputChannel = new PublishSubscribeChannel();
QueueChannel outputChannel1 = new QueueChannel();
QueueChannel outputChannel2 = new QueueChannel();
final CountDownLatch latch = new CountDownLatch(2);

470
org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java
View File

@@ -1,470 +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.bus;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
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;
import org.junit.Test;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.MessageEndpointBeanPostProcessor;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
import org.springframework.integration.handler.TestHandlers;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.selector.PayloadTypeSelector;
import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
import org.springframework.integration.scheduling.SimpleTaskScheduler;
/**
* @author Mark Fisher
*/
public class SubscriptionManagerTests {
private SimpleTaskScheduler scheduler = new SimpleTaskScheduler(new ScheduledThreadPoolExecutor(10));
@Test
public void testNonBroadcastingDispatcherSendsToExactlyOneEndpoint() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
QueueChannel channel = new QueueChannel();
channel.send(new StringMessage(1, "test"));
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());
}
@Test
public void testBroadcastingDispatcherSendsToAllEndpoints() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(2);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true));
channel.send(new StringMessage(1, "test"));
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());
}
@Test
public void testNonBroadcastingDispatcherSkipsInactiveExecutor() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
QueueChannel channel = new QueueChannel();
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
MessageTarget inactiveEndpoint = createEndpoint(handler1, true);
manager.addTarget(inactiveEndpoint);
manager.addTarget(createEndpoint(handler2, true));
manager.addTarget(createEndpoint(handler3, true));
manager.start();
((Lifecycle) inactiveEndpoint).stop();
channel.send(new StringMessage(1, "test"));
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
assertEquals("inactive handler should not have received message", 0, counter1.get());
assertEquals("exactly one handler should have received message", 1, counter2.get() + counter3.get());
}
@Test
public void testBroadcastingDispatcherSkipsInactiveExecutor() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(2);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true));
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
MessageTarget inactiveEndpoint = createEndpoint(handler2, true);
manager.addTarget(createEndpoint(handler1, true));
manager.addTarget(inactiveEndpoint);
manager.addTarget(createEndpoint(handler3, true));
manager.start();
((Lifecycle) inactiveEndpoint).stop();
channel.send(new StringMessage(1, "test"));
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
assertEquals("inactive handler should not have received message", 0, counter2.get());
assertEquals("both active handlers should have received message", 2, counter1.get() + counter3.get());
}
@Test
public void testDispatcherWithNoExecutorsDoesNotFail() {
QueueChannel channel = new QueueChannel();
channel.send(new StringMessage(1, "test"));
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
manager.start();
}
@Test
public void testBroadcastingDispatcherReachesRejectionLimitAndShouldFail() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(2);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true));
channel.getDispatcherPolicy().setRejectionLimit(2);
channel.getDispatcherPolicy().setRetryInterval(3);
channel.send(new StringMessage(1, "test"));
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
manager.addTarget(createEndpoint(handler1, true));
manager.addTarget(new MessageTarget() {
public boolean send(Message<?> message) {
throw new MessageHandlerRejectedExecutionException(message);
}
});
manager.addTarget(createEndpoint(handler3, true));
QueueChannel errorChannel = new QueueChannel();
scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel));
manager.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
Message<?> errorMessage = errorChannel.receive(1000);
assertNotNull(errorMessage);
assertTrue(errorMessage instanceof ErrorMessage);
assertEquals(MessageDeliveryException.class, ((ErrorMessage) errorMessage).getPayload().getClass());
}
@Test
public void testBroadcastingDispatcherReachesRejectionLimitAndShouldNotFail() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(3);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true));
channel.getDispatcherPolicy().setRejectionLimit(2);
channel.getDispatcherPolicy().setRetryInterval(3);
channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false);
channel.send(new StringMessage(1, "test"));
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));
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());
}
@Test
public void testNonBroadcastingDispatcherReachesRejectionLimitAndShouldFail() throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(4);
MessageHandler handler1 = TestHandlers.rejectingCountDownHandler(latch);
MessageHandler handler2 = TestHandlers.rejectingCountDownHandler(latch);
QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(false));
channel.send(new StringMessage(1, "test"));
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
channel.getDispatcherPolicy().setRejectionLimit(2);
channel.getDispatcherPolicy().setRetryInterval(3);
manager.addTarget(createEndpoint(handler1, false));
manager.addTarget(createEndpoint(handler2, false));
QueueChannel errorChannel = new QueueChannel();
scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel));
manager.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
Message<?> errorMessage = errorChannel.receive(500);
assertNotNull(errorMessage);
assertTrue(errorMessage instanceof ErrorMessage);
assertEquals(MessageDeliveryException.class, ((ErrorMessage) errorMessage).getPayload().getClass());
}
@Test
public void testNonBroadcastingDispatcherReachesRejectionLimitButShouldNotFail() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger rejectedCounter1 = new AtomicInteger();
final AtomicInteger rejectedCounter2 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(4);
QueueChannel channel = new QueueChannel();
channel.getDispatcherPolicy().setRejectionLimit(2);
channel.getDispatcherPolicy().setRetryInterval(3);
channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false);
channel.send(new StringMessage(1, "test"));
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));
manager.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
rejectedCounter2.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
}, false));
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());
assertEquals("rejecting handlers should not have received message", 0, counter1.get() + counter2.get());
assertEquals("handler1 should have rejected two times", 2, rejectedCounter1.get());
assertEquals("handler2 should have rejected two times", 2, rejectedCounter2.get());
}
@Test
public void testNonBroadcastingDispatcherWithOneEndpointSucceeding() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final AtomicInteger rejectedCounter1 = new AtomicInteger();
final AtomicInteger rejectedCounter2 = new AtomicInteger();
final AtomicInteger rejectedCounter3 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(5);
final MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setRejectionLimit(2);
dispatcherPolicy.setRetryInterval(3);
dispatcherPolicy.setShouldFailOnRejectionLimit(false);
QueueChannel channel = new QueueChannel(25, dispatcherPolicy);
channel.send(new StringMessage(1, "test"));
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));
manager.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
if (rejectedCounter2.get() == 1) {
return handler2.handle(message);
}
rejectedCounter2.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
}, false));
manager.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
rejectedCounter3.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
}, false));
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());
assertEquals("handler2 should have received message the second time", 1, counter2.get());
assertEquals("handler3 should not have received message", 0, counter3.get());
assertEquals("handler1 should have rejected two times", 2, rejectedCounter1.get());
assertEquals("handler2 should have rejected one time", 1, rejectedCounter2.get());
assertEquals("handler3 should have rejected one time", 1, rejectedCounter3.get());
}
@Test
public void testBroadcastingDispatcherStillRetriesRejectedExecutorAfterOtherSucceeds() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger rejectedCounter1 = new AtomicInteger();
final AtomicInteger rejectedCounter2 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(8);
final MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
final MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true);
dispatcherPolicy.setRejectionLimit(5);
dispatcherPolicy.setRetryInterval(3);
dispatcherPolicy.setShouldFailOnRejectionLimit(false);
QueueChannel channel = new QueueChannel(25, dispatcherPolicy);
channel.send(new StringMessage(1, "test"));
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
manager.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
if (rejectedCounter1.get() == 2) {
return handler1.handle(message);
}
rejectedCounter1.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
}, false));
manager.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
if (rejectedCounter2.get() == 4) {
return handler2.handle(message);
}
rejectedCounter2.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
}, false));
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());
assertEquals("handler2 should have received one message", 1, counter2.get());
assertEquals("handler1 should have rejected two times", 2, rejectedCounter1.get());
assertEquals("handler2 should have rejected four times", 4, rejectedCounter2.get());
}
@Test
public void testTwoExecutorsWithSelectorsAndOneAccepts() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
QueueChannel channel = new QueueChannel();
channel.send(new StringMessage(1, "test"));
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1);
HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2);
endpoint1.setMessageSelector(new PayloadTypeSelector(Integer.class));
endpoint2.setMessageSelector(new PayloadTypeSelector(String.class));
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());
assertEquals("handler2 should have accepted the message", 1, counter2.get());
}
@Test
public void testTwoExecutorsWithSelectorsAndNeitherAccepts() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger selectorCounter1 = new AtomicInteger();
final AtomicInteger selectorCounter2 = new AtomicInteger();
final CountDownLatch selectorLatch = new CountDownLatch(2);
final CountDownLatch handlerLatch = new CountDownLatch(1);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, handlerLatch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, handlerLatch);
QueueChannel channel = new QueueChannel();
channel.send(new StringMessage(1, "test"));
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
final HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1);
final HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2);
endpoint1.setMessageSelector(new PayloadTypeSelector(Integer.class) {
@Override
public boolean accept(Message<?> message) {
selectorCounter1.incrementAndGet();
selectorLatch.countDown();
return super.accept(message);
}
});
endpoint2.setMessageSelector(new PayloadTypeSelector(Integer.class) {
@Override
public boolean accept(Message<?> message) {
selectorCounter2.incrementAndGet();
selectorLatch.countDown();
return super.accept(message);
}
});
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());
assertEquals("handler2 should not have accepted the message", 0, counter2.get());
assertEquals("executor1 should have had exactly one attempt", 1, selectorCounter1.get());
assertEquals("executor2 should have had exactly one attempt", 1, selectorCounter2.get());
assertEquals("handlerLatch should not have counted down", 1, handlerLatch.getCount());
}
@Test
public void testBroadcastingDispatcherWithSelectorsAndOneAccepts() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true));
channel.send(new StringMessage(1, "test"));
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1);
HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2);
endpoint1.setMessageSelector(new PayloadTypeSelector(Integer.class));
endpoint2.setMessageSelector(new PayloadTypeSelector(String.class));
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());
assertEquals("endpoint2 should have accepted the message", 1, counter2.get());
}
private static MessageTarget createEndpoint(MessageHandler handler, boolean asynchronous) {
MessageTarget endpoint = new HandlerEndpoint(handler);
if (asynchronous) {
MessageEndpointBeanPostProcessor postProcessor = new MessageEndpointBeanPostProcessor();
((AbstractEndpoint) endpoint).addInterceptor(new ConcurrencyInterceptor(new ConcurrencyPolicy(1, 1)));
endpoint = (MessageTarget) postProcessor.postProcessAfterInitialization(endpoint, "test-endpoint");
}
try {
((InitializingBean) endpoint).afterPropertiesSet();
}
catch (Exception e) {
throw new RuntimeException("failed to initialize endpoint", e);
}
return endpoint;
}
}

98
org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java
View File

@@ -20,27 +20,22 @@ import static org.junit.Assert.assertEquals;
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;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
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.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.TestChannelInterceptor;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessagePriority;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.SimpleTaskScheduler;
/**
* @author Mark Fisher
@@ -65,43 +60,11 @@ public class ChannelParserTests {
}
@Test
public void testPointToPointChannelByDefault() throws InterruptedException {
public void testQueueChannelByDefault() throws InterruptedException {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"channelParserTests.xml", this.getClass());
MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelByDefault");
channel.send(new StringMessage("test"));
SimpleTaskScheduler scheduler = new SimpleTaskScheduler(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);
manager.addTarget(target1);
manager.addTarget(target2);
manager.start();
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
assertEquals(1, counter.get());
}
@Test
public void testPointToPointChannelExplicitlyConfigured() throws InterruptedException {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"channelParserTests.xml", this.getClass());
MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelExplicitlyConfigured");
channel.send(new StringMessage("test"));
SimpleTaskScheduler scheduler = new SimpleTaskScheduler(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);
manager.addTarget(target1);
manager.addTarget(target2);
manager.start();
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
assertEquals(1, counter.get());
MessageChannel channel = (MessageChannel) context.getBean("queueChannelByDefault");
assertEquals(QueueChannel.class, channel.getClass());
}
@Test
@@ -109,26 +72,27 @@ public class ChannelParserTests {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"channelParserTests.xml", this.getClass());
MessageChannel channel = (MessageChannel) context.getBean("publishSubscribeChannel");
channel.send(new StringMessage("test"));
SimpleTaskScheduler scheduler = new SimpleTaskScheduler(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);
manager.addTarget(target1);
manager.addTarget(target2);
manager.start();
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
assertEquals(2, counter.get());
assertEquals(PublishSubscribeChannel.class, channel.getClass());
}
@Test
public void testPublishSubscribeChannelWithTaskExecutorReference() throws InterruptedException {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"channelParserTests.xml", this.getClass());
MessageChannel channel = (MessageChannel) context.getBean("publishSubscribeChannelWithTaskExecutorRef");
assertEquals(PublishSubscribeChannel.class, channel.getClass());
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
accessor = new DirectFieldAccessor(accessor.getPropertyValue("dispatcher"));
Object taskExecutorProperty = accessor.getPropertyValue("taskExecutor");
Object taskExecutorBean = context.getBean("taskExecutor");
assertEquals(taskExecutorBean, taskExecutorProperty);
}
@Test
public void testDefaultDispatcherPolicy() throws InterruptedException {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"channelParserTests.xml", this.getClass());
MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelByDefault");
MessageChannel channel = (MessageChannel) context.getBean("queueChannelByDefault");
DispatcherPolicy dispatcherPolicy = channel.getDispatcherPolicy();
assertFalse(dispatcherPolicy.isPublishSubscribe());
assertEquals(DispatcherPolicy.DEFAULT_REJECTION_LIMIT, dispatcherPolicy.getRejectionLimit());
@@ -142,7 +106,6 @@ public class ChannelParserTests {
"channelParserTests.xml", this.getClass());
MessageChannel channel = (MessageChannel) context.getBean("channelWithDispatcherPolicy");
DispatcherPolicy dispatcherPolicy = channel.getDispatcherPolicy();
assertTrue(dispatcherPolicy.isPublishSubscribe());
assertEquals(7, dispatcherPolicy.getRejectionLimit());
assertEquals(77, dispatcherPolicy.getRetryInterval());
assertFalse(dispatcherPolicy.getShouldFailOnRejectionLimit());
@@ -267,23 +230,4 @@ public class ChannelParserTests {
assertTrue(threwException);
}
private static class TestTarget implements MessageTarget {
private AtomicInteger counter;
private CountDownLatch latch;
TestTarget(AtomicInteger counter, CountDownLatch latch) {
this.counter = counter;
this.latch = latch;
}
public boolean send(Message<?> message) {
this.counter.incrementAndGet();
this.latch.countDown();
return true;
}
}
}

11
org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml
View File

@@ -9,13 +9,14 @@
<queue-channel id="capacityChannel" capacity="10"/>
<channel id="pointToPointChannelByDefault"/>
<channel id="queueChannelByDefault"/>
<channel id="pointToPointChannelExplicitlyConfigured" publish-subscribe="false"/>
<publish-subscribe-channel id="publishSubscribeChannel"/>
<channel id="publishSubscribeChannel" publish-subscribe="true"/>
<publish-subscribe-channel id="publishSubscribeChannelWithTaskExecutorRef"
task-executor="taskExecutor"/>
<channel id="channelWithDispatcherPolicy" publish-subscribe="true">
<channel id="channelWithDispatcherPolicy">
<dispatcher-policy rejection-limit="7"
retry-interval="77"
should-fail-on-rejection-limit="false"/>
@@ -27,4 +28,6 @@
<channel id="stringOrNumberChannel" datatype="java.lang.String,java.lang.Number"/>
<beans:bean id="taskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor"/>
</beans:beans>

9
org.springframework.integration/src/test/java/org/springframework/integration/config/MessageEndpointBeanPostProcessorTests.java
View File

@@ -47,14 +47,13 @@ public class MessageEndpointBeanPostProcessorTests {
public void testHandlerEndpointWithAdviceChain() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"messageEndpointBeanPostProcessorTests.xml", this.getClass());
context.start();
MessageEndpoint endpoint = (MessageEndpoint) context.getBean("handlerEndpointWithAdvice");
assertTrue(AopUtils.isAopProxy(endpoint));
TestBeforeAdvice beforeAdvice = (TestBeforeAdvice) context.getBean("simpleAdvice");
TestEndpointInterceptor interceptor = (TestEndpointInterceptor) context.getBean("interceptor");
assertEquals(0, beforeAdvice.getCount());
assertEquals(0, interceptor.getCount());
endpoint.invoke(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
assertEquals(1, beforeAdvice.getCount());
assertEquals(2, interceptor.getCount());
context.stop();
@@ -72,14 +71,13 @@ public class MessageEndpointBeanPostProcessorTests {
public void testTargetEndpointWithAdviceChain() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"messageEndpointBeanPostProcessorTests.xml", this.getClass());
context.start();
MessageEndpoint endpoint = (MessageEndpoint) context.getBean("targetEndpointWithAdvice");
assertTrue(AopUtils.isAopProxy(endpoint));
TestBeforeAdvice beforeAdvice = (TestBeforeAdvice) context.getBean("simpleAdvice");
TestEndpointInterceptor interceptor = (TestEndpointInterceptor) context.getBean("interceptor");
assertEquals(0, beforeAdvice.getCount());
assertEquals(0, interceptor.getCount());
endpoint.invoke(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
assertEquals(1, beforeAdvice.getCount());
assertEquals(2, interceptor.getCount());
context.stop();
@@ -97,14 +95,13 @@ public class MessageEndpointBeanPostProcessorTests {
public void testSourceEndpointWithAdviceChain() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"messageEndpointBeanPostProcessorTests.xml", this.getClass());
context.start();
MessageEndpoint endpoint = (MessageEndpoint) context.getBean("sourceEndpointWithAdvice");
assertTrue(AopUtils.isAopProxy(endpoint));
TestBeforeAdvice beforeAdvice = (TestBeforeAdvice) context.getBean("simpleAdvice");
TestEndpointInterceptor interceptor = (TestEndpointInterceptor) context.getBean("interceptor");
assertEquals(0, beforeAdvice.getCount());
assertEquals(0, interceptor.getCount());
endpoint.invoke(new CommandMessage(new PollCommand()));
endpoint.send(new CommandMessage(new PollCommand()));
assertEquals(1, beforeAdvice.getCount());
assertEquals(2, interceptor.getCount());
context.stop();

2
org.springframework.integration/src/test/java/org/springframework/integration/config/messageEndpointBeanPostProcessorTests.xml
View File

@@ -7,7 +7,7 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<message-bus/>
<message-bus auto-startup="false"/>
<channel id="testChannel"/>

10
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java
View File

@@ -40,7 +40,7 @@ public class SimpleDispatcherTests {
public void testSingleMessage() throws InterruptedException {
SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy());
final CountDownLatch latch = new CountDownLatch(1);
dispatcher.subscribe(createEndpoint(TestHandlers.countDownHandler(latch)));
dispatcher.addTarget(createEndpoint(TestHandlers.countDownHandler(latch)));
dispatcher.send(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -52,8 +52,8 @@ public class SimpleDispatcherTests {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)));
dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)));
dispatcher.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)));
dispatcher.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)));
dispatcher.send(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -66,8 +66,8 @@ public class SimpleDispatcherTests {
final CountDownLatch latch = new CountDownLatch(2);
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)));
dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)));
dispatcher.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)));
dispatcher.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)));
dispatcher.send(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());

4
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java
View File

@@ -42,7 +42,7 @@ public class SourceEndpointTests {
QueueChannel channel = new QueueChannel();
SourceEndpoint endpoint = new SourceEndpoint(source, channel);
endpoint.afterPropertiesSet();
endpoint.invoke(new CommandMessage(new PollCommand()));
endpoint.send(new CommandMessage(new PollCommand()));
Message<?> message = channel.receive(1000);
assertNotNull("message should not be null", message);
assertEquals("testing.1", message.getPayload());
@@ -55,7 +55,7 @@ public class SourceEndpointTests {
SourceEndpoint endpoint = new SourceEndpoint(source, channel);
endpoint.setAutoStartup(false);
endpoint.afterPropertiesSet();
endpoint.invoke(new CommandMessage(new PollCommand()));
endpoint.send(new CommandMessage(new PollCommand()));
}
private static class TestSource implements MessageSource<String> {

18
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java
View File

@@ -79,13 +79,13 @@ public class TransactionInterceptorTests {
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("required");
assertEquals(0, txManager.getCommitCount());
endpoint.invoke(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
assertEquals(1, txManager.getCommitCount());
TestTransactionManager outerTxManager = new TestTransactionManager();
TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager);
txTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
return endpoint.invoke(new StringMessage("test"));
return endpoint.send(new StringMessage("test"));
}
});
assertEquals(1, outerTxManager.getCommitCount());
@@ -100,13 +100,13 @@ public class TransactionInterceptorTests {
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("requiresNew");
assertEquals(0, txManager.getCommitCount());
endpoint.invoke(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
assertEquals(1, txManager.getCommitCount());
TestTransactionManager outerTxManager = new TestTransactionManager();
TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager);
txTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
return endpoint.invoke(new StringMessage("test"));
return endpoint.send(new StringMessage("test"));
}
});
assertEquals(1, outerTxManager.getCommitCount());
@@ -121,13 +121,13 @@ public class TransactionInterceptorTests {
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("supports");
assertEquals(0, txManager.getCommitCount());
endpoint.invoke(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
assertEquals(0, txManager.getCommitCount());
TestTransactionManager outerTxManager = new TestTransactionManager();
TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager);
txTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
return endpoint.invoke(new StringMessage("test"));
return endpoint.send(new StringMessage("test"));
}
});
assertEquals(0, txManager.getCommitCount());
@@ -141,13 +141,13 @@ public class TransactionInterceptorTests {
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("notSupported");
assertEquals(0, txManager.getCommitCount());
endpoint.invoke(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
assertEquals(0, txManager.getCommitCount());
TestTransactionManager outerTxManager = new TestTransactionManager();
TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager);
txTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
return endpoint.invoke(new StringMessage("test"));
return endpoint.send(new StringMessage("test"));
}
});
assertEquals(0, txManager.getCommitCount());
@@ -159,7 +159,7 @@ public class TransactionInterceptorTests {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"transactionInterceptorPropagationTests.xml", this.getClass());
final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("mandatory");
endpoint.invoke(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
}
}