Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Consumers now call endpoint.messageReceived() instead of invoking a MessageHandler

Browse Source
This commit is contained in:
Mark Fisher
2007-12-07 21:37:03 +00:00
parent 105ff9804b
commit a670fb1717
10 changed files with 119 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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);
}

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

@@ -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();
}
}

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

@@ -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<MessageHandlerInvoker> scheduledInvokers = new HashSet<MessageHandlerInvoker>();
private final Set<MessageEndpointInvoker> scheduledInvokers = new HashSet<MessageEndpointInvoker>();
private int activeInvokerCount = 0;
private final Object activeInvokerMonitor = new Object();
private final List<MessageHandlerInvoker> pausedInvokers = new LinkedList<MessageHandlerInvoker>();
private final List<MessageEndpointInvoker> pausedInvokers = new LinkedList<MessageEndpointInvoker>();
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<MessageHandlerInvoker> it = this.pausedInvokers.iterator(); it.hasNext();) {
MessageHandlerInvoker invoker = it.next();
for (Iterator<MessageEndpointInvoker> 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);
}

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

@@ -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 <code>pollInterval</code> 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);
}

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

@@ -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 <code>pollInterval</code> 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);
}

72
spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java
View File

@@ -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 '<i>replyChannelName</i>'.
@@ -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);
}
}

7
spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
View File

@@ -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);
}

13
spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java
View File

@@ -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);

24
spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java
View File

@@ -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++) {

24
spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java
View File

@@ -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++) {