From f01125a5163f60323f037f0d7d8de910e3567570 Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Sat, 26 Sep 2009 18:41:17 +0000 Subject: [PATCH] INT-806: pulled up resolveReplyChannel and made it final --- .../aggregator/BufferingMessageHandler.java | 29 +------ .../handler/AbstractMessageHandler.java | 84 ++++++++++++------- .../AbstractReplyProducingMessageHandler.java | 28 +------ 3 files changed, 59 insertions(+), 82 deletions(-) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java index c93ea9fd9e..e4672bce70 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java @@ -112,7 +112,7 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L if (tracker.aquireLockFor(correlationKey)) { store(message, correlationKey); List> 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> 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) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index 754d19c223..6110efc1b8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -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; + } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 232a04688a..91e739df27 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -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; - } - }