From 193f29a0b5308205df0ed8bcbd6f2d8e3be34439 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sun, 2 Dec 2007 18:17:07 +0000 Subject: [PATCH] Added consumer class hierarchy --- .../channel/consumer/AbstractConsumer.java | 160 +++++++ .../consumer/AbstractPollingConsumer.java | 105 +++++ .../channel/consumer/EventDrivenConsumer.java | 410 ++++++++++++++++++ .../channel/consumer/FixedDelayConsumer.java | 42 ++ .../channel/consumer/FixedRateConsumer.java | 42 ++ 5 files changed, 759 insertions(+) create mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractConsumer.java create mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractPollingConsumer.java create mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/EventDrivenConsumer.java create mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedDelayConsumer.java create mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedRateConsumer.java diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractConsumer.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractConsumer.java new file mode 100644 index 0000000000..63d662bd6e --- /dev/null +++ b/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); + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractPollingConsumer.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractPollingConsumer.java new file mode 100644 index 0000000000..13a1d21d97 --- /dev/null +++ b/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(); + } + + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/EventDrivenConsumer.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/EventDrivenConsumer.java new file mode 100644 index 0000000000..b522c7a39a --- /dev/null +++ b/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 scheduledInvokers = new HashSet(); + + private int activeInvokerCount = 0; + + private final Object activeInvokerMonitor = new Object(); + + private final List pausedInvokers = new LinkedList(); + + + 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 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}. + *

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; + } + + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedDelayConsumer.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedDelayConsumer.java new file mode 100644 index 0000000000..373ba1401e --- /dev/null +++ b/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 pollInterval 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); + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedRateConsumer.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedRateConsumer.java new file mode 100644 index 0000000000..92f46057e2 --- /dev/null +++ b/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 pollInterval 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); + } + +}