From f45437801301d2b1aa285d66732f77aca75c30a5 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 13 Sep 2010 13:10:53 -0400 Subject: [PATCH] INT-1412 added async send methods to AsyncMessagingOperations and AsyncMessagingTemplate --- .../core/AsyncMessagingOperations.java | 12 +++ .../core/AsyncMessagingTemplate.java | 48 ++++++++++ .../core/AsyncMessagingTemplateTests.java | 92 +++++++++++++++++++ 3 files changed, 152 insertions(+) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingOperations.java b/spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingOperations.java index 7b29161f94..4d735344be 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingOperations.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingOperations.java @@ -27,6 +27,18 @@ import org.springframework.integration.MessageChannel; */ public interface AsyncMessagingOperations { + Future asyncSend(Message message); + + Future asyncSend(MessageChannel channel, Message message); + + Future asyncSend(String channelName, Message message); + + Future asyncConvertAndSend(Object message); + + Future asyncConvertAndSend(MessageChannel channel, Object message); + + Future asyncConvertAndSend(String channelName, Object message); + Future> asyncReceive(); Future> asyncReceive(PollableChannel channel); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingTemplate.java index 12fc4fb0cc..6118cea71d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingTemplate.java @@ -42,6 +42,54 @@ public class AsyncMessagingTemplate extends MessagingTemplate implements AsyncMe (AsyncTaskExecutor) executor : new TaskExecutorAdapter(executor); } + public Future asyncSend(final Message message) { + return this.executor.submit(new Runnable() { + public void run() { + send(message); + } + }); + } + + public Future asyncSend(final MessageChannel channel, final Message message) { + return this.executor.submit(new Runnable() { + public void run() { + send(channel, message); + } + }); + } + + public Future asyncSend(final String channelName, final Message message) { + return this.executor.submit(new Runnable() { + public void run() { + send(channelName, message); + } + }); + } + + public Future asyncConvertAndSend(final Object object) { + return this.executor.submit(new Runnable() { + public void run() { + convertAndSend(object); + } + }); + } + + public Future asyncConvertAndSend(final MessageChannel channel, final Object object) { + return this.executor.submit(new Runnable() { + public void run() { + convertAndSend(channel, object); + } + }); + } + + public Future asyncConvertAndSend(final String channelName, final Object object) { + return this.executor.submit(new Runnable() { + public void run() { + convertAndSend(channelName, object); + } + }); + } + public Future> asyncReceive() { return this.executor.submit(new Callable>() { public Message call() throws Exception { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java index 5458cbcbed..cb31007ac5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.core; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -47,6 +48,97 @@ import org.springframework.util.Assert; */ public class AsyncMessagingTemplateTests { + @Test + public void asyncSendWithDefaultChannel() throws Exception { + QueueChannel channel = new QueueChannel(); + AsyncMessagingTemplate template = new AsyncMessagingTemplate(); + template.setDefaultChannel(channel); + Message message = MessageBuilder.withPayload("test").build(); + Future future = template.asyncSend(message); + assertNull(future.get(1000, TimeUnit.MILLISECONDS)); + Message result = channel.receive(0); + assertEquals(message, result); + } + + @Test + public void asyncSendWithExplicitChannel() throws Exception { + QueueChannel channel = new QueueChannel(); + AsyncMessagingTemplate template = new AsyncMessagingTemplate(); + Message message = MessageBuilder.withPayload("test").build(); + Future future = template.asyncSend(channel, message); + assertNull(future.get(1000, TimeUnit.MILLISECONDS)); + Message result = channel.receive(0); + assertEquals(message, result); + } + + @Test + public void asyncSendWithResolvedChannel() 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); + Message message = MessageBuilder.withPayload("test").build(); + Future future = template.asyncSend("testChannel", message); + assertNull(future.get(1000, TimeUnit.MILLISECONDS)); + Message result = channel.receive(0); + assertEquals(message, result); + } + + @Test(expected = TimeoutException.class) + public void asyncSendWithTimeoutException() throws Exception { + QueueChannel channel = new QueueChannel(1); + channel.send(MessageBuilder.withPayload("blocker").build()); + AsyncMessagingTemplate template = new AsyncMessagingTemplate(); + Future result = template.asyncSend(channel, MessageBuilder.withPayload("test").build()); + result.get(100, TimeUnit.MILLISECONDS); + } + + @Test + public void asyncConvertAndSendWithDefaultChannel() throws Exception { + QueueChannel channel = new QueueChannel(); + AsyncMessagingTemplate template = new AsyncMessagingTemplate(); + template.setDefaultChannel(channel); + Future future = template.asyncConvertAndSend("test"); + assertNull(future.get(1000, TimeUnit.MILLISECONDS)); + Message result = channel.receive(0); + assertEquals("test", result.getPayload()); + } + + @Test + public void asyncConvertAndSendWithExplicitChannel() throws Exception { + QueueChannel channel = new QueueChannel(); + AsyncMessagingTemplate template = new AsyncMessagingTemplate(); + Future future = template.asyncConvertAndSend(channel, "test"); + assertNull(future.get(1000, TimeUnit.MILLISECONDS)); + Message result = channel.receive(0); + assertEquals("test", result.getPayload()); + } + + @Test + public void asyncConvertAndSendWithResolvedChannel() 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 future = template.asyncConvertAndSend("testChannel", "test"); + assertNull(future.get(1000, TimeUnit.MILLISECONDS)); + Message result = channel.receive(0); + assertEquals("test", result.getPayload()); + } + + @Test(expected = TimeoutException.class) + public void asyncConvertAndSendWithTimeoutException() throws Exception { + QueueChannel channel = new QueueChannel(1); + channel.send(MessageBuilder.withPayload("blocker").build()); + AsyncMessagingTemplate template = new AsyncMessagingTemplate(); + Future result = template.asyncConvertAndSend(channel, "test"); + result.get(100, TimeUnit.MILLISECONDS); + } + @Test public void asyncReceiveWithDefaultChannel() throws Exception { QueueChannel channel = new QueueChannel();