From 9a1714fae4859f44d422a5e365355c4fb9cb6c29 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sat, 9 Aug 2008 00:26:41 +0000 Subject: [PATCH] Added new SimpleEndpoint implementation. This is the first step of a gradual migration process (i.e. it is not yet in use). Also added the MessageEndpointReplyException. --- .../MessageEndpointReplyException.java | 58 ++++ .../integration/endpoint/SimpleEndpoint.java | 248 ++++++++++++++++++ 2 files changed, 306 insertions(+) create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpointReplyException.java create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SimpleEndpoint.java diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpointReplyException.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpointReplyException.java new file mode 100644 index 0000000000..7341447778 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpointReplyException.java @@ -0,0 +1,58 @@ +/* + * Copyright 2002-2008 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.endpoint; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageDeliveryException; + +/** + * An Exception that indicates a failure occurred while a Message Endpoint + * was attempting to send a reply message. The Exception provides access to + * both the reply message and the original request message. + * + * @author Mark Fisher + */ +public class MessageEndpointReplyException extends MessageDeliveryException { + + private final Message requestMessage; + + + public MessageEndpointReplyException(Message undeliveredMessage, Message requestMessage) { + super(undeliveredMessage); + this.requestMessage = requestMessage; + } + + public MessageEndpointReplyException(Message undeliveredMessage, Message requestMessage, String description) { + super(undeliveredMessage, description); + this.requestMessage = requestMessage; + } + + public MessageEndpointReplyException(Message undeliveredMessage, Message requestMessage, String description, Throwable cause) { + super(undeliveredMessage, description, cause); + this.requestMessage = requestMessage; + } + + + public Message getOriginalRequestMessage() { + return this.requestMessage; + } + + public Message getUndeliveredReplyMessage() { + return this.getFailedMessage(); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SimpleEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SimpleEndpoint.java new file mode 100644 index 0000000000..0ef0bdca42 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SimpleEndpoint.java @@ -0,0 +1,248 @@ +/* + * Copyright 2002-2008 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.endpoint; + +import java.lang.reflect.Array; +import java.util.Collection; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.CompositeMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageExchangeTemplate; +import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.message.MessageHeaders; +import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.MessagingException; +import org.springframework.integration.util.ErrorHandler; +import org.springframework.util.Assert; + +/** + * The most basic Message Endpoint implementation. Serves as a "host" for any + * {@link MessageHandler} and resolves the target for any reply Message(s) + * returned by that handler. If the handler returns a non-empty + * {@link CompositeMessage}, each Message in its list will be sent + * as a separate reply Message. + * + *

The reply target is resolved according to the following order: + *

    + *
  1. the 'nextTarget' header value of the reply Message
    2. + *
  2. the 'outputChannel' of this Message Endpoint
    3. + *
  3. the 'returnAddress' header value of the request Message
    4. + *
+ * For the 'nextTarget' and 'returnAddress' values, either a + * {@link MessageTarget} instance or String is accepted. If the + * value is a String, then the endpoint will consult its + * {@link ChannelRegistry} (typically provided by the MessageBus). + * If no reply target can be determined for a non-null reply Message, + * a {@link MessageEndpointReplyException} will be thrown. + * + * @author Mark Fisher + */ +public class SimpleEndpoint implements MessageTarget, BeanNameAware { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private volatile String name; + + private final T handler; + + private volatile MessageChannel outputChannel; + + private volatile ErrorHandler errorHandler; + + private volatile ChannelRegistry channelRegistry; + + private volatile boolean requiresReply = false; + + private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate(); + + + /** + * Create an endpoint for the given handler. + */ + public SimpleEndpoint(T handler) { + Assert.notNull(handler, "handler must not be null"); + this.handler = handler; + } + + + public void setBeanName(String name) { + this.name = name; + } + + /** + * Return the name of this endpoint. + */ + public String getName() { + return this.name; + } + + /** + * Specify the channel where reply Messages should be sent if + * no 'nextTarget' header value is available on the reply Message. + */ + public void setOutputChannel(MessageChannel outputChannel) { + this.outputChannel = outputChannel; + } + + protected T getHandler() { + return this.handler; + } + + /** + * Provide an error handler for any Exceptions that occur + * upon invocation of this endpoint. If none is provided, + * the Exception messages will be logged (at warn level), + * and the Exception rethrown. + */ + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + + public void setChannelRegistry(ChannelRegistry channelRegistry) { + this.channelRegistry = channelRegistry; + } + + protected ChannelRegistry getChannelRegistry() { + return this.channelRegistry; + } + + /** + * Specify whether this endpoint should throw an Exception when + * its handler returns an invalid reply Message. The reply Message + * is considered invalid if it is null, has a null payload, or has + * an empty Array or Collection as its payload. + */ + public void setRequiresReply(boolean requiresReply) { + this.requiresReply = requiresReply; + } + + /** + * Specify the timeout for sending reply Messages to the reply + * target. The default value indicates an indefinite timeout. + */ + public void setReplyTimeout(long replyTimeout) { + this.messageExchangeTemplate.setSendTimeout(replyTimeout); + } + + public boolean send(Message requestMessage) { + try { + Message replyMessage = this.handler.handle(requestMessage); + if (!this.isValidReply(replyMessage)) { + if (this.requiresReply) { + throw new MessageHandlingException(requestMessage, + "endpoint requires reply but none was received"); + } + } + else if (replyMessage instanceof CompositeMessage) { + for (Message nextReply : (CompositeMessage) replyMessage) { + this.sendReplyMessage(nextReply, requestMessage); + } + } + else { + this.sendReplyMessage(replyMessage, requestMessage); + } + return true; + } + catch (Exception e) { + if (e instanceof MessagingException) { + this.handleException((MessagingException) e); + } + else { + this.handleException(new MessageHandlingException(requestMessage, + "failure occurred in endpoint's send operation", e)); + } + return false; + } + } + + private void sendReplyMessage(Message replyMessage, Message requestMessage) { + if (replyMessage == null) { + throw new MessageHandlingException(requestMessage, "reply message must not be null"); + } + MessageTarget replyTarget = this.resolveReplyTarget(replyMessage, requestMessage.getHeaders()); + if (replyTarget == null) { + throw new MessageEndpointReplyException(replyMessage, requestMessage, + "unable to resolve reply target"); + } + if (!this.messageExchangeTemplate.send(replyMessage, replyTarget)) { + throw new MessageEndpointReplyException(replyMessage, requestMessage, + "failed to send reply to '" + replyTarget + "'"); + } + } + + private void handleException(MessagingException exception) { + if (this.errorHandler == null) { + if (this.logger.isWarnEnabled()) { + this.logger.warn("exception occurred in endpoint '" + this.name + "'", exception); + } + throw exception; + } + this.errorHandler.handle(exception); + } + + private boolean isValidReply(Message replyMessage) { + if (replyMessage == null) { + return false; + } + Object payload = replyMessage.getPayload(); + if (payload == null) { + return false; + } + if (payload.getClass().isArray() && Array.getLength(payload) == 0) { + return false; + } + if (payload instanceof Collection && ((Collection) payload).size() == 0) { + return false; + } + return true; + } + + private MessageTarget resolveReplyTarget(Message replyMessage, MessageHeaders requestHeaders) { + MessageTarget replyTarget = this.resolveTargetAttribute(replyMessage.getHeaders().getNextTarget()); + if (replyTarget == null) { + replyTarget = this.outputChannel; + } + if (replyTarget == null) { + replyTarget = this.resolveTargetAttribute(requestHeaders.getReturnAddress()); + } + return replyTarget; + } + + private MessageTarget resolveTargetAttribute(Object targetAttribute) { + MessageTarget replyTarget = null; + if (targetAttribute != null) { + if (targetAttribute instanceof MessageTarget) { + replyTarget = (MessageTarget) targetAttribute; + } + else if (targetAttribute instanceof String) { + ChannelRegistry registry = getChannelRegistry(); + if (registry != null) { + replyTarget = registry.lookupChannel((String) targetAttribute); + } + } + } + return replyTarget; + } + +}