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

Split MessageSource types into 2 sub-interfaces: PollableSource and SubscribableSource. The MessageChannel hierarchy has also been revised accordingly. DirectChannel and PublishSubscribeChannel are now SubscribableSources, while the other queue-based channels are PollableSources. The PollableChannel interface extends BlockingSource which in turn is an extension of PollableSource that adds timeout-aware methods.

Browse Source
This commit is contained in:
Mark Fisher
2008-07-30 20:48:00 +00:00
parent 759b5f6d0e
commit fa58dc9457
77 changed files with 422 additions and 497 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultChannelFactoryBean.java
View File

@@ -31,6 +31,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.factory.ChannelFactory;
import org.springframework.integration.channel.factory.QueueChannelFactory;
import org.springframework.util.Assert;
@@ -73,11 +74,11 @@ public class DefaultChannelFactoryBean implements ApplicationContextAware, Facto
}
public void afterPropertiesSet() throws Exception {
synchronized (initializationMonitor) {
synchronized (this.initializationMonitor) {
if (!initialized) {
this.proxyBean = Proxy.newProxyInstance(
getClass().getClassLoader(),
new Class[]{MessageChannel.class},
new Class[] { PollableChannel.class },
new DefaultChannelInvocationHandler());
this.initialized = true;
}
@@ -85,14 +86,14 @@ public class DefaultChannelFactoryBean implements ApplicationContextAware, Facto
}
public Object getObject() throws Exception {
if (!initialized) {
if (!this.initialized) {
afterPropertiesSet();
}
return proxyBean;
}
public Class<?> getObjectType() {
return MessageChannel.class;
return PollableChannel.class;
}
public boolean isSingleton() {

6
org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java
View File

@@ -56,7 +56,7 @@ import org.springframework.integration.endpoint.TargetEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.message.SubscribableSource;
import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
@@ -330,8 +330,8 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
endpoint.setSource(source);
}
}
if (source != null && source instanceof Subscribable) {
((Subscribable) source).subscribe(endpoint);
if (source != null && source instanceof SubscribableSource) {
((SubscribableSource) source).subscribe(endpoint);
if (logger.isInfoEnabled()) {
logger.info("activated subscription to channel '"
+ source + "' for endpoint '" + endpoint + "'");

52
org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -81,6 +81,13 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName
this.interceptors.add(interceptor);
}
/**
* Exposes the interceptor list for subclasses.
*/
protected ChannelInterceptorList getInterceptors() {
return this.interceptors;
}
/**
* Send a message on this channel. If the channel is at capacity, this
* method will block until either space becomes available or the sending
@@ -119,44 +126,10 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName
return sent;
}
/**
* Receive the first available message from this channel. If the channel
* contains no messages, this method will block.
*
* @return the first available message or <code>null</code> if the
* receiving thread is interrupted.
*/
public final Message<?> receive() {
return this.receive(-1);
}
/**
* Receive the first available message from this channel. If the channel
* contains no messages, this method will block until the allotted timeout
* elapses. If the specified timeout is 0, the method will return
* immediately. If less than zero, it will block indefinitely (see
* {@link #receive()}).
*
* @param timeout the timeout in milliseconds
*
* @return the first available message or <code>null</code> if no message
* is available within the allotted time or the receiving thread is
* interrupted.
*/
public final Message<?> receive(long timeout) {
if (!this.interceptors.preReceive(this)) {
return null;
}
Message<?> message = this.doReceive(timeout);
message = this.interceptors.postReceive(message, this);
return message;
}
public String toString() {
return (this.name != null) ? this.name : super.toString();
}
/**
* Subclasses must implement this method. A non-negative timeout indicates
* how long to wait if the channel is at capacity (if the value is 0, it
@@ -166,20 +139,11 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName
*/
protected abstract boolean doSend(Message<?> message, long timeout);
/**
* Subclasses must implement this method. A non-negative timeout indicates
* how long to wait if the channel is empty (if the value is 0, it must
* return immediately with or without success). A negative timeout value
* indicates that the method should block until either a message is
* available or the blocking thread is interrupted.
*/
protected abstract Message<?> doReceive(long timeout);
/**
* A convenience wrapper class for the list of ChannelInterceptors.
*/
private class ChannelInterceptorList {
protected class ChannelInterceptorList {
private final List<ChannelInterceptor> interceptors = new CopyOnWriteArrayList<ChannelInterceptor>();

71
org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java Normal file
View File

@@ -0,0 +1,71 @@
/*
* 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 org.springframework.integration.message.Message;
/**
* Base class for all pollable channels.
*
* @author Mark Fisher
*/
public abstract class AbstractPollableChannel extends AbstractMessageChannel implements PollableChannel {
/**
* Receive the first available message from this channel. If the channel
* contains no messages, this method will block.
*
* @return the first available message or <code>null</code> if the
* receiving thread is interrupted.
*/
public final Message<?> receive() {
return this.receive(-1);
}
/**
* Receive the first available message from this channel. If the channel
* contains no messages, this method will block until the allotted timeout
* elapses. If the specified timeout is 0, the method will return
* immediately. If less than zero, it will block indefinitely (see
* {@link #receive()}).
*
* @param timeout the timeout in milliseconds
*
* @return the first available message or <code>null</code> if no message
* is available within the allotted time or the receiving thread is
* interrupted.
*/
public final Message<?> receive(long timeout) {
if (!this.getInterceptors().preReceive(this)) {
return null;
}
Message<?> message = this.doReceive(timeout);
message = this.getInterceptors().postReceive(message, this);
return message;
}
/**
* Subclasses must implement this method. A non-negative timeout indicates
* how long to wait if the channel is empty (if the value is 0, it must
* return immediately with or without success). A negative timeout value
* indicates that the method should block until either a message is
* available or the blocking thread is interrupted.
*/
protected abstract Message<?> doReceive(long timeout);
}

10
org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelPurger.java
View File

@@ -25,7 +25,7 @@ import org.springframework.util.Assert;
/**
* A utility class for purging {@link Message Messages} from one or more
* {@link MessageChannel MessageChannels}. Any message that does <em>not</em>
* {@link PollableChannel PollableChannels}. Any message that does <em>not</em>
* match the provided {@link MessageSelector} will be removed from the channel.
* If no {@link MessageSelector} is provided, then <em>all</em> messages will be
* cleared from the channel.
@@ -41,16 +41,16 @@ import org.springframework.util.Assert;
*/
public class ChannelPurger {
private final MessageChannel[] channels;
private final PollableChannel[] channels;
private final MessageSelector selector;
public ChannelPurger(MessageChannel ... channels) {
public ChannelPurger(PollableChannel ... channels) {
this(null, channels);
}
public ChannelPurger(MessageSelector selector, MessageChannel ... channels) {
public ChannelPurger(MessageSelector selector, PollableChannel ... channels) {
Assert.notEmpty(channels, "at least one channel is required");
if (channels.length == 1) {
Assert.notNull(channels[0], "channel must not be null");
@@ -62,7 +62,7 @@ public class ChannelPurger {
public final List<Message<?>> purge() {
List<Message<?>> purgedMessages = new ArrayList<Message<?>>();
for (MessageChannel channel : this.channels) {
for (PollableChannel channel : this.channels) {
List<Message<?>> results = (this.selector == null) ?
channel.clear() : channel.purge(this.selector);
if (results != null) {

18
org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannel.java
View File

@@ -16,19 +16,15 @@
package org.springframework.integration.channel;
import java.util.List;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.message.MessageSource;
/**
* Base channel interface defining common behavior for message sending and receiving.
*
* @author Mark Fisher
*/
public interface MessageChannel extends BlockingSource, BlockingTarget {
public interface MessageChannel extends MessageSource, BlockingTarget {
/**
* Return the name of this channel.
@@ -40,14 +36,4 @@ public interface MessageChannel extends BlockingSource, BlockingTarget {
*/
void setName(String name);
/**
* Remove all {@link Message Messages} from this channel.
*/
List<Message<?>> clear();
/**
* Remove any {@link Message Messages} that are not accepted by the provided selector.
*/
List<Message<?>> purge(MessageSelector selector);
}

40
org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* 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.integration.message.BlockingSource;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
/**
* @author Mark Fisher
*/
public interface PollableChannel extends MessageChannel, BlockingSource {
/**
* Remove all {@link Message Messages} from this channel.
*/
List<Message<?>> clear();
/**
* Remove any {@link Message Messages} that are not accepted by the provided selector.
*/
List<Message<?>> purge(MessageSelector selector);
}

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/QueueChannel.java
View File

@@ -35,7 +35,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class QueueChannel extends AbstractMessageChannel {
public class QueueChannel extends AbstractPollableChannel {
public static final int DEFAULT_CAPACITY = 100;

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/ThreadLocalChannel.java
View File

@@ -33,7 +33,7 @@ import org.springframework.integration.message.selector.MessageSelector;
* @author Dave Syer
* @author Mark Fisher
*/
public class ThreadLocalChannel extends AbstractMessageChannel {
public class ThreadLocalChannel extends AbstractPollableChannel {
private static final ThreadLocalMessageHolder messageHolder = new ThreadLocalMessageHolder();

10
org.springframework.integration/src/main/java/org/springframework/integration/config/DirectChannelParser.java
View File

@@ -18,10 +18,8 @@ package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.integration.channel.config.AbstractChannelParser;
import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;direct-channel&gt; element.
@@ -35,12 +33,4 @@ public class DirectChannelParser extends AbstractChannelParser {
return DirectChannel.class;
}
@Override
protected void configureConstructorArgs(BeanDefinitionBuilder builder, Element element) {
String source = element.getAttribute("source");
if (StringUtils.hasText(source)) {
builder.addConstructorArgReference(source);
}
}
}

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

@@ -79,19 +79,12 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="direct-channel">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
<xsd:element name="direct-channel" type="channelType">
<xsd:annotation>
<xsd:documentation>
Defines a channel that invokes its handlers directly in the sender's thread.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="channelType">
<xsd:attribute name="source" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="priority-channel">

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

@@ -16,50 +16,28 @@
package org.springframework.integration.dispatcher;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.message.SubscribableSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.selector.MessageSelector;
/**
* A channel that invokes the subscribed {@link MessageHandler handler(s)} in a
* sender's thread (returning after at most one handles the message). If a
* {@link MessageSource} is provided, then that source will likewise be polled
* within a receiver's thread.
* A channel that invokes the subscribed {@link MessageHandler handler(s)} in
* the sender's thread (returning after at most one handles the message).
*
* @author Dave Syer
* @author Mark Fisher
*/
public class DirectChannel extends AbstractMessageChannel implements Subscribable {
public class DirectChannel extends AbstractMessageChannel implements SubscribableSource {
private volatile MessageSource<?> source;
private final SimpleDispatcher dispatcher;
private final SimpleDispatcher dispatcher = new SimpleDispatcher();
private final AtomicInteger handlerCount = new AtomicInteger();
public DirectChannel() {
this(null);
}
public DirectChannel(MessageSource<?> source) {
this.source = source;
this.dispatcher = new SimpleDispatcher();
}
public void setSource(MessageSource<?> source) {
this.source = source;
}
public boolean subscribe(MessageTarget target) {
boolean added = this.dispatcher.addTarget(target);
if (added) {
@@ -76,15 +54,6 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl
return removed;
}
@Override
protected Message<?> doReceive(long timeout) {
if (this.source != null) {
return this.source.receive();
}
return null;
}
@Override
protected boolean doSend(Message<?> message, long timeout) {
if (message != null && this.handlerCount.get() > 0) {
@@ -93,12 +62,4 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl
return false;
}
public List<Message<?>> clear() {
return new ArrayList<Message<?>>();
}
public List<Message<?>> purge(MessageSelector selector) {
return new ArrayList<Message<?>>();
}
}

8
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java
View File

@@ -19,8 +19,8 @@ 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.message.PollableSource;
import org.springframework.integration.scheduling.SchedulableTask;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
@@ -30,7 +30,7 @@ import org.springframework.util.Assert;
*/
public class PollingDispatcher implements SchedulableTask {
private final MessageSource<?> source;
private final PollableSource<?> source;
private final MessageDispatcher dispatcher;
@@ -42,10 +42,10 @@ public class PollingDispatcher implements SchedulableTask {
/**
* Create a PollingDispatcher for the provided {@link MessageSource}.
* Create a PollingDispatcher for the provided {@link PollableSource}.
* It can be scheduled according to the specified {@link Schedule}.
*/
public PollingDispatcher(MessageSource<?> source, MessageDispatcher dispatcher, Schedule schedule) {
public PollingDispatcher(PollableSource<?> source, MessageDispatcher dispatcher, Schedule schedule) {
Assert.notNull(source, "source must not be null");
Assert.notNull(dispatcher, "dispatcher must not be null");
this.source = source;

20
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PublishSubscribeChannel.java
View File

@@ -16,19 +16,16 @@
package org.springframework.integration.dispatcher;
import java.util.List;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.AbstractMessageChannel;
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;
import org.springframework.integration.message.SubscribableSource;
/**
* @author Mark Fisher
*/
public class PublishSubscribeChannel extends AbstractMessageChannel implements Subscribable {
public class PublishSubscribeChannel extends AbstractMessageChannel implements SubscribableSource {
private final BroadcastingDispatcher dispatcher = new BroadcastingDispatcher();
@@ -64,17 +61,4 @@ public class PublishSubscribeChannel extends AbstractMessageChannel implements S
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;
}
}

15
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointPoller.java
View File

@@ -16,7 +16,10 @@
package org.springframework.integration.endpoint;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.PollableSource;
/**
* @author Mark Fisher
@@ -25,13 +28,23 @@ public class EndpointPoller implements EndpointVisitor {
private final MessageExchangeTemplate template;
public EndpointPoller() {
this.template = new MessageExchangeTemplate();
this.template.setSendTimeout(0);
}
public void visitEndpoint(MessageEndpoint endpoint) {
template.receiveAndForward(endpoint.getSource(), endpoint);
MessageSource<?> source = endpoint.getSource();
if (source == null) {
throw new ConfigurationException("unable to poll for endpoint '"
+ endpoint + "', source is null");
}
if (!(source instanceof PollableSource)) {
throw new ConfigurationException("unable to poll for endpoint '"
+ endpoint + ", source is not a PollableSource");
}
this.template.receiveAndForward((PollableSource<?>) source, endpoint);
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java
View File

@@ -19,7 +19,7 @@ package org.springframework.integration.endpoint;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
@@ -55,7 +55,7 @@ public class EndpointTrigger extends PollingDispatcher {
}
private static class TriggerSource implements MessageSource<EndpointPoller> {
private static class TriggerSource implements PollableSource<EndpointPoller> {
public Message<EndpointPoller> receive() {
return new TriggerMessage();

5
org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java
View File

@@ -20,6 +20,7 @@ import org.springframework.integration.ConfigurationException;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.bus.MessageBusAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.endpoint.EndpointRegistry;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.MessagingGateway;
@@ -46,7 +47,7 @@ public class SimpleMessagingGateway extends MessagingGatewaySupport implements M
private volatile MessageChannel requestChannel;
private volatile MessageChannel replyChannel;
private volatile PollableChannel replyChannel;
private volatile long replyTimeout = 5000;
@@ -87,7 +88,7 @@ public class SimpleMessagingGateway extends MessagingGatewaySupport implements M
*
* @param replyChannel the channel from which reply messages will be received
*/
public void setReplyChannel(MessageChannel replyChannel) {
public void setReplyChannel(PollableChannel replyChannel) {
this.replyChannel = replyChannel;
}

5
org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java
View File

@@ -22,7 +22,6 @@ import java.util.concurrent.FutureTask;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.Assert;
@@ -78,7 +77,7 @@ public class AsyncMessageExchangeTemplate extends MessageExchangeTemplate {
*/
@Override
@SuppressWarnings("unchecked")
public Message<?> receive(final MessageSource<?> source) {
public Message<?> receive(final PollableSource<?> source) {
FutureTask<Message<?>> task = new FutureTask<Message<?>>(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return AsyncMessageExchangeTemplate.super.receive(source);
@@ -95,7 +94,7 @@ public class AsyncMessageExchangeTemplate extends MessageExchangeTemplate {
* unless an exception is thrown by the executor.
*/
@Override
public boolean receiveAndForward(final MessageSource<?> source, final MessageTarget target) {
public boolean receiveAndForward(final PollableSource<?> source, final MessageTarget target) {
this.taskExecutor.execute(new Runnable() {
public void run() {
AsyncMessageExchangeTemplate.super.receiveAndForward(source, target);

2
org.springframework.integration/src/main/java/org/springframework/integration/message/BlockingSource.java
View File

@@ -21,7 +21,7 @@ package org.springframework.integration.message;
*
* @author Mark Fisher
*/
public interface BlockingSource<T> extends MessageSource<T> {
public interface BlockingSource<T> extends PollableSource<T> {
/**
* Receive a message, blocking indefinitely if necessary.

13
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java
View File

@@ -27,7 +27,6 @@ import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
@@ -36,9 +35,9 @@ import org.springframework.transaction.support.TransactionTemplate;
/**
* This is the central class for invoking message exchange operations
* across {@link MessageSource}s and {@link MessageTarget}s. It supports
* across {@link PollableSource}s and {@link MessageTarget}s. It supports
* one-way send and receive calls as well as request/reply. Additionally,
* the {@link #receiveAndForward(MessageSource, MessageTarget)} method
* the {@link #receiveAndForward(PollableSource, MessageTarget)} method
* plays the role of a polling-consumer while actually sending any
* received message to an event-driven consumer.
*
@@ -167,7 +166,7 @@ public class MessageExchangeTemplate implements InitializingBean {
return this.doSendAndReceive(request, target);
}
public Message<?> receive(final MessageSource<?> source) {
public Message<?> receive(final PollableSource<?> source) {
TransactionTemplate txTemplate = this.getTransactionTemplate();
if (txTemplate != null) {
return (Message<?>) txTemplate.execute(new TransactionCallback() {
@@ -179,7 +178,7 @@ public class MessageExchangeTemplate implements InitializingBean {
return this.doReceive(source);
}
public boolean receiveAndForward(final MessageSource<?> source, final MessageTarget target) {
public boolean receiveAndForward(final PollableSource<?> source, final MessageTarget target) {
TransactionTemplate txTemplate = this.getTransactionTemplate();
if (txTemplate != null) {
return (Boolean) txTemplate.execute(new TransactionCallback() {
@@ -203,7 +202,7 @@ public class MessageExchangeTemplate implements InitializingBean {
return sent;
}
private Message<?> doReceive(MessageSource<?> source) {
private Message<?> doReceive(PollableSource<?> source) {
long timeout = this.receiveTimeout;
Message<?> message = (timeout >= 0 && source instanceof BlockingSource)
? ((BlockingSource<?>) source).receive(timeout)
@@ -223,7 +222,7 @@ public class MessageExchangeTemplate implements InitializingBean {
return this.doReceive(returnAddress);
}
private boolean doReceiveAndForward(MessageSource<?> source, MessageTarget target) {
private boolean doReceiveAndForward(PollableSource<?> source, MessageTarget target) {
Message<?> message = this.doReceive(source);
if (message == null) {
return false;

5
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageSource.java
View File

@@ -23,9 +23,4 @@ package org.springframework.integration.message;
*/
public interface MessageSource<T> {
/**
* Retrieve a message from this source or <code>null</code> if no message is available.
*/
Message<T> receive();
}

2
org.springframework.integration/src/main/java/org/springframework/integration/message/MethodInvokingSource.java
View File

@@ -31,7 +31,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class MethodInvokingSource implements MessageSource<Object>, InitializingBean {
public class MethodInvokingSource implements PollableSource<Object>, InitializingBean {
private Object object;

31
org.springframework.integration/src/main/java/org/springframework/integration/message/PollableSource.java Normal file
View File

@@ -0,0 +1,31 @@
/*
* Copyright 2002-2008 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.message;
/**
* Base interface for any source of {@link Message Messages} that can be polled.
*
* @author Mark Fisher
*/
public interface PollableSource<T> extends MessageSource<T> {
/**
* Retrieve a message from this source or <code>null</code> if no message is available.
*/
Message<T> receive();
}

4
org.springframework.integration/src/main/java/org/springframework/integration/message/Subscribable.java → org.springframework.integration/src/main/java/org/springframework/integration/message/SubscribableSource.java
View File

@@ -17,11 +17,11 @@
package org.springframework.integration.message;
/**
* Interface for any component that accepts subscribers.
* Interface for any source of messages that accepts subscribers.
*
* @author Mark Fisher
*/
public interface Subscribable {
public interface SubscribableSource extends MessageSource {
/**
* Register a {@link MessageTarget} as a subscriber to this source.