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

INT-1411 added tests for asyncReceive operations

Browse Source
This commit is contained in:
Mark Fisher
2010-09-03 00:37:43 +00:00
parent e600243ea8
commit 9e4a84257b
3 changed files with 82 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingOperations.java
View File

@@ -27,11 +27,11 @@ import org.springframework.integration.MessageChannel;
*/
public interface AsyncMessagingOperations {
<P> Future<Message<P>> asyncReceive();
Future<Message<?>> asyncReceive();
<P> Future<Message<P>> asyncReceive(PollableChannel channel);
Future<Message<?>> asyncReceive(PollableChannel channel);
<P> Future<Message<P>> asyncReceive(String channelName);
Future<Message<?>> asyncReceive(String channelName);
<R> Future<R> asyncReceiveAndConvert();

18
spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingTemplate.java
View File

@@ -42,25 +42,25 @@ public class AsyncMessagingTemplate extends MessagingTemplate implements AsyncMe
(AsyncTaskExecutor) executor : new TaskExecutorAdapter(executor);
}
public <P> Future<Message<P>> asyncReceive() {
return this.executor.submit(new Callable<Message<P>>() {
public Message<P> call() throws Exception {
public Future<Message<?>> asyncReceive() {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return receive();
}
});
}
public <P> Future<Message<P>> asyncReceive(final PollableChannel channel) {
return this.executor.submit(new Callable<Message<P>>() {
public Message<P> call() throws Exception {
public Future<Message<?>> asyncReceive(final PollableChannel channel) {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return receive(channel);
}
});
}
public <P> Future<Message<P>> asyncReceive(final String channelName) {
return this.executor.submit(new Callable<Message<P>>() {
public Message<P> call() throws Exception {
public Future<Message<?>> asyncReceive(final String channelName) {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return receive(channelName);
}
});

70
spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java
View File

@@ -23,6 +23,7 @@ import static org.junit.Assert.fail;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -31,9 +32,12 @@ import org.junit.Test;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
@@ -43,6 +47,57 @@ import org.springframework.util.Assert;
*/
public class AsyncMessagingTemplateTests {
@Test
public void asyncReceiveWithDefaultChannel() throws Exception {
QueueChannel channel = new QueueChannel();
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
template.setDefaultChannel(channel);
Future<Message<?>> result = template.asyncReceive();
sendMessageAfterDelay(channel, new GenericMessage<String>("test"), 200);
long start = System.currentTimeMillis();
assertNotNull(result.get(1000, TimeUnit.MILLISECONDS));
long elapsed = System.currentTimeMillis() - start;
assertEquals("test", result.get().getPayload());
assertTrue(elapsed >= 200);
}
@Test
public void asyncReceiveWithExplicitChannel() throws Exception {
QueueChannel channel = new QueueChannel();
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
Future<Message<?>> result = template.asyncReceive(channel);
sendMessageAfterDelay(channel, new GenericMessage<String>("test"), 200);
long start = System.currentTimeMillis();
assertNotNull(result.get(1000, TimeUnit.MILLISECONDS));
long elapsed = System.currentTimeMillis() - start;
assertEquals("test", result.get().getPayload());
assertTrue(elapsed >= 200);
}
@Test
public void asyncReceiveWithResolvedChannel() throws Exception {
StaticApplicationContext context = new StaticApplicationContext();
context.registerSingleton("testChannel", QueueChannel.class);
context.refresh();
QueueChannel channel = context.getBean("testChannel", QueueChannel.class);
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
template.setBeanFactory(context);
Future<Message<?>> result = template.asyncReceive("testChannel");
sendMessageAfterDelay(channel, new GenericMessage<String>("test"), 200);
long start = System.currentTimeMillis();
assertNotNull(result.get(1000, TimeUnit.MILLISECONDS));
long elapsed = System.currentTimeMillis() - start;
assertTrue(elapsed >= 200);
assertEquals("test", result.get().getPayload());
}
@Test(expected = TimeoutException.class)
public void asyncReceiveWithTimeoutException() throws Exception {
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
Future<Message<?>> result = template.asyncReceive(new QueueChannel());
result.get(100, TimeUnit.MILLISECONDS);
}
@Test
public void asyncSendAndReceiveWithDefaultChannel() throws Exception {
DirectChannel channel = new DirectChannel();
@@ -177,6 +232,21 @@ public class AsyncMessagingTemplateTests {
}
private static void sendMessageAfterDelay(final MessageChannel channel, final GenericMessage<String> message, final int delay) {
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
Thread.sleep(delay);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
channel.send(message);
}
});
}
private static class EchoHandler extends AbstractReplyProducingMessageHandler {
private final long delay;