From a3ceb15cf11626770e435e0de4e00cf813b6821f Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 10 Jan 2008 03:14:48 +0000 Subject: [PATCH] Added MessageSelector list to MessageReceivingExecutor. --- .../bus/DefaultMessageDispatcher.java | 10 +++--- .../bus/MessageReceivingExecutor.java | 18 ++++++++++- .../bus/DefaultMessageDispatcherTests.java | 31 +++++++++---------- 3 files changed, 36 insertions(+), 23 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultMessageDispatcher.java index 2aa2c1f7a6..c285186d71 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultMessageDispatcher.java @@ -108,14 +108,11 @@ public class DefaultMessageDispatcher extends AbstractMessageDispatcher { continue; } try { - executor.processMessage(message); - if (!this.broadcast) { + boolean accepted = executor.acceptMessage(message); + if (accepted && !this.broadcast) { return true; } iter.remove(); - if (!iter.hasNext() && !encounteredRejection) { - return true; - } } catch (RejectedExecutionException rex) { encounteredRejection = true; @@ -124,6 +121,9 @@ public class DefaultMessageDispatcher extends AbstractMessageDispatcher { } } } + if (this.broadcast && !encounteredRejection) { + return true; + } attempts++; } if (this.shouldFailOnRejectionLimit) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageReceivingExecutor.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageReceivingExecutor.java index 4a42a0557c..8232bca37e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageReceivingExecutor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageReceivingExecutor.java @@ -16,6 +16,8 @@ package org.springframework.integration.bus; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -27,6 +29,7 @@ import org.springframework.context.Lifecycle; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageReceiver; +import org.springframework.integration.message.selector.MessageSelector; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; @@ -41,6 +44,8 @@ public class MessageReceivingExecutor implements Lifecycle { private MessageReceiver receiver; + private List selectors = new CopyOnWriteArrayList(); + private ThreadPoolExecutor threadPoolExecutor; private int corePoolSize; @@ -78,6 +83,11 @@ public class MessageReceivingExecutor implements Lifecycle { this.maxPoolSize = maxPoolSize; } + public void addMessageSelector(MessageSelector messageSelector) { + Assert.notNull(messageSelector, "'messageSelector' must not be null"); + this.selectors.add(messageSelector); + } + public boolean isRunning() { return this.running; } @@ -101,11 +111,17 @@ public class MessageReceivingExecutor implements Lifecycle { } } - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { if (threadPoolExecutor == null) { throw new MessageHandlingException("executor is not running"); } + for (MessageSelector selector : this.selectors) { + if (!selector.accept(message)) { + return false; + } + } this.threadPoolExecutor.execute(new MessageReceivingTask(this.receiver, message)); + return true; } /** diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java index f048e40c98..1d37483374 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java @@ -160,7 +160,7 @@ public class DefaultMessageDispatcherTests { dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1)); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { throw new RejectedExecutionException(); } }); @@ -189,7 +189,7 @@ public class DefaultMessageDispatcherTests { dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1)); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { throw new RejectedExecutionException(); } }); @@ -216,13 +216,13 @@ public class DefaultMessageDispatcherTests { dispatcher.setRetryInterval(3); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { throw new RejectedExecutionException(); } }); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { throw new RejectedExecutionException(); } }); @@ -248,14 +248,14 @@ public class DefaultMessageDispatcherTests { dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { rejectedCounter1.incrementAndGet(); throw new RejectedExecutionException(); } }); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { rejectedCounter2.incrementAndGet(); throw new RejectedExecutionException(); } @@ -288,17 +288,16 @@ public class DefaultMessageDispatcherTests { dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { rejectedCounter1.incrementAndGet(); throw new RejectedExecutionException(); } }); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { if (rejectedCounter2.get() == 1) { - super.processMessage(message); - return; + return super.acceptMessage(message); } rejectedCounter2.incrementAndGet(); throw new RejectedExecutionException(); @@ -306,7 +305,7 @@ public class DefaultMessageDispatcherTests { }); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint3, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { rejectedCounter3.incrementAndGet(); throw new RejectedExecutionException(); } @@ -341,10 +340,9 @@ public class DefaultMessageDispatcherTests { dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { if (rejectedCounter1.get() == 2) { - super.processMessage(message); - return; + return super.acceptMessage(message); } rejectedCounter1.incrementAndGet(); throw new RejectedExecutionException(); @@ -352,10 +350,9 @@ public class DefaultMessageDispatcherTests { }); dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1) { @Override - public void processMessage(Message message) { + public boolean acceptMessage(Message message) { if (rejectedCounter2.get() == 4) { - super.processMessage(message); - return; + return super.acceptMessage(message); } rejectedCounter2.incrementAndGet(); throw new RejectedExecutionException();