Added DefaultMessageEndpoint

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:
+ * <ul>
+ *   <li><i>source channel-adapter:</i> source adapter + target channel</li>
+ *   <li><i>target channel-adapter:</i> source channel + target adapter</li>
+ *   <li><i>one-way:</i> source + handler that returns null and no target</li>
+ *   <li><i>request-reply:</i> source + handler and either a reply channel
+ *   specified on the request message or a default target on the endpoint</i>
+ * </ul>
+ * 
+ * @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);
+		}
+	}
+
+}