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

INT-1398 added MessagePostProcessor for use with MessagingTemplate convertAndSend methods as well as convertSendAndReceive methods

Browse Source
This commit is contained in:
Mark Fisher
2010-09-01 20:17:46 +00:00
parent cd5ca10ae2
commit 5afe552d82
3 changed files with 156 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
spring-integration-core/src/main/java/org/springframework/integration/core/MessagePostProcessor.java Normal file
View File

@@ -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.
*
* <p>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);
}

66
spring-integration-core/src/main/java/org/springframework/integration/core/MessagingOperations.java
View File

@@ -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;
<T> 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;
<T> 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;
<T> 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.
* <p>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);
}

57
spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java
View File

@@ -244,6 +244,30 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware,
}
}
public <T> void convertAndSend(T object, MessagePostProcessor postProcessor) {
Message<?> message = this.messageConverter.toMessage(object);
message = postProcessor.postProcessMessage(message);
if (message != null) {
this.send(message);
}
}
public <T> 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 <T> 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 <P> Message<P> 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) {