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

MessageExchangeTemplate uses channels in all operations (rather than PollableSource) and no longer provides a receiveAndForward method.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-09 08:54:07 +00:00
parent 722f143244
commit ab2abb33d2
5 changed files with 61 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java
View File

@@ -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 <code>null</code> 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 <code>null</code> if the
* specified timeout period elapses or the message reception is interrupted
*/
Message<?> receive(long timeout);
/**
* Remove all {@link Message Messages} from this channel.

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

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

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

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

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

@@ -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<Message<?>> task = new FutureTask<Message<?>>(new Callable<Message<?>>() {
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 <code>null</code>,
* send it to the given channel. Note that the receive and send operations
* occur asynchronously, so this method will always return <code>true</code>
* 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;
}
}

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

@@ -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.
*
* <p>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.
* <p>
* 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<Message<?>> clear() {
return null;
}
public List<Message<?>> purge(MessageSelector selector) {
return null;
}
}
}