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

INT-1411 added convertSendAndReceive methods that accept MessagePostProcessors for the requests to AsyncMessagingTemplate and AsyncMessagingOperations

Browse Source
This commit is contained in:
Mark Fisher
2010-09-02 23:39:01 +00:00
parent eb341b234a
commit c8b9c618d5
2 changed files with 84 additions and 51 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

@@ -51,4 +51,10 @@ public interface AsyncMessagingOperations {
<R> Future<R> asyncConvertSendAndReceive(String channelName, Object request);
<R> Future<R> asyncConvertSendAndReceive(Object request, MessagePostProcessor requestPostProcessor);
<R> Future<R> asyncConvertSendAndReceive(MessageChannel channel, Object request, MessagePostProcessor requestPostProcessor);
<R> Future<R> asyncConvertSendAndReceive(String channelName, Object request, MessagePostProcessor requestPostProcessor);
}

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

@@ -42,57 +42,6 @@ public class AsyncMessagingTemplate extends MessagingTemplate implements AsyncMe
(AsyncTaskExecutor) executor : new TaskExecutorAdapter(executor);
}
public Future<Message<?>> asyncSendAndReceive(final Message<?> requestMessage) {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return sendAndReceive(requestMessage);
}
});
}
public Future<Message<?>> asyncSendAndReceive(final MessageChannel channel, final Message<?> requestMessage) {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return sendAndReceive(channel, requestMessage);
}
});
}
public Future<Message<?>> asyncSendAndReceive(final String channelName, final Message<?> requestMessage) {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return sendAndReceive(channelName, requestMessage);
}
});
}
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final Object request) {
return this.executor.submit(new Callable<R>() {
public R call() throws Exception {
return (R) convertSendAndReceive(request);
}
});
}
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final MessageChannel channel, final Object request) {
return this.executor.submit(new Callable<R>() {
public R call() throws Exception {
return (R) convertSendAndReceive(channel, request);
}
});
}
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final String channelName, final Object request) {
return this.executor.submit(new Callable<R>() {
public R call() throws Exception {
return (R) convertSendAndReceive(channelName, request);
}
});
}
public <P> Future<Message<P>> asyncReceive() {
return this.executor.submit(new Callable<Message<P>>() {
public Message<P> call() throws Exception {
@@ -144,4 +93,82 @@ public class AsyncMessagingTemplate extends MessagingTemplate implements AsyncMe
});
}
public Future<Message<?>> asyncSendAndReceive(final Message<?> requestMessage) {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return sendAndReceive(requestMessage);
}
});
}
public Future<Message<?>> asyncSendAndReceive(final MessageChannel channel, final Message<?> requestMessage) {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return sendAndReceive(channel, requestMessage);
}
});
}
public Future<Message<?>> asyncSendAndReceive(final String channelName, final Message<?> requestMessage) {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return sendAndReceive(channelName, requestMessage);
}
});
}
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final Object request) {
return this.executor.submit(new Callable<R>() {
public R call() throws Exception {
return (R) convertSendAndReceive(request);
}
});
}
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final MessageChannel channel, final Object request) {
return this.executor.submit(new Callable<R>() {
public R call() throws Exception {
return (R) convertSendAndReceive(channel, request);
}
});
}
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final String channelName, final Object request) {
return this.executor.submit(new Callable<R>() {
public R call() throws Exception {
return (R) convertSendAndReceive(channelName, request);
}
});
}
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final Object request, final MessagePostProcessor requestPostProcessor) {
return this.executor.submit(new Callable<R>() {
public R call() throws Exception {
return (R) convertSendAndReceive(request, requestPostProcessor);
}
});
}
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final MessageChannel channel, final Object request, final MessagePostProcessor requestPostProcessor) {
return this.executor.submit(new Callable<R>() {
public R call() throws Exception {
return (R) convertSendAndReceive(channel, request, requestPostProcessor);
}
});
}
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final String channelName, final Object request, final MessagePostProcessor requestPostProcessor) {
return this.executor.submit(new Callable<R>() {
public R call() throws Exception {
return (R) convertSendAndReceive(channelName, request, requestPostProcessor);
}
});
}
}