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 index 59ec2d2ca3..a93bf64fe0 100644 --- 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 @@ -21,7 +21,7 @@ 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.endpoint.MessageEndpoint; import org.springframework.integration.message.Message; import org.springframework.util.Assert; @@ -44,7 +44,7 @@ public abstract class AbstractConsumer implements Lifecycle { private MessageSource source; - private MessageHandler handler; + private MessageEndpoint endpoint; private boolean active = false; @@ -55,11 +55,11 @@ public abstract class AbstractConsumer implements Lifecycle { protected final Object lifecycleMonitor = new Object(); - public AbstractConsumer(MessageSource source, MessageHandler handler) { + public AbstractConsumer(MessageSource source, MessageEndpoint endpoint) { Assert.notNull(source, "source must not be null"); - Assert.notNull(handler, "handler must not be null"); + Assert.notNull(endpoint, "endpoint must not be null"); this.source = source; - this.handler = handler; + this.endpoint = endpoint; } @@ -128,7 +128,7 @@ public abstract class AbstractConsumer implements Lifecycle { } } - protected boolean receiveAndHandle() { + protected boolean receiveAndPassToEndpoint() { boolean messageReceived = false; Message message = null; if (this.receiveTimeout < 0) { // indefinite timeout @@ -140,10 +140,7 @@ public abstract class AbstractConsumer implements Lifecycle { if (message != null) { messageReceived = true; messageReceived(message); - Message replyMessage = this.handler.handle(message); - if (replyMessage != null) { - handlerReplied(replyMessage); - } + this.endpoint.messageReceived(message); } return messageReceived; } @@ -157,6 +154,4 @@ public abstract class AbstractConsumer implements Lifecycle { 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 index 13a1d21d97..91e61645ab 100644 --- 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 @@ -20,7 +20,7 @@ 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.endpoint.MessageEndpoint; import org.springframework.integration.message.Message; /** @@ -42,8 +42,8 @@ public abstract class AbstractPollingConsumer extends AbstractConsumer { private TimeUnit timeUnit = TimeUnit.MILLISECONDS; - public AbstractPollingConsumer(MessageSource source, MessageHandler handler) { - super(source, handler); + public AbstractPollingConsumer(MessageSource source, MessageEndpoint endpoint) { + super(source, endpoint); this.setReceiveTimeout(0); } @@ -89,15 +89,11 @@ public abstract class AbstractPollingConsumer extends AbstractConsumer { protected void messageReceived(Message message) { } - @Override - protected void handlerReplied(Message message) { - } - private class PollingInvoker implements Runnable { public void run() { - receiveAndHandle(); + receiveAndPassToEndpoint(); } } 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 index b522c7a39a..b5a4de5c59 100644 --- 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 @@ -26,7 +26,7 @@ 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.endpoint.MessageEndpoint; import org.springframework.integration.message.Message; import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.scheduling.SchedulingTaskExecutor; @@ -34,8 +34,8 @@ 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. + * A consumer that runs tasks repeatedly in order to pass to the endpoint as + * soon as a message is received by one of those tasks. * * @author Mark Fisher * @author Juergen Hoeller @@ -61,17 +61,17 @@ public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle { private int idleTaskExecutionLimit = DEFAULT_IDLE_TASK_EXECUTION_LIMIT; - private final Set scheduledInvokers = new HashSet(); + private final Set scheduledInvokers = new HashSet(); private int activeInvokerCount = 0; private final Object activeInvokerMonitor = new Object(); - private final List pausedInvokers = new LinkedList(); + private final List pausedInvokers = new LinkedList(); - public EventDrivenConsumer(MessageSource source, MessageHandler handler) { - super(source, handler); + public EventDrivenConsumer(MessageSource source, MessageEndpoint endpoint) { + super(source, endpoint); } @@ -194,22 +194,18 @@ public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle { 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(); + MessageEndpointInvoker invoker = new MessageEndpointInvoker(); if (rescheduleInvokerIfNecessary(invoker)) { this.scheduledInvokers.add(invoker); } } - private boolean rescheduleInvokerIfNecessary(MessageHandlerInvoker invoker) { + private boolean rescheduleInvokerIfNecessary(MessageEndpointInvoker invoker) { synchronized (this.lifecycleMonitor) { if (this.isRunning()) { try { @@ -231,7 +227,7 @@ public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle { } } - protected void doRescheduleInvoker(final MessageHandlerInvoker invoker) { + protected void doRescheduleInvoker(final MessageEndpointInvoker invoker) { this.executor.execute(invoker); } @@ -243,7 +239,7 @@ public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle { } private boolean hasIdleInvokers() { - for (MessageHandlerInvoker invoker : this.scheduledInvokers) { + for (MessageEndpointInvoker invoker : this.scheduledInvokers) { if (invoker.isIdle()) { return true; } @@ -258,8 +254,8 @@ public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle { protected void resumePausedTasks() { synchronized (this.lifecycleMonitor) { if (!this.pausedInvokers.isEmpty()) { - for (Iterator it = this.pausedInvokers.iterator(); it.hasNext();) { - MessageHandlerInvoker invoker = it.next(); + for (Iterator it = this.pausedInvokers.iterator(); it.hasNext();) { + MessageEndpointInvoker invoker = it.next(); try { doRescheduleInvoker(invoker); it.remove(); @@ -322,7 +318,7 @@ public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle { } - private class MessageHandlerInvoker implements SchedulingAwareRunnable { + private class MessageEndpointInvoker implements SchedulingAwareRunnable { private int idleTaskExecutionCount = 0; @@ -340,14 +336,15 @@ public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle { while (isActive()) { waitWhileNotRunning(); if (isActive()) { - messageReceived = invokeHandler(); + messageReceived = receiveAndPassToEndpoint(); + this.idle = !messageReceived; } } } else { int messageCount = 0; while (isRunning() && messageCount < maxMessagesPerTask) { - boolean messageHandled = invokeHandler(); + boolean messageHandled = receiveAndPassToEndpoint(); this.idle = !messageHandled; messageReceived = (messageHandled || messageReceived); messageCount++; @@ -391,12 +388,6 @@ public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle { } } - private boolean invokeHandler() { - boolean messageReceived = receiveAndHandle(); - this.idle = !messageReceived; - return messageReceived; - } - public boolean isLongLived() { return (maxMessagesPerTask < 0); } 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 index 373ba1401e..1bfb28fccb 100644 --- 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 @@ -19,7 +19,7 @@ package org.springframework.integration.channel.consumer; import java.util.concurrent.TimeUnit; import org.springframework.integration.MessageSource; -import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.endpoint.MessageEndpoint; /** * A consumer that measures the pollInterval between the @@ -29,8 +29,8 @@ import org.springframework.integration.handler.MessageHandler; */ public class FixedDelayConsumer extends AbstractPollingConsumer { - public FixedDelayConsumer(MessageSource source, MessageHandler handler) { - super(source, handler); + public FixedDelayConsumer(MessageSource source, MessageEndpoint endpoint) { + super(source, endpoint); } 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 index 92f46057e2..b6535f1050 100644 --- 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 @@ -19,7 +19,7 @@ package org.springframework.integration.channel.consumer; import java.util.concurrent.TimeUnit; import org.springframework.integration.MessageSource; -import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.endpoint.MessageEndpoint; /** * A consumer that measures the pollInterval between each @@ -29,8 +29,8 @@ import org.springframework.integration.handler.MessageHandler; */ public class FixedRateConsumer extends AbstractPollingConsumer { - public FixedRateConsumer(MessageSource source, MessageHandler handler) { - super(source, handler); + public FixedRateConsumer(MessageSource source, MessageEndpoint endpoint) { + super(source, endpoint); } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java index 6cad431dcd..b3468c27cd 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java @@ -61,6 +61,8 @@ public class GenericMessageEndpoint implements MessageEndpoint, Lifecycle { private Object lifecycleMonitor = new Object(); + public GenericMessageEndpoint() {} + /** * Create an endpoint to consume messages from the given source. */ @@ -68,6 +70,12 @@ public class GenericMessageEndpoint implements MessageEndpoint, Lifecycle { this.source = source; } + /** + * Set the source from which this endpoint receives messages. + */ + public void setSource(MessageSource source) { + this.source = source; + } /** * Set the target to which this endpoint can send messages. @@ -91,6 +99,13 @@ public class GenericMessageEndpoint implements MessageEndpoint, Lifecycle { this.consumerType = consumerType; } + /** + * Return the consumer type to use for this endpoint's source. + */ + public ConsumerType getConsumerType() { + return this.consumerType; + } + /** * Set the channel resolver strategy to use when a message * provides a 'replyChannelName'. @@ -104,15 +119,14 @@ public class GenericMessageEndpoint implements MessageEndpoint, Lifecycle { * Create a consumer based upon the specified consumer type. */ protected AbstractConsumer createDefaultConsumer() { - MessageHandler handlerAdapter = new MessageHandlerAdapter(); if (this.consumerType.equals(ConsumerType.EVENT_DRIVEN)) { - return new EventDrivenConsumer(this.source, handlerAdapter); + return new EventDrivenConsumer(this.source, this); } else if (this.consumerType.equals(ConsumerType.FIXED_RATE)) { - return new FixedRateConsumer(this.source, handlerAdapter); + return new FixedRateConsumer(this.source, this); } else if (this.consumerType.equals(ConsumerType.FIXED_DELAY)) { - return new FixedDelayConsumer(this.source, handlerAdapter); + return new FixedDelayConsumer(this.source, this); } else { throw new UnsupportedOperationException("the consumerType '" @@ -156,37 +170,31 @@ public class GenericMessageEndpoint implements MessageEndpoint, Lifecycle { } } - - 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 'target' " - + "on the message endpoint."); - } - replyTarget.send(replyMessage); - } - return null; + public void messageReceived(Message message) { + if (this.handler == null) { + target.send(message); } - - 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); - } + 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 'target' " + + "on the message endpoint."); } - return (replyTo != null ? replyTo : target); + replyTarget.send(replyMessage); } } + private MessageTarget resolveReplyTarget(Message message) { + MessageTarget replyTo = null; + if (this.channelResolver != null) { + String replyChannelName = message.getHeader().getReplyChannelName(); + if (replyChannelName != null && replyChannelName.trim().length() > 0) { + replyTo = this.channelResolver.resolve(replyChannelName); + } + } + return (replyTo != null ? replyTo : target); + } + } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index 7b5f0c2fb0..67e14f0524 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -16,6 +16,9 @@ package org.springframework.integration.endpoint; +import org.springframework.integration.channel.consumer.ConsumerType; +import org.springframework.integration.message.Message; + /** * Base interface for message endpoints. * @@ -23,4 +26,8 @@ package org.springframework.integration.endpoint; */ public interface MessageEndpoint { + ConsumerType getConsumerType(); + + void messageReceived(Message message); + } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java index a6eb4c0b17..1d6414328c 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java @@ -26,7 +26,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.springframework.integration.channel.PointToPointChannel; -import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @@ -50,16 +50,19 @@ public class EventDrivenConsumerTests { executor.setMaxPoolSize(maxConcurrency); executor.setQueueCapacity(0); PointToPointChannel channel = new PointToPointChannel(); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { + MessageEndpoint endpoint = new MessageEndpoint() { + public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); activeSum.set(activeSum.addAndGet(executor.getActiveCount())); maxActive.set(Math.max(executor.getActiveCount(), maxActive.get())); - return null; + } + + public ConsumerType getConsumerType() { + return ConsumerType.EVENT_DRIVEN; } }; - EventDrivenConsumer consumer = new EventDrivenConsumer(channel, handler); + EventDrivenConsumer consumer = new EventDrivenConsumer(channel, endpoint); consumer.setExecutor(executor); consumer.setConcurrency(concurrency); consumer.setMaxConcurrency(maxConcurrency); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java index 48edce2ec5..c6237ebfd7 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java @@ -26,7 +26,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.springframework.integration.channel.PointToPointChannel; -import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; @@ -41,14 +41,17 @@ public class FixedDelayConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); PointToPointChannel channel = new PointToPointChannel(); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { + MessageEndpoint endpoint = new MessageEndpoint() { + public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); - return null; + } + + public ConsumerType getConsumerType() { + return ConsumerType.FIXED_DELAY; } }; - FixedDelayConsumer consumer = new FixedDelayConsumer(channel, handler); + FixedDelayConsumer consumer = new FixedDelayConsumer(channel, endpoint); consumer.setPollInterval(10); consumer.initialize(); for (int i = 0; i < messagesToSend; i++) { @@ -64,14 +67,17 @@ public class FixedDelayConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); PointToPointChannel channel = new PointToPointChannel(); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { + MessageEndpoint endpoint = new MessageEndpoint() { + public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); - return null; + } + + public ConsumerType getConsumerType() { + return ConsumerType.FIXED_DELAY; } }; - FixedDelayConsumer consumer = new FixedDelayConsumer(channel, handler); + FixedDelayConsumer consumer = new FixedDelayConsumer(channel, endpoint); consumer.setPollInterval(10); consumer.initialize(); for (int i = 0; i < messagesToSend; i++) { diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java index 82c0ca98bb..5388e2057c 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java @@ -26,7 +26,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.springframework.integration.channel.PointToPointChannel; -import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; @@ -41,14 +41,17 @@ public class FixedRateConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); PointToPointChannel channel = new PointToPointChannel(); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { + MessageEndpoint endpoint = new MessageEndpoint() { + public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); - return null; + } + + public ConsumerType getConsumerType() { + return ConsumerType.FIXED_RATE; } }; - FixedRateConsumer consumer = new FixedRateConsumer(channel, handler); + FixedRateConsumer consumer = new FixedRateConsumer(channel, endpoint); consumer.setPollInterval(10); consumer.initialize(); for (int i = 0; i < messagesToSend; i++) { @@ -64,14 +67,17 @@ public class FixedRateConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); PointToPointChannel channel = new PointToPointChannel(); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { + MessageEndpoint endpoint = new MessageEndpoint() { + public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); - return null; + } + + public ConsumerType getConsumerType() { + return ConsumerType.FIXED_RATE; } }; - FixedRateConsumer consumer = new FixedRateConsumer(channel, handler); + FixedRateConsumer consumer = new FixedRateConsumer(channel, endpoint); consumer.setPollInterval(10); consumer.initialize(); for (int i = 0; i < messagesToSend; i++) {