From e16ef8f95b9d926ecf7d1153c81a087f6c7c45d9 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sun, 2 Dec 2007 18:39:49 +0000 Subject: [PATCH] Added DefaultMessageEndpoint --- .../endpoint/DefaultMessageEndpoint.java | 161 ++++++++++++++++++ 1 file changed, 161 insertions(+) create mode 100644 spring-eai-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java new file mode 100644 index 0000000000..60308c2116 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -0,0 +1,161 @@ +/* + * Copyright 2002-2007 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.MessageHandlingException; +import org.springframework.integration.MessageSource; +import org.springframework.integration.MessageTarget; +import org.springframework.integration.channel.ChannelResolver; +import org.springframework.integration.channel.consumer.AbstractConsumer; +import org.springframework.integration.channel.consumer.ConsumerType; +import org.springframework.integration.channel.consumer.EventDrivenConsumer; +import org.springframework.integration.channel.consumer.FixedDelayConsumer; +import org.springframework.integration.channel.consumer.FixedRateConsumer; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; + +/** + * A generic endpoint implementation designed to accommodate a variety of + * strategies including: + * + * + * @author Mark Fisher + */ +public class DefaultMessageEndpoint { + + private MessageSource source; + + private MessageTarget target; + + private MessageHandler handler; + + private ConsumerType consumerType = ConsumerType.EVENT_DRIVEN; + + private AbstractConsumer consumer; + + private ChannelResolver channelResolver; + + private boolean running; + + private Object lifecycleMonitor = new Object(); + + + public DefaultMessageEndpoint(MessageSource source) { + this.source = source; + } + + public void setTarget(MessageTarget target) { + this.target = target; + } + + public void setHandler(MessageHandler handler) { + this.handler = handler; + } + + public void setConsumerType(ConsumerType consumerType) { + this.consumerType = consumerType; + } + + public void setChannelResolver(ChannelResolver channelResolver) { + this.channelResolver = channelResolver; + } + + + protected AbstractConsumer createDefaultConsumer() { + MessageHandler handlerAdapter = new MessageHandlerAdapter(); + if (this.consumerType.equals(ConsumerType.EVENT_DRIVEN)) { + return new EventDrivenConsumer(this.source, handlerAdapter); + } + else if (this.consumerType.equals(ConsumerType.FIXED_RATE)) { + return new FixedRateConsumer(this.source, handlerAdapter); + } + else if (this.consumerType.equals(ConsumerType.FIXED_DELAY)) { + return new FixedDelayConsumer(this.source, handlerAdapter); + } + else { + throw new UnsupportedOperationException("the consumerType '" + + this.consumerType.name() + "' is not supported."); + } + } + + public final void start() { + synchronized (this.lifecycleMonitor) { + if (this.source != null && this.consumer == null) { + this.consumer = createDefaultConsumer(); + } + this.consumer.initialize(); + this.running = true; + } + } + + public final void stop() { + synchronized (this.lifecycleMonitor) { + if (this.running) { + if (this.consumer != null) { + this.consumer.stop(); + } + this.running = false; + } + } + } + + public final boolean isRunning() { + synchronized (this.lifecycleMonitor) { + return this.running; + } + } + + + private class MessageHandlerAdapter implements MessageHandler { + + public Message handle(Message message) { + if (handler == null) { + target.send(message); + return null; + } + Message replyMessage = handler.handle(message); + if (replyMessage != null) { + MessageTarget replyTarget = resolveReplyTarget(message); + if (replyTarget == null) { + throw new MessageHandlingException("Unable to determine reply target for message. " + + "Provide a 'replyChannelName' in the message header or a 'defaultReplyChannel' " + + "on the message endpoint."); + } + replyTarget.send(replyMessage); + } + return null; + } + + private MessageTarget resolveReplyTarget(Message message) { + MessageTarget replyTo = null; + if (channelResolver != null) { + String replyChannelName = message.getHeader().getReplyChannelName(); + if (replyChannelName != null && replyChannelName.trim().length() > 0) { + replyTo = channelResolver.resolve(replyChannelName); + } + } + return (replyTo != null ? replyTo : target); + } + } + +}