From 5afe552d82f6950854ce0a225a16041f134e40fb Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 1 Sep 2010 20:17:46 +0000 Subject: [PATCH] INT-1398 added MessagePostProcessor for use with MessagingTemplate convertAndSend methods as well as convertSendAndReceive methods --- .../core/MessagePostProcessor.java | 44 +++++++++++++ .../integration/core/MessagingOperations.java | 66 +++++++++++++++++-- .../integration/core/MessagingTemplate.java | 57 ++++++++++++++-- 3 files changed, 156 insertions(+), 11 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/core/MessagePostProcessor.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagePostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagePostProcessor.java new file mode 100644 index 0000000000..fc7d6f58a7 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagePostProcessor.java @@ -0,0 +1,44 @@ +/* + * Copyright 2002-2010 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.core; + +import org.springframework.integration.Message; + +/** + * To be used with MessagingTemplate's send method that converts an object to a message. + * It allows for further modification of the message after it has been processed + * by the converter. + * + *

This is often implemented as an anonymous class within a method implementation. + * + * @author Mark Fisher + * @since 2.0 + * @see MessagingTemplate#convertAndSend(String, Object, MessagePostProcessor) + * @see MessagingTemplate#convertAndSend(org.springframework.integration.MessageChannel, Object, MessagePostProcessor) + * @see org.springframework.integration.support.converter.MessageConverter + */ +public interface MessagePostProcessor { + + /** + * Apply a MessagePostProcessor to the message. The returned message is + * typically a modified version of the original. + * @param message the message returned from the MessageConverter + * @return the modified version of the Message + */ + Message postProcessMessage(Message message); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java index efa1c52cd8..a15183dd2b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java @@ -114,17 +114,30 @@ public interface MessagingOperations { * @param postProcessor the callback to modify the message * @throws MessagingException if an error occurs */ - //void convertAndSend(Object message, MessagePostProcessor postProcessor) throws MessagingException; + void convertAndSend(T message, MessagePostProcessor postProcessor) throws MessagingException; /** - * TODO: either define a MessagePostProcessor that accepts a builder or accept HeaderMapper here + * Send the given object to the specified channel, converting the object + * to a message with a configured MessageConverter. The MessagePostProcessor + * callback allows for modification of the message after conversion. + * @param channel the channel to which the message will be sent + * @param message the object to convert to a message + * @param postProcessor the callback to modify the message + * @throws MessagingException if an error occurs */ - //void convertAndSend(MessageChannel channel, Object message, MessagePostProcessor postProcessor) throws MessagingException; + void convertAndSend(MessageChannel channel, T message, MessagePostProcessor postProcessor) throws MessagingException; /** - * TODO: see above + * Send the given object to the specified channel, converting the object + * to a message with a configured MessageConverter. The MessagePostProcessor + * callback allows for modification of the message after conversion. + * @param channelName the name of the channel to which the message will be sent + * (to be resolved to an actual channel by a ChannelResolver) + * @param message the object to convert to a message + * @param postProcessor the callback to modify the message + * @throws MessagingException if an error occurs */ - //void convertAndSend(String channelName, Object message, MessagePostProcessor postProcessor) throws MessagingException; + void convertAndSend(String channelName, T message, MessagePostProcessor postProcessor) throws MessagingException; //------------------------------------------------------------------------- @@ -276,4 +289,47 @@ public interface MessagingOperations { */ Object convertSendAndReceive(String channelName, Object request); + /** + * Send the given request object to the default channel, converting the object + * to a message with a configured MessageConverter. The MessagePostProcessor + * callback allows for modification of the request message after conversion. + * If a reply Message is received within the receive timeout, it will be + * converted and returned. + *

This will only work with a default channel specified! + * @param request the object to convert to a request message + * @param requestPostProcessor the callback to modify the request message + * @return the result of converting the reply Message + * @throws MessagingException if an error occurs + */ + Object convertSendAndReceive(Object request, MessagePostProcessor requestPostProcessor); + + /** + * Send the given request object to the specified channel, converting the object + * to a message with a configured MessageConverter. The MessagePostProcessor + * callback allows for modification of the request message after conversion. + * If a reply Message is received within the receive timeout, it will be + * converted and returned. + * @param channel the channel to which the request message will be sent + * @param request the object to convert to a request message + * @param requestPostProcessor the callback to modify the request message + * @return the result of converting the reply Message + * @throws MessagingException if an error occurs + */ + Object convertSendAndReceive(MessageChannel channel, Object request, MessagePostProcessor requestPostProcessor); + + /** + * Send the given request object to the specified channel, converting the object + * to a message with a configured MessageConverter. The MessagePostProcessor + * callback allows for modification of the request message after conversion. + * If a reply Message is received within the receive timeout, it will be + * converted and returned. + * @param channelName the name of the channel to which the request message will be sent + * (to be resolved to an actual channel by a ChannelResolver) + * @param request the object to convert to a request message + * @param requestPostProcessor the callback to modify the request message + * @return the result of converting the reply Message + * @throws MessagingException if an error occurs + */ + Object convertSendAndReceive(String channelName, Object request, MessagePostProcessor requestPostProcessor); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java index 191f4bb98c..a3e6346881 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java @@ -244,6 +244,30 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } } + public void convertAndSend(T object, MessagePostProcessor postProcessor) { + Message message = this.messageConverter.toMessage(object); + message = postProcessor.postProcessMessage(message); + if (message != null) { + this.send(message); + } + } + + public void convertAndSend(MessageChannel channel, T object, MessagePostProcessor postProcessor) { + Message message = this.messageConverter.toMessage(object); + message = postProcessor.postProcessMessage(message); + if (message != null) { + this.send(channel, message); + } + } + + public void convertAndSend(String channelName, T object, MessagePostProcessor postProcessor) { + Message message = this.messageConverter.toMessage(object); + message = postProcessor.postProcessMessage(message); + if (message != null) { + this.send(channelName, message); + } + } + public

Message

receive() { MessageChannel channel = this.getRequiredDefaultChannel(); Assert.state(channel instanceof PollableChannel, @@ -312,15 +336,36 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } public Object convertSendAndReceive(final MessageChannel channel, final Object request) { - Message message = this.messageConverter.toMessage(request); - Message reply = this.sendAndReceive(channel, message); - return this.messageConverter.fromMessage(reply); + Message requestMessage = this.messageConverter.toMessage(request); + Message replyMessage = this.sendAndReceive(channel, requestMessage); + return this.messageConverter.fromMessage(replyMessage); } public Object convertSendAndReceive(final String channelName, final Object request) { - Message message = this.messageConverter.toMessage(request); - Message reply = this.sendAndReceive(channelName, message); - return this.messageConverter.fromMessage(reply); + Message requestMessage = this.messageConverter.toMessage(request); + Message replyMessage = this.sendAndReceive(channelName, requestMessage); + return this.messageConverter.fromMessage(replyMessage); + } + + public Object convertSendAndReceive(final Object request, MessagePostProcessor requestPostProcessor) { + Message requestMessage = this.messageConverter.toMessage(request); + requestMessage = requestPostProcessor.postProcessMessage(requestMessage); + Message replyMessage = this.sendAndReceive(requestMessage); + return this.messageConverter.fromMessage(replyMessage); + } + + public Object convertSendAndReceive(final MessageChannel channel, final Object request, MessagePostProcessor requestPostProcessor) { + Message requestMessage = this.messageConverter.toMessage(request); + requestMessage = requestPostProcessor.postProcessMessage(requestMessage); + Message replyMessage = this.sendAndReceive(channel, requestMessage); + return this.messageConverter.fromMessage(replyMessage); + } + + public Object convertSendAndReceive(final String channelName, final Object request, MessagePostProcessor requestPostProcessor) { + Message requestMessage = this.messageConverter.toMessage(request); + requestMessage = requestPostProcessor.postProcessMessage(requestMessage); + Message replyMessage = this.sendAndReceive(channelName, requestMessage); + return this.messageConverter.fromMessage(replyMessage); } private void doSend(MessageChannel channel, Message message) {