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

INT-3132 Add minSubscribers to Pub/SubChannel

Browse Source 
Send is deemed to be successful if sent to at least this
number of subscribers (default 0).

Note: when using a task executor, if there is at least one
subscriber, a send is always good, regardless of success
or failure of invoking the handler.

Polishing - Docs - PR Comments
This commit is contained in:
Gary Russell
2013-09-04 13:34:31 -04:00
parent e24d683478
commit 4abeebdb3a
9 changed files with 152 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -42,6 +42,8 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
private volatile boolean applySequence;
private volatile int minSubscribers;
private volatile int maxSubscribers = Integer.MAX_VALUE;
@Override
@@ -116,6 +118,18 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
this.maxSubscribers = maxSubscribers;
this.getDispatcher().setMaxSubscribers(maxSubscribers);
}
/**
* If at least this number of subscribers receive the message,
* {@link #send(org.springframework.integration.Message)}
* will return true. Default: 0.
* @param minSubscribers The minimum number of subscribers.
*/
public void setMinSubscribers(int minSubscribers) {
this.minSubscribers = minSubscribers;
this.getDispatcher().setMinSubscribers(minSubscribers);
}
/**
* Callback method for initialization.
*/
@@ -132,6 +146,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
this.dispatcher = new BroadcastingDispatcher(this.executor);
this.dispatcher.setIgnoreFailures(this.ignoreFailures);
this.dispatcher.setApplySequence(this.applySequence);
this.dispatcher.setMinSubscribers(this.minSubscribers);
this.dispatcher.setMaxSubscribers(this.maxSubscribers);
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,11 +16,12 @@
package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Parser for the <publish-subscribe-channel> element.
@@ -43,6 +44,7 @@ public class PublishSubscribeChannelParser extends AbstractChannelParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence");
this.setMaxSubscribersProperty(parserContext, builder, element,
IntegrationNamespaceUtils.DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "min-subscribers");
return builder;
}

33
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -50,6 +50,8 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
private final Executor executor;
private volatile int minSubscribers;
public BroadcastingDispatcher() {
this(null, false);
}
@@ -89,8 +91,17 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
this.applySequence = applySequence;
}
/**
* If at least this number of subscribers receive the message, {@link #dispatch(Message)}
* will return true. Default: 0.
* @param minSubscribers The minimum number of subscribers.
*/
public void setMinSubscribers(int minSubscribers) {
this.minSubscribers = minSubscribers;
}
public boolean dispatch(Message<?> message) {
boolean dispatched = false;
int dispatched = 0;
int sequenceNumber = 1;
Collection<MessageHandler> handlers = this.getHandlers();
if (this.requireSubscribers && handlers.size() == 0) {
@@ -106,14 +117,23 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
invokeHandler(handler, messageToSend);
}
});
dispatched = true;
dispatched++;
}
else {
boolean success = this.invokeHandler(handler, messageToSend);
dispatched = (success || dispatched);
if (this.invokeHandler(handler, messageToSend)) {
dispatched++;
}
}
}
return dispatched;
if (dispatched == 0 && this.minSubscribers == 0 && logger.isDebugEnabled()) {
if (sequenceSize > 0) {
logger.debug("No subscribers received message, default behavior is ignore");
}
else {
logger.debug("No subscribers, default behavior is ignore");
}
}
return dispatched >= minSubscribers;
}
private boolean invokeHandler(MessageHandler handler, Message<?> message) {
@@ -135,4 +155,5 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
}
}
}

9
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd
View File

@@ -449,6 +449,15 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="min-subscribers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the minimum subscribers required to be subscribed to this channel; if the minimum number
of subscribers receive the message, the send is deemed to be successful (returns true).
Defaults to 0.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="subscribersAttributeGroup" />
</xsd:extension>
</xsd:complexContent>

2
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests-context.xml
View File

@@ -29,6 +29,6 @@
<int:publish-subscribe-channel id="pubSubDefaultChannel" />
<int:publish-subscribe-channel id="pubSubExplicitChannel" max-subscribers="2 "/>
<int:publish-subscribe-channel id="pubSubExplicitChannel" max-subscribers="2 " min-subscribers="1"/>
</beans>

4
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -78,5 +78,7 @@ public abstract class DispatcherMaxSubscribersTests {
Integer explicitMax = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(pubSubExplicitChannel, "dispatcher"), "maxSubscribers", Integer.class);
assertEquals(val2, explicitMax.intValue());
Integer explicitMin = TestUtils.getPropertyValue(pubSubExplicitChannel, "dispatcher.minSubscribers", Integer.class);
assertEquals(1, explicitMin.intValue());
}
}

65
spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java
View File

@@ -36,9 +36,11 @@ import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -547,6 +549,69 @@ public class MessagingTemplateTests {
}
}
@Test
public void testNoSubs() {
MessagingTemplate template = new MessagingTemplate();
PublishSubscribeChannel channel = new PublishSubscribeChannel();
template.setDefaultChannel(channel);
template.convertAndSend("foo");
}
@Test
public void testSubsAllFail() {
MessagingTemplate template = new MessagingTemplate();
PublishSubscribeChannel channel = new PublishSubscribeChannel();
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException();
}
});
channel.setIgnoreFailures(true);
template.setDefaultChannel(channel);
template.convertAndSend("foo");
}
@Test
public void testEnoughSubs() {
MessagingTemplate template = new MessagingTemplate();
PublishSubscribeChannel channel = new PublishSubscribeChannel();
channel.setMinSubscribers(1);
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
}
});
template.setDefaultChannel(channel);
template.convertAndSend("foo");
}
@Test(expected=MessageDeliveryException.class)
public void testNoSubsFatal() {
MessagingTemplate template = new MessagingTemplate();
PublishSubscribeChannel channel = new PublishSubscribeChannel();
channel.setMinSubscribers(1);
template.setDefaultChannel(channel);
template.convertAndSend("foo");
}
@Test(expected=MessageDeliveryException.class)
public void testNotEnoughSubsFatal() {
MessagingTemplate template = new MessagingTemplate();
PublishSubscribeChannel channel = new PublishSubscribeChannel();
channel.setMinSubscribers(2);
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
}
});
template.setDefaultChannel(channel);
template.convertAndSend("foo");
}
@Test
public void testNeverReceive() {
MessagingTemplate template = new MessagingTemplate();