Added consumer class hierarchy

spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractConsumer.java

@@ -0,0 +1,160 @@
+/*
+ * 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.channel.consumer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.MessageSource;
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.message.Message;
+import org.springframework.util.Assert;
+
+/**
+ * Base class for consumers defining common properties and behavior.
+ * 
+ * @author Mark Fisher
+ */
+public abstract class AbstractConsumer implements Lifecycle {
+
+	/**
+	 * The default receive timeout: 1000 ms = 1 second.
+	 */
+	public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
+
+
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
+
+	private MessageSource source;
+
+	private MessageHandler handler;
+
+	private boolean active = false;
+
+	private boolean running = false;
+
+	private boolean autoStartup = true;
+
+	protected final Object lifecycleMonitor = new Object();
+
+
+	public AbstractConsumer(MessageSource source, MessageHandler handler) {
+		Assert.notNull(source, "source must not be null");
+		Assert.notNull(handler, "handler must not be null");
+		this.source = source;
+		this.handler = handler;
+	}
+
+
+	public void setReceiveTimeout(long receiveTimeout) {
+		this.receiveTimeout = receiveTimeout;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	public final boolean isRunning() {
+		synchronized (this.lifecycleMonitor) {
+			return this.running;
+		}
+	}
+
+	public final void start() {
+		synchronized (this.lifecycleMonitor) {
+			this.running = true;
+			this.lifecycleMonitor.notifyAll();
+		}
+		this.doStart();
+	}
+
+	public final void stop() {
+		synchronized (this.lifecycleMonitor) {
+			this.running = false;
+			this.lifecycleMonitor.notifyAll();
+		}
+		this.doStop();
+	}
+
+	public final boolean isActive() {
+		synchronized (this.lifecycleMonitor) {
+			return this.active;
+		}
+	}
+
+	public final void initialize() {
+		synchronized (this.lifecycleMonitor) {
+			this.active = true;
+			this.lifecycleMonitor.notifyAll();
+			if (this.autoStartup) {
+				this.start();
+			}
+		}
+		doInitialize();
+	}
+
+	/**
+	 * If the consumer is active but not yet running, then wait until it is running.
+	 */
+	protected void waitWhileNotRunning() {
+		synchronized (this.lifecycleMonitor) {
+			while (this.active && !this.running) {
+				try {
+					this.lifecycleMonitor.wait();
+				}
+				catch (InterruptedException ex) {
+					Thread.currentThread().interrupt();
+				}
+			}
+		}
+	}
+
+	protected boolean receiveAndHandle() {
+		boolean messageReceived = false;
+		Message message = null;
+		if (this.receiveTimeout < 0) { // indefinite timeout
+			message = this.source.receive();
+		}
+		else {
+			message = this.source.receive(this.receiveTimeout);
+		}
+		if (message != null) {
+			messageReceived = true;
+			messageReceived(message);
+			Message replyMessage = this.handler.handle(message);
+			if (replyMessage != null) {
+				handlerReplied(replyMessage);
+			}
+		}
+		return messageReceived;
+	}
+
+
+	protected abstract void doInitialize();
+
+	protected abstract void doStart();
+
+	protected abstract void doStop();
+
+	protected abstract void messageReceived(Message message);
+
+	protected abstract void handlerReplied(Message message);
+
+}

spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractPollingConsumer.java

@@ -0,0 +1,105 @@
+/*
+ * 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.channel.consumer;
+
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.integration.MessageSource;
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.message.Message;
+
+/**
+ * Base class for consumers that poll on a given interval.
+ * 
+ * @author Mark Fisher
+ */
+public abstract class AbstractPollingConsumer extends AbstractConsumer {
+
+	private static final int DEFAULT_POLL_INTERVAL = 1000;
+
+
+	protected ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
+
+	private int initialDelay = 0;
+
+	private int pollInterval = DEFAULT_POLL_INTERVAL;
+
+	private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
+
+
+	public AbstractPollingConsumer(MessageSource source, MessageHandler handler) {
+		super(source, handler);
+		this.setReceiveTimeout(0);
+	}
+
+
+	public void setPollInterval(int pollInterval) {
+		this.pollInterval = pollInterval;
+	}
+
+	/**
+	 * Specify the {@link TimeUnit} for polling. Default is milliseconds.
+	 */
+	public void setTimeUnit(TimeUnit timeUnit) {
+		this.timeUnit = timeUnit;
+	}
+
+	@Override
+	protected void doStart() {
+		scheduleInvoker(new PollingInvoker(), this.initialDelay, this.pollInterval, this.timeUnit);
+	}
+
+	/**
+	 * Each subclass must implement this method depending on its scheduling
+	 * behavior (e.g. fixed-rate versus fixed-delay).
+	 * 
+	 * @param invoker the invoker task to schedule
+	 * @param initialDelay the time in milliseconds to wait before the first
+	 * poll
+	 * @param pollInterval the polling interval in milliseconds
+	 */
+	protected abstract void scheduleInvoker(Runnable invoker, int initialDelay, int pollInterval, TimeUnit timeUnit);
+
+
+	@Override
+	protected void doInitialize() {
+	}
+
+	@Override
+	protected void doStop() {
+		this.executor.shutdown();
+	}
+
+	@Override
+	protected void messageReceived(Message message) {
+	}
+
+	@Override
+	protected void handlerReplied(Message message) {
+	}
+
+
+	private class PollingInvoker implements Runnable {
+
+		public void run() {
+			receiveAndHandle();
+		}
+
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/EventDrivenConsumer.java

@@ -0,0 +1,410 @@
+/*
+ * 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.channel.consumer;
+
+import java.lang.reflect.Method;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+
+import org.springframework.context.Lifecycle;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.MessageSource;
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.message.Message;
+import org.springframework.scheduling.SchedulingAwareRunnable;
+import org.springframework.scheduling.SchedulingTaskExecutor;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.util.Assert;
+
+/**
+ * A consumer that runs tasks repeatedly in order to invoke the handler as soon
+ * as a message is received by one of those tasks.
+ * 
+ * @author Mark Fisher
+ * @author Juergen Hoeller
+ */
+public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle {
+
+	private static final int DEFAULT_CONCURRENCY = 1;
+
+	private static final int DEFAULT_MAX_CONCURRENCY = 10;
+
+	private static final int DEFAULT_MAX_MESSAGES_PER_TASK = 10;
+
+	private static final int DEFAULT_IDLE_TASK_EXECUTION_LIMIT = 1;
+
+
+	private TaskExecutor executor;
+
+	private int concurrency = DEFAULT_CONCURRENCY;
+
+	private int maxConcurrency = DEFAULT_MAX_CONCURRENCY;
+
+	private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK;
+
+	private int idleTaskExecutionLimit = DEFAULT_IDLE_TASK_EXECUTION_LIMIT;
+
+	private final Set<MessageHandlerInvoker> scheduledInvokers = new HashSet<MessageHandlerInvoker>();
+
+	private int activeInvokerCount = 0;
+
+	private final Object activeInvokerMonitor = new Object();
+
+	private final List<MessageHandlerInvoker> pausedInvokers = new LinkedList<MessageHandlerInvoker>();
+
+
+	public EventDrivenConsumer(MessageSource source, MessageHandler handler) {
+		super(source, handler);
+	}
+
+
+	public void setExecutor(TaskExecutor executor) {
+		Assert.notNull(executor, "executor must not be null");
+		this.executor = executor;
+	}
+
+	public void setConcurrency(int concurrency) {
+		if (concurrency < 1) {
+			throw new IllegalArgumentException("'concurrency' value must be at least 1");
+		}
+		synchronized (this.activeInvokerMonitor) {
+			this.concurrency = concurrency;
+			if (this.maxConcurrency < concurrency) {
+				this.maxConcurrency = concurrency;
+			}
+		}
+	}
+
+	public void setMaxConcurrency(int maxConcurrency) {
+		if (maxConcurrency < 1) {
+			throw new IllegalArgumentException("'maxConcurrency' value must be at least 1");
+		}
+		synchronized (this.activeInvokerMonitor) {
+			this.maxConcurrency = Math.max(maxConcurrency, this.concurrency);
+		}
+	}
+
+	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
+		if (maxMessagesPerTask == 0) {
+			throw new IllegalArgumentException("'maxMessagesPerTask' must not be 0");
+		}
+		synchronized (this.activeInvokerMonitor) {
+			this.maxMessagesPerTask = maxMessagesPerTask;
+		}
+	}
+
+	public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) {
+		if (idleTaskExecutionLimit < 1) {
+			throw new IllegalArgumentException("'idleTaskExecutionLimit' must be at least 1");
+		}
+		synchronized (this.activeInvokerMonitor) {
+			this.idleTaskExecutionLimit = idleTaskExecutionLimit;
+		}
+	}
+
+	public void doStart() {
+		synchronized (this.lifecycleMonitor) {
+			this.resumePausedTasks();
+		}
+	}
+
+	public void doStop() {
+		this.shutdown();
+	}
+
+	public void doInitialize() {
+		synchronized (this.activeInvokerMonitor) {
+			if (this.executor == null) {
+				this.executor = createDefaultExecutor();
+			}
+			else if (this.executor instanceof SchedulingTaskExecutor &&
+					((SchedulingTaskExecutor) this.executor).prefersShortLivedTasks() &&
+					this.maxMessagesPerTask == Integer.MIN_VALUE) {
+				this.maxMessagesPerTask = 1;
+			}
+			initializeExecutorIfPossible();
+			for (int i = 0; i < this.concurrency; i++) {
+				scheduleNewInvoker();
+			}
+		}
+	}
+
+	private void initializeExecutorIfPossible() {
+		try {
+			Method initMethod = this.executor.getClass().getMethod("initialize");
+			initMethod.invoke(this.executor);
+		}
+		catch (Exception e) {
+			// do nothing
+		}
+	}
+
+	private void shutdown() {
+		synchronized (this.lifecycleMonitor) {
+			this.shutdownExecutorIfPossible();
+		}
+	}
+
+	private void shutdownExecutorIfPossible() {
+		try {
+			if (this.executor instanceof Lifecycle) {
+				((Lifecycle) this.executor).stop();
+			}
+			else {
+				Method shutdownMethod = this.executor.getClass().getMethod("shutdown");
+				shutdownMethod.invoke(this.executor);
+			}
+		}
+		catch (Exception e) {
+			// do nothing
+		}
+	}
+
+	private TaskExecutor createDefaultExecutor() {
+		ThreadPoolTaskExecutor defaultExecutor = new ThreadPoolTaskExecutor();
+		defaultExecutor.setCorePoolSize(this.maxConcurrency);
+		defaultExecutor.setMaxPoolSize(this.maxConcurrency);
+		defaultExecutor.setQueueCapacity(5);
+		return defaultExecutor;
+	}
+
+	/**
+	 * Try scheduling a new invoker, since we know messages are being received.
+	 * @see #scheduleNewInvokerIfAppropriate()
+	 */
+	@Override
+	protected void messageReceived(Message message) {
+		scheduleNewInvokerIfAppropriate();
+	}
+
+	@Override
+	protected void handlerReplied(Message message) {
+	}
+
+	/**
+	 * Schedule a new invoker, increasing the total number of scheduled
+	 * invokers for this consumer.
+	 */
+	private void scheduleNewInvoker() {
+		MessageHandlerInvoker invoker = new MessageHandlerInvoker();
+		if (rescheduleInvokerIfNecessary(invoker)) {
+			this.scheduledInvokers.add(invoker);
+		}
+	}
+
+	private boolean rescheduleInvokerIfNecessary(MessageHandlerInvoker invoker) {
+		synchronized (this.lifecycleMonitor) {
+			if (this.isRunning()) {
+				try {
+					doRescheduleInvoker(invoker);
+				}
+				catch (RuntimeException ex) {
+					logRejectedInvoker(invoker, ex);
+					this.pausedInvokers.add(invoker);
+				}
+				return true;
+			}
+			if (this.isActive()) {
+				this.pausedInvokers.add(invoker);
+				return true;
+			}
+			else {
+				return false;
+			}
+		}
+	}
+
+	protected void doRescheduleInvoker(final MessageHandlerInvoker invoker) {
+		this.executor.execute(invoker);
+	}
+
+	private boolean shouldRescheduleInvoker(int idleTaskExecutionCount) {
+		synchronized (this.activeInvokerMonitor) {
+			boolean idle = (idleTaskExecutionCount >= this.idleTaskExecutionLimit);
+			return (this.scheduledInvokers.size() <= (idle ? this.concurrency : this.maxConcurrency));
+		}
+	}
+
+	private boolean hasIdleInvokers() {
+		for (MessageHandlerInvoker invoker : this.scheduledInvokers) {
+			if (invoker.isIdle()) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	/**
+	 * Try to resume all paused tasks.
+	 * Tasks for which rescheduling failed simply remain in paused mode.
+	 */
+	protected void resumePausedTasks() {
+		synchronized (this.lifecycleMonitor) {
+			if (!this.pausedInvokers.isEmpty()) {
+				for (Iterator<MessageHandlerInvoker> it = this.pausedInvokers.iterator(); it.hasNext();) {
+					MessageHandlerInvoker invoker = it.next();
+					try {
+						doRescheduleInvoker(invoker);
+						it.remove();
+						if (logger.isDebugEnabled()) {
+							logger.debug("Resumed paused invoker: " + invoker);
+						}
+					}
+					catch (RuntimeException e) {
+						logRejectedInvoker(invoker, e);
+						// Keep the task in paused mode...
+					}
+				}
+			}
+		}
+	}
+
+	public int getPausedInvokerCount() {
+		synchronized (this.lifecycleMonitor) {
+			return this.pausedInvokers.size();
+		}
+	}
+
+	/**
+	 * Log an invoker that has been rejected by {@link #doRescheduleInvoker}.
+	 * <p>The default implementation simply logs a corresponding message
+	 * at debug level.
+	 * @param invoker the rejected invoker object
+	 * @param ex the exception thrown from {@link #doRescheduleInvoker}
+	 */
+	protected void logRejectedInvoker(Object invoker, RuntimeException ex) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Invoker [" + invoker + "] has been rejected and paused: " + ex);
+		}
+	}
+
+	private void scheduleNewInvokerIfAppropriate() {
+		if (this.isRunning()) {
+			this.resumePausedTasks();
+			synchronized (this.activeInvokerMonitor) {
+				if (this.scheduledInvokers.size() < this.maxConcurrency && !hasIdleInvokers()) {
+					scheduleNewInvoker();
+					if (logger.isDebugEnabled()) {
+						logger.debug("Raised scheduled invoker count: " + scheduledInvokers.size());
+					}
+				}
+			}
+		}
+	}
+
+	public final int getScheduledInvokerCount() {
+		synchronized (this.activeInvokerMonitor) {
+			return this.scheduledInvokers.size();
+		}
+	}
+
+	public final int getActiveInvokerCount() {
+		synchronized (this.activeInvokerMonitor) {
+			return this.activeInvokerCount;
+		}
+	}
+
+
+	private class MessageHandlerInvoker implements SchedulingAwareRunnable {
+
+		private int idleTaskExecutionCount = 0;
+
+		private volatile boolean idle = true;
+
+
+		public void run() {
+			synchronized (activeInvokerMonitor) {
+				activeInvokerCount++;
+				activeInvokerMonitor.notifyAll();
+			}
+			boolean messageReceived = false;
+			//TODO: try {
+				if (maxMessagesPerTask < 0) {
+					while (isActive()) {
+						waitWhileNotRunning();
+						if (isActive()) {
+							messageReceived = invokeHandler();
+						}
+					}
+				}
+				else {
+					int messageCount = 0;
+					while (isRunning() && messageCount < maxMessagesPerTask) {
+						boolean messageHandled = invokeHandler();
+						this.idle = !messageHandled;
+						messageReceived = (messageHandled || messageReceived);
+						messageCount++;
+					}
+				}
+			// TODO: } catch (Throwable t) { check if last message succeeded, else sleep between recovery attempts } 
+			synchronized (activeInvokerMonitor) {
+				activeInvokerCount--;
+				activeInvokerMonitor.notifyAll();
+			}
+			if (!messageReceived) {
+				this.idleTaskExecutionCount++;
+			}
+			else {
+				this.idleTaskExecutionCount = 0;
+			}
+			if (!shouldRescheduleInvoker(this.idleTaskExecutionCount) || !rescheduleInvokerIfNecessary(this)) {
+				this.shutdown();
+			}
+			else if (isRunning()) {
+				int nonPausedInvokers = getScheduledInvokerCount() - getPausedInvokerCount();
+				if (nonPausedInvokers < 1) {
+					logger.error("All scheduled invokers have been paused, probably due to tasks having been rejected. " +
+							"Check your thread pool configuration! Manual recovery necessary through a start() call.");
+				}
+				else if (nonPausedInvokers < concurrency) {
+					logger.warn("Number of scheduled invokers has dropped below concurrency limit, probably " +
+							"due to tasks having been rejected. Check your thread pool configuration! Automatic recovery " +
+							"to be triggered by remaining invokers.");
+				}
+			}
+		}
+
+		private void shutdown() {
+			synchronized (activeInvokerMonitor) {
+				scheduledInvokers.remove(this);
+				if (logger.isDebugEnabled()) {
+					logger.debug("Lowered scheduled invoker count: " + scheduledInvokers.size());
+				}
+				activeInvokerMonitor.notifyAll();
+			}			
+		}
+
+		private boolean invokeHandler() {
+			boolean messageReceived = receiveAndHandle();
+			this.idle = !messageReceived;
+			return messageReceived;
+		}
+
+		public boolean isLongLived() {
+			return (maxMessagesPerTask < 0);
+		}
+
+		public boolean isIdle() {
+			return this.idle;
+		}
+
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedDelayConsumer.java

@@ -0,0 +1,42 @@
+/*
+ * 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.channel.consumer;
+
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.integration.MessageSource;
+import org.springframework.integration.handler.MessageHandler;
+
+/**
+ * A consumer that measures the <code>pollInterval</code> between the
+ * invoker's execution completion time and the subsequent start.
+ * 
+ * @author Mark Fisher
+ */
+public class FixedDelayConsumer extends AbstractPollingConsumer {
+
+	public FixedDelayConsumer(MessageSource source, MessageHandler handler) {
+		super(source, handler);
+	}
+
+
+	@Override
+	protected void scheduleInvoker(Runnable invoker, int initialDelay, int pollInterval, TimeUnit timeUnit) {
+		this.executor.scheduleWithFixedDelay(invoker, initialDelay, pollInterval, timeUnit);
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedRateConsumer.java

@@ -0,0 +1,42 @@
+/*
+ * 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.channel.consumer;
+
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.integration.MessageSource;
+import org.springframework.integration.handler.MessageHandler;
+
+/**
+ * A consumer that measures the <code>pollInterval</code> between each
+ * execution's start.
+ * 
+ * @author Mark Fisher
+ */
+public class FixedRateConsumer extends AbstractPollingConsumer {
+
+	public FixedRateConsumer(MessageSource source, MessageHandler handler) {
+		super(source, handler);
+	}
+
+
+	@Override
+	protected void scheduleInvoker(Runnable invoker, int initialDelay, int pollInterval, TimeUnit timeUnit) {
+		this.executor.scheduleAtFixedRate(invoker, initialDelay, pollInterval, timeUnit);
+	}
+
+}