diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java index 5d8536ecd5..bce136d94d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java @@ -18,14 +18,32 @@ 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 { +public interface PollableChannel extends MessageChannel { + + /** + * Receive a message from this channel, blocking indefinitely if necessary. + * + * @return the next available {@link Message} or null if + * interrupted + */ + Message receive(); + + /** + * Receive a message from this channel, blocking until either a message is + * available or the specified timeout period elapses. + * + * @param timeout the timeout in milliseconds + * + * @return the next available {@link Message} or null if the + * specified timeout period elapses or the message reception is interrupted + */ + Message receive(long timeout); /** * Remove all {@link Message Messages} from this channel. diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingGateway.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingGateway.java index 4367dc641e..dcb83c87a4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingGateway.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingGateway.java @@ -20,7 +20,7 @@ import org.springframework.integration.message.Message; /** * Base interface for gateway adapters. In Spring Integration, a "gateway" is a - * component that sends messages to and receives messages from message channels + * component that sends messages to and/or receives messages from message channels * so that application code does not need to be aware of channels or even messages. * * @author Mark Fisher @@ -35,6 +35,4 @@ public interface MessagingGateway { Message sendAndReceiveMessage(Object object); - void receiveAndForward(); - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java index 0d459aadad..196e642c9b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java @@ -141,15 +141,6 @@ public class SimpleMessagingGateway extends MessagingGatewaySupport implements M return (message != null) ? this.messageMapper.mapMessage(message) : null; } - public void receiveAndForward() { - if (this.replyChannel == null || this.requestChannel == null) { - throw new IllegalStateException( - "receiveAndForward is not supported, because either the request or reply channel" - + " has not been configured"); - } - this.getMessageExchangeTemplate().receiveAndForward(this.replyChannel, this.requestChannel); - } - public Object sendAndReceive(Object object) { return this.sendAndReceive(object, true); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java index b3a84d51c2..40fec6ba2d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java @@ -21,6 +21,7 @@ import java.util.concurrent.FutureTask; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; import org.springframework.util.Assert; /** @@ -71,34 +72,18 @@ public class AsyncMessageExchangeTemplate extends MessageExchangeTemplate { } /** - * Receive an {@link AsyncMessage} from the provided source. + * Receive an {@link AsyncMessage} from the provided channel. */ @Override @SuppressWarnings("unchecked") - public Message receive(final PollableSource source) { + public Message receive(final PollableChannel channel) { FutureTask> task = new FutureTask>(new Callable>() { public Message call() throws Exception { - return AsyncMessageExchangeTemplate.super.receive(source); + return AsyncMessageExchangeTemplate.super.receive(channel); } }); this.taskExecutor.execute(task); return new AsyncMessage(task); } - /** - * Receive a Message from the provided source and if not null, - * send it to the given channel. Note that the receive and send operations - * occur asynchronously, so this method will always return true - * unless an exception is thrown by the executor. - */ - @Override - public boolean receiveAndForward(final PollableSource source, final MessageChannel channel) { - this.taskExecutor.execute(new Runnable() { - public void run() { - AsyncMessageExchangeTemplate.super.receiveAndForward(source, channel); - } - }); - return true; - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java index 1ac3b40a4a..771295d670 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java @@ -16,6 +16,7 @@ package org.springframework.integration.message; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -25,6 +26,8 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.channel.BlockingChannel; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.message.selector.MessageSelector; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; @@ -32,15 +35,12 @@ import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.Assert; /** - * This is the central class for invoking message exchange operations - * across {@link PollableSource}s and {@link MessageChannel}s. It supports - * one-way send and receive calls as well as request/reply. Additionally, - * the {@link #receiveAndForward(PollableSource, MessageChannel)} method - * plays the role of a polling-consumer while actually sending any - * received message to an event-driven consumer. - * - *

To enable transactions, configure the 'transactionManager' property - * with a reference to an instance of Spring's {@link PlatformTransactionManager} + * This is the central class for invoking message exchange operations across + * {@link MessageChannel}s. It supports one-way send and receive calls as well + * as request/reply. + *

+ * To enable transactions, configure the 'transactionManager' property with a + * reference to an instance of Spring's {@link PlatformTransactionManager} * strategy and optionally provide the other transactional attributes * (e.g. 'propagationBehaviorName'). * @@ -83,7 +83,6 @@ public class MessageExchangeTemplate implements InitializingBean { /** * Specify the timeout value to use for receive operations. - * Note that this value will only apply to {@link BlockingSource}s. * * @param receiveTimeout the receive timeout in milliseconds */ @@ -152,6 +151,18 @@ public class MessageExchangeTemplate implements InitializingBean { return this.doSend(message, channel); } + public Message receive(final PollableChannel channel) { + TransactionTemplate txTemplate = this.getTransactionTemplate(); + if (txTemplate != null) { + return (Message) txTemplate.execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + return doReceive(channel); + } + }); + } + return this.doReceive(channel); + } + public Message sendAndReceive(final Message request, final MessageChannel channel) { TransactionTemplate txTemplate = this.getTransactionTemplate(); if (txTemplate != null) { @@ -164,31 +175,6 @@ public class MessageExchangeTemplate implements InitializingBean { return this.doSendAndReceive(request, channel); } - public Message receive(final PollableSource source) { - TransactionTemplate txTemplate = this.getTransactionTemplate(); - if (txTemplate != null) { - return (Message) txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { - return doReceive(source); - } - }); - } - return this.doReceive(source); - } - - public boolean receiveAndForward(final PollableSource source, final MessageChannel channel) { - TransactionTemplate txTemplate = this.getTransactionTemplate(); - if (txTemplate != null) { - return (Boolean) txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { - return doReceiveAndForward(source, channel); - } - }); - } - return this.doReceiveAndForward(source, channel); - } - - private boolean doSend(Message message, MessageChannel channel) { Assert.notNull(channel, "channel must not be null"); long timeout = this.sendTimeout; @@ -201,14 +187,14 @@ public class MessageExchangeTemplate implements InitializingBean { return sent; } - private Message doReceive(PollableSource source) { - Assert.notNull(source, "source must not be null"); + private Message doReceive(PollableChannel channel) { + Assert.notNull(channel, "channel must not be null"); long timeout = this.receiveTimeout; - Message message = (timeout >= 0 && source instanceof BlockingSource) - ? ((BlockingSource) source).receive(timeout) - : source.receive(); + Message message = (timeout >= 0) + ? channel.receive(timeout) + : channel.receive(); if (message == null && this.logger.isTraceEnabled()) { - this.logger.trace("failed to receive message from source '" + source + "' within timeout: " + timeout); + this.logger.trace("failed to receive message from channel '" + channel + "' within timeout: " + timeout); } return message; } @@ -222,42 +208,9 @@ public class MessageExchangeTemplate implements InitializingBean { return this.doReceive(returnAddress); } - private boolean doReceiveAndForward(PollableSource source, MessageChannel channel) { - Message message = null; - try { - message = this.doReceive(source); - if (message == null) { - return false; - } - boolean sent = this.doSend(message, channel); - if (source instanceof MessageDeliveryAware) { - if (sent) { - ((MessageDeliveryAware) source).onSend(message); - } - else { - ((MessageDeliveryAware) source).onFailure(message, new MessageDeliveryException(message, "failed to send message")); - } - } - return sent; - } - catch (Exception e) { - if (source instanceof MessageDeliveryAware) { - ((MessageDeliveryAware) source).onFailure(message, e); - } - if (e instanceof MessagingException) { - throw (MessagingException) e; - } - String description = "exception occurred in receive-and-forward exchange"; - if (message != null) { - throw new MessagingException(message, description, e); - } - throw new MessagingException(description, e); - } - } - @SuppressWarnings("unchecked") - private static class TemporaryReturnAddress implements BlockingSource, MessageChannel { + private static class TemporaryReturnAddress implements PollableChannel { private volatile Message message; @@ -299,6 +252,14 @@ public class MessageExchangeTemplate implements InitializingBean { this.latch.countDown(); return true; } + + public List> clear() { + return null; + } + + public List> purge(MessageSelector selector) { + return null; + } } }