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

INT-806: pulled up resolveReplyChannel and made it final

Browse Source
This commit is contained in:
Iwein Fuld
2009-09-26 18:41:17 +00:00
parent 7f41a1b89f
commit f01125a516
3 changed files with 59 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java
View File

@@ -112,7 +112,7 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L
if (tracker.aquireLockFor(correlationKey)) {
store(message, correlationKey);
List<Message<?>> all = store.getAll(correlationKey);
complete(correlationKey, all, this.resolveReplyChannel(message));
complete(correlationKey, all, this.resolveReplyChannel(message, this.outputChannel, this.channelResolver));
} else {
discardChannel.send(message);
}
@@ -151,31 +151,6 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L
}
}
//TODO copied from AbstractReplyProducingMessageHandler
private MessageChannel resolveReplyChannel(Message<?> requestMessage) {
MessageChannel replyChannel = outputChannel;
if (replyChannel == null) {
Object replyChannelHeader = requestMessage.getHeaders().getReplyChannel();
if (replyChannelHeader != null) {
if (replyChannelHeader instanceof MessageChannel) {
replyChannel = (MessageChannel) replyChannelHeader;
} else if (replyChannelHeader instanceof String) {
Assert.state(this.channelResolver != null,
"ChannelResolver is required for resolving a reply channel by name");
replyChannel = this.channelResolver.resolveChannelName((String) replyChannelHeader);
} else {
throw new ChannelResolutionException("expected a MessageChannel or String for 'replyChannel', but type is ["
+ replyChannelHeader.getClass() + "]");
}
}
}
if (replyChannel == null) {
throw new ChannelResolutionException(
"unable to resolve reply channel for message: " + requestMessage);
}
return replyChannel;
}
public boolean isRunning() {
synchronized (this.lifecycleMonitor) {
return this.reaperFutureTask != null;
@@ -226,7 +201,7 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L
List<Message<?>> all = store.getAll(key);
if (all.size() > 0) {
//last chance for normal completion
MessageChannel outputChannel = resolveReplyChannel(all.get(0));
MessageChannel outputChannel = resolveReplyChannel(all.get(0), this.outputChannel, this.channelResolver);
boolean fullyCompleted = complete(key, all, outputChannel);
if (!fullyCompleted) {
if (sendPartialResultOnTimeout) {

84
org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java
View File

@@ -22,51 +22,79 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.core.Ordered;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.channel.ChannelResolutionException;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.util.Assert;
/**
* Base class for MessageHandler implementations that provides basic
* validation and error handling capabilities. Asserts that the incoming
* Message is not null and that it does not contain a null payload. Converts
* checked exceptions into runtime {@link MessagingException}s.
*
* checked exceptions into runtime {@link MessagingException}s.
*
* @author Mark Fisher
*/
public abstract class AbstractMessageHandler implements MessageHandler, Ordered {
protected final Log logger = LogFactory.getLog(this.getClass());
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile int order = Ordered.LOWEST_PRECEDENCE;
private volatile int order = Ordered.LOWEST_PRECEDENCE;
public void setOrder(int order) {
this.order = order;
}
public void setOrder(int order) {
this.order = order;
}
public int getOrder() {
return this.order;
}
public int getOrder() {
return this.order;
}
public final void handleMessage(Message<?> message) {
Assert.notNull(message == null, "Message must not be null");
Assert.notNull(message.getPayload(), "Message payload must not be null");
if (this.logger.isDebugEnabled()) {
this.logger.debug(this + " received message: " + message);
}
try {
this.handleMessageInternal(message);
}
catch (Exception e) {
if (e instanceof MessagingException) {
throw (MessagingException) e;
}
throw new MessageHandlingException(message,
"error occurred in message handler [" + this + "]", e);
}
}
public final void handleMessage(Message<?> message) {
Assert.notNull(message == null, "Message must not be null");
Assert.notNull(message.getPayload(), "Message payload must not be null");
if (this.logger.isDebugEnabled()) {
this.logger.debug(this + " received message: " + message);
}
try {
this.handleMessageInternal(message);
}
catch (Exception e) {
if (e instanceof MessagingException) {
throw (MessagingException) e;
}
throw new MessageHandlingException(message,
"error occurred in message handler [" + this + "]", e);
}
}
protected abstract void handleMessageInternal(Message<?> message) throws Exception;
protected abstract void handleMessageInternal(Message<?> message) throws Exception;
protected final MessageChannel resolveReplyChannel(Message<?> requestMessage,
MessageChannel defaultOutputChannel,
ChannelResolver channelResolver) {
MessageChannel replyChannel = defaultOutputChannel;
if (replyChannel == null) {
Object replyChannelHeader = requestMessage.getHeaders().getReplyChannel();
if (replyChannelHeader != null) {
if (replyChannelHeader instanceof MessageChannel) {
replyChannel = (MessageChannel) replyChannelHeader;
} else if (replyChannelHeader instanceof String) {
Assert.state(channelResolver != null,
"ChannelResolver is required for resolving a reply channel by name");
replyChannel = channelResolver.resolveChannelName((String) replyChannelHeader);
} else {
throw new ChannelResolutionException("expected a MessageChannel or String for 'replyChannel', but type is ["
+ replyChannelHeader.getClass() + "]");
}
}
}
if (replyChannel == null) {
throw new ChannelResolutionException(
"unable to resolve reply channel for message: " + requestMessage);
}
return replyChannel;
}
}

28
org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
View File

@@ -99,7 +99,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
}
return;
}
MessageChannel replyChannel = this.resolveReplyChannel(message);
MessageChannel replyChannel = resolveReplyChannel(message, this.outputChannel, this.channelResolver);
MessageHeaders requestHeaders = message.getHeaders();
for (MessageBuilder<?> builder : replyMessageHolder.builders()) {
builder.copyHeadersIfAbsent(requestHeaders);
@@ -119,30 +119,4 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
return this.channelTemplate.send(replyMessage, replyChannel);
}
private MessageChannel resolveReplyChannel(Message<?> requestMessage) {
MessageChannel replyChannel = this.getOutputChannel();
if (replyChannel == null) {
Object replyChannelHeader= requestMessage.getHeaders().getReplyChannel();
if (replyChannelHeader != null) {
if (replyChannelHeader instanceof MessageChannel) {
replyChannel = (MessageChannel) replyChannelHeader;
}
else if (replyChannelHeader instanceof String) {
Assert.state(this.channelResolver != null,
"ChannelResolver is required for resolving a reply channel by name");
replyChannel = this.channelResolver.resolveChannelName((String) replyChannelHeader);
}
else {
throw new ChannelResolutionException("expected a MessageChannel or String for 'replyChannel', but type is ["
+ replyChannelHeader.getClass() + "]");
}
}
}
if (replyChannel == null) {
throw new ChannelResolutionException(
"unable to resolve reply channel for message: " + requestMessage);
}
return replyChannel;
}
}