From 4abeebdb3aaaaa8717a77c7df5e898c14fb4cfa9 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 4 Sep 2013 13:34:31 -0400 Subject: [PATCH] INT-3132 Add minSubscribers to Pub/SubChannel 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 --- .../channel/PublishSubscribeChannel.java | 17 ++++- .../xml/PublishSubscribeChannelParser.java | 6 +- .../dispatcher/BroadcastingDispatcher.java | 33 ++++++++-- .../config/xml/spring-integration-3.0.xsd | 9 +++ ...ibersDefaultConfigurationTests-context.xml | 2 +- .../xml/DispatcherMaxSubscribersTests.java | 4 +- .../core/MessagingTemplateTests.java | 65 +++++++++++++++++++ src/reference/docbook/channel.xml | 14 ++++ src/reference/docbook/whats-new.xml | 13 ++++ 9 files changed, 152 insertions(+), 11 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index 992cc2637f..e80dbf70b8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java index 6c4cc255ed..17e39dfa32 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java @@ -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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java index e326f83cba..b17c4cdfd6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java @@ -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 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 { } } + } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd index bde0e093df..591be6ee30 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd @@ -449,6 +449,15 @@ + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests-context.xml index 8e770337a2..7d4a9c1182 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests-context.xml @@ -29,6 +29,6 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersTests.java index 2fa8b66694..79c7701824 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersTests.java @@ -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()); } } \ No newline at end of file diff --git a/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java index 9c3ce73aae..0d084a45fb 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java @@ -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(); diff --git a/src/reference/docbook/channel.xml b/src/reference/docbook/channel.xml index 5340954496..0ee28aa35a 100644 --- a/src/reference/docbook/channel.xml +++ b/src/reference/docbook/channel.xml @@ -81,6 +81,20 @@ must be a MessageHandler itself, and the subscriber's handleMessage(Message) method will be invoked in turn. + + Prior to version 3.0, invoking the send method on a PublishSubscribeChannel that + had no subscribers returned false. When used in conjunction with a + MessagingTemplate, a MessageDeliveryException + was thrown. Starting with version 3.0, the behavior has changed such that a send is + always considered successful if at least the minimum subscribers are present (and successfully + handle the message). This behavior can be modified by setting the minSubscribers + property, which defaults to 0. + + + If a TaskExecutor is used, only the presence of the correct number + of subscribers is used for this determination, because the actual handling of the message + is performed asynchronously. +
QueueChannel diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index fa41e2cc1b..9f3c436bf9 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -363,5 +363,18 @@ For more information see .
+
+ PublishSubscribeChannel Behavior + + Previously, sending to a <publish-subscribe-channel/> that had + no subscribers would return a false result. If used in conjunction with + a MessagingTemplate, this would result in an exception being thrown. + Now, the PublishSubscribeChannel has a property + minSubscribers (default 0). If the message is sent to at least the minimum + number of subscribers, the send is deemed to be successful (even if zero). If an application + is expecting to get an exception under these conditions, set the minimum subscribers to at + least 1. + +