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

BroadcastingDispatcher tries to send to all consumers even if one "Selective Consumer" throws a MessageRejectedException (consistent with SimpleDispatcher's behavior).

Browse Source
This commit is contained in:
Mark Fisher
2008-09-25 03:52:35 +00:00
parent 91c094a174
commit 60e245f5a6
4 changed files with 24 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd
View File

@@ -226,9 +226,9 @@
<xsd:element ref="poller" minOccurs="0" maxOccurs="1"/>
<xsd:element name="interceptors" type="endpointInterceptorsType" minOccurs="0" maxOccurs="1"/>
</xsd:all>
<xsd:attribute name="input-channel" type="xsd:string" use="required"/>
<xsd:attribute name="ref" type="xsd:string"/>
<xsd:attribute name="method" type="xsd:string"/>
<xsd:attribute name="input-channel" type="xsd:string" use="required"/>
<xsd:attribute name="selector" type="xsd:string"/>
<xsd:attribute name="error-handler" type="xsd:string"/>
</xsd:extension>

19
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java
View File

@@ -23,7 +23,9 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageRejectedException;
/**
* Base class for {@link MessageDispatcher} implementations.
@@ -64,4 +66,21 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
return this.getClass().getSimpleName() + " with subscribers: " + this.subscribers;
}
/**
* Convenience method available for subclasses. Returns 'true' unless a
* "Selective Consumer" throws a {@link MessageRejectedException}.
*/
protected boolean sendMessageToConsumer(Message<?> message, MessageConsumer consumer) {
try {
consumer.onMessage(message);
return true;
}
catch (MessageRejectedException e) {
if (logger.isDebugEnabled()) {
logger.debug("Consumer '" + consumer + "' rejected Message, continuing with other subscribers if available.", e);
}
return false;
}
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java
View File

@@ -56,12 +56,12 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
if (executor != null) {
executor.execute(new Runnable() {
public void run() {
consumer.onMessage(messageToSend);
BroadcastingDispatcher.this.sendMessageToConsumer(messageToSend, consumer);
}
});
}
else {
consumer.onMessage(messageToSend);
this.sendMessageToConsumer(messageToSend, consumer);
}
}
return true;

10
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java
View File

@@ -42,16 +42,10 @@ public class SimpleDispatcher extends AbstractDispatcher {
int rejectedExceptionCount = 0;
for (MessageConsumer consumer : this.subscribers) {
count++;
try {
consumer.onMessage(message);
if (this.sendMessageToConsumer(message, consumer)) {
return true;
}
catch (MessageRejectedException e) {
rejectedExceptionCount++;
if (logger.isDebugEnabled()) {
logger.debug("Consumer '" + consumer + "' rejected Message, continuing with other subscribers if available.", e);
}
}
rejectedExceptionCount++;
}
if (rejectedExceptionCount == count) {
throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message.");