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

Added MessageSelector list to MessageReceivingExecutor.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-10 03:14:48 +00:00
parent 7667ec63c4
commit a3ceb15cf1
3 changed files with 36 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultMessageDispatcher.java
View File

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

18
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageReceivingExecutor.java
View File

@@ -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<MessageSelector> selectors = new CopyOnWriteArrayList<MessageSelector>();
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;
}
/**

31
spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java
View File

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