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

Added initial version of RequestReplyTemplate (INT-86).

Browse Source
This commit is contained in:
Mark Fisher
2008-02-20 15:30:41 +00:00
parent d9d15cab76
commit a07f031300
3 changed files with 238 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java Normal file
View File

@@ -0,0 +1,150 @@
/*
* Copyright 2002-2007 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.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.springframework.integration.handler.ReplyHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHeader;
import org.springframework.util.Assert;
/**
* A template that facilitates the implementation of request-reply usage
* scenarios above one-way {@link MessageChannel MessageChannels}.
*
* @author Mark Fisher
*/
public class RequestReplyTemplate {
private final MessageChannel requestChannel;
private final ExecutorService executor;
private volatile long defaultSendTimeout = -1;
private volatile long defaultReceiveTimeout = -1;
/**
* Create a RequestReplyTemplate.
*
* @param requestChannel the channel to which request messages will be sent
* @param executor the executor that will handle asynchronous (non-blocking)
* requests
*/
public RequestReplyTemplate(MessageChannel requestChannel, ExecutorService executor) {
Assert.notNull(requestChannel, "'requestChannel' must not be null");
Assert.notNull(executor, "'executor' must not be null");
this.requestChannel = requestChannel;
this.executor = executor;
}
/**
* Create a RequestReplyTemplate with a default single-threaded executor.
*
* @param requestChannel the channel to which request messages will be sent
*/
public RequestReplyTemplate(MessageChannel requestChannel) {
this(requestChannel, Executors.newSingleThreadExecutor());
}
/**
* Set the default timeout value for sending request messages. If not
* explicitly configured, the default will be an indefinite timeout.
*
* @param defaultSendTimeout the timeout value in milliseconds
*/
public void setDefaultSendTimeout(long defaultSendTimeout) {
this.defaultSendTimeout = defaultSendTimeout;
}
/**
* Set the default timeout value for receiving reply messages. If not
* explicitly configured, the default will be an indefinite timeout.
*
* @param defaultReceiveTimeout the timeout value in milliseconds
*/
public void setDefaultReceiveTimeout(long defaultReceiveTimeout) {
this.defaultReceiveTimeout = defaultReceiveTimeout;
}
/**
* Send a request message and wait for a reply message using the provided
* timeout values.
*
* @param requestMessage the request message to send
* @param sendTimeout the timeout value for sending the request message
* @param receiveTimeout the timeout value for receiving a reply message
*
* @return the reply message or <code>null</code>
*/
public Message<?> request(Message<?> requestMessage, long sendTimeout, long receiveTimeout) {
SimpleChannel replyChannel = new SimpleChannel(0);
requestMessage.getHeader().setReturnAddress(replyChannel);
this.requestChannel.send(requestMessage, sendTimeout);
return replyChannel.receive(receiveTimeout);
}
/**
* Send a request message and wait for a reply message using the default
* timeout values.
*
* @param requestMessage the request message to send
*
* @return the reply message or <code>null</code>
*/
public Message<?> request(Message<?> requestMessage) {
return this.request(requestMessage, this.defaultSendTimeout, this.defaultReceiveTimeout);
}
/**
* Send a request message asynchronously to be handled by the provided
* {@link ReplyHandler}. The provided values will be used for the send
* and receive timeouts.
*
* @param requestMessage the request message to send
* @param replyHandler the callback that will handle a reply message
* @param sendTimeout the timeout value for sending the request message
* @param receiveTimeout the timeout value for receiving the reply message
*/
public void request(final Message<?> requestMessage, final ReplyHandler replyHandler, final long sendTimeout,
final long receiveTimeout) {
final MessageHeader header = requestMessage.getHeader();
this.executor.submit(new Runnable() {
public void run() {
Message<?> reply = request(requestMessage, sendTimeout, receiveTimeout);
replyHandler.handle(reply, header);
}
});
}
/**
* Send a request message asynchronously to be handled by the provided
* {@link ReplyHandler}. Default values will be used for the send and
* receive timeouts.
*
* @param requestMessage the request message to send
* @param replyHandler the callback that will handle a reply message
*/
public void request(final Message<?> requestMessage, final ReplyHandler replyHandler) {
this.request(requestMessage, replyHandler, this.defaultSendTimeout, this.defaultReceiveTimeout);
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java
View File

@@ -77,7 +77,7 @@ public class SimpleChannel extends AbstractMessageChannel {
}
public boolean doSend(Message message, long timeout) {
protected boolean doSend(Message message, long timeout) {
Assert.notNull(message, "'message' must not be null");
try {
if (timeout > 0) {
@@ -95,7 +95,7 @@ public class SimpleChannel extends AbstractMessageChannel {
}
}
public Message doReceive(long timeout) {
protected Message doReceive(long timeout) {
try {
if (timeout > 0) {
return queue.poll(timeout, TimeUnit.MILLISECONDS);

86
spring-integration-core/src/test/java/org/springframework/integration/channel/RequestReplyTemplateTests.java Normal file
View File

@@ -0,0 +1,86 @@
/*
* Copyright 2002-2007 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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.ReplyHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHeader;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.Subscription;
/**
* @author Mark Fisher
*/
public class RequestReplyTemplateTests {
private final SimpleChannel requestChannel = new SimpleChannel();
public RequestReplyTemplateTests() {
MessageHandler testHandler = new MessageHandler() {
public Message<?> handle(Message<?> message) {
return new StringMessage(message.getPayload().toString().toUpperCase());
}
};
MessageBus bus = new MessageBus();
bus.registerChannel("requestChannel", requestChannel);
bus.registerHandler("testHandler", testHandler, new Subscription(requestChannel));
bus.start();
}
@Test
public void testSynchronousRequestReply() {
RequestReplyTemplate template = new RequestReplyTemplate(requestChannel);
Message<?> reply = template.request(new StringMessage("test"));
assertEquals("TEST", reply.getPayload());
}
@Test
public void testAsynchronousRequestAndReply() throws InterruptedException {
final List<String> replies = new ArrayList<String>(3);
final CountDownLatch latch = new CountDownLatch(3);
ReplyHandler replyHandler = new ReplyHandler() {
public void handle(Message<?> replyMessage, MessageHeader originalMessageHeader) {
replies.add((String) replyMessage.getPayload());
latch.countDown();
}
};
RequestReplyTemplate template = new RequestReplyTemplate(requestChannel);
template.request(new StringMessage("test1"), replyHandler);
template.request(new StringMessage("test2"), replyHandler);
template.request(new StringMessage("test3"), replyHandler);
latch.await(1000, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
assertTrue(replies.contains("TEST1"));
assertTrue(replies.contains("TEST2"));
assertTrue(replies.contains("TEST3"));
}
}