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

INT-1412 added async send methods to AsyncMessagingOperations and AsyncMessagingTemplate

Browse Source
This commit is contained in:
Mark Fisher
2010-09-13 13:10:53 -04:00
parent 7fc8de4a78
commit f454378013
3 changed files with 152 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Message<?>> asyncReceive();
Future<Message<?>> asyncReceive(PollableChannel channel);

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

@@ -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<Message<?>> asyncReceive() {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {

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

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