From 2121dd5f7f6f1dc3ca214b504f28ced222ec6fce Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 16 Jan 2008 03:39:52 +0000 Subject: [PATCH] Renamed 'broadcaster' property to 'publishSubscribe'. --- .../integration/channel/MessageChannel.java | 8 ++++---- .../integration/channel/SimpleChannel.java | 10 +++++----- .../integration/config/ChannelParser.java | 4 +++- .../dispatcher/DefaultMessageDispatcher.java | 9 +++++---- .../integration/dispatcher/DispatcherTask.java | 12 ++++++------ .../bus/DefaultMessageDispatcherTests.java | 12 ++++++------ .../integration/bus/MessageBusTests.java | 2 +- 7 files changed, 30 insertions(+), 27 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java index 4186ea208b..9f0eeb0f7d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java @@ -36,11 +36,11 @@ public interface MessageChannel { String getName(); /** - * Return whether this channel has been designated as a broadcaster. If so, - * any dispatcher retrieving messages from this channel should send each - * message to each of its receivers. + * Return whether this channel has been designated as a publish-subscribe channel. + * If so, any dispatcher retrieving messages from this channel should send each + * message to each of the dispatcher's handlers. */ - boolean isBroadcaster(); + boolean isPublishSubscribe(); /** * Send a message, blocking indefinitely if necessary. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java index ba7b29ae80..b7aec8a16b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java @@ -36,7 +36,7 @@ public class SimpleChannel implements MessageChannel, BeanNameAware { private String name; - private boolean broadcaster = false; + private boolean publishSubscribe = false; private BlockingQueue> queue; @@ -70,12 +70,12 @@ public class SimpleChannel implements MessageChannel, BeanNameAware { return this.name; } - public boolean isBroadcaster() { - return this.broadcaster; + public boolean isPublishSubscribe() { + return this.publishSubscribe; } - public void setBroadcaster(boolean broadcaster) { - this.broadcaster = broadcaster; + public void setPublishSubscribe(boolean publishSubscribe) { + this.publishSubscribe = publishSubscribe; } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java index 0b96868621..c58bae544a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java @@ -39,6 +39,8 @@ public class ChannelParser implements BeanDefinitionParser { private static final String PUBLISH_SUBSCRIBE_ATTRIBUTE = "publish-subscribe"; + private static final String PUBLISH_SUBSCRIBE_PROPERTY = "publishSubscribe"; + public BeanDefinition parse(Element element, ParserContext parserContext) { RootBeanDefinition channelDef = new RootBeanDefinition(SimpleChannel.class); @@ -49,7 +51,7 @@ public class ChannelParser implements BeanDefinitionParser { } String isPublishSubscribe = element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE); if ("true".equals(isPublishSubscribe)) { - channelDef.getPropertyValues().addPropertyValue("broadcaster", Boolean.TRUE); + channelDef.getPropertyValues().addPropertyValue(PUBLISH_SUBSCRIBE_PROPERTY, Boolean.TRUE); } String beanName = element.getAttribute(ID_ATTRIBUTE); if (!StringUtils.hasText(beanName)) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java index b01c76e797..6d6b399c42 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java @@ -121,10 +121,11 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas if (schedule == null) { schedule = this.defaultSchedule; } - else if (this.channel.isBroadcaster()) { + else if (this.channel.isPublishSubscribe()) { if (logger.isInfoEnabled()) { - logger.info("This dispatcher's channel is a broadcaster, and therefore all handlers are " + - "scheduled with its 'defaultSchedule'. The provided schedule will be ignored."); + logger.info("This dispatcher broadcasts messages for a publish-subscribe channel. " + + "Therefore all handlers are scheduled with its 'defaultSchedule', " + + "and the provided schedule will be ignored."); } schedule = this.defaultSchedule; } @@ -167,7 +168,7 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas task.setSchedule(schedule); task.setRejectionLimit(this.rejectionLimit); task.setRetryInterval(this.retryInterval); - task.setBroadcast(channel.isBroadcaster()); + task.setPublishSubscribe(channel.isPublishSubscribe()); task.setShouldFailOnRejectionLimit(this.shouldFailOnRejectionLimit); for (MessageHandler handler : handlers) { if (handler instanceof Lifecycle) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java index 011fc265a7..f9d92505c3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java @@ -46,7 +46,7 @@ public class DispatcherTask implements MessagingTask { private Log logger = LogFactory.getLog(this.getClass()); - private boolean broadcast = false; + private boolean publishSubscribe = false; private Schedule schedule; @@ -64,20 +64,20 @@ public class DispatcherTask implements MessagingTask { public DispatcherTask(MessageChannel channel) { Assert.notNull(channel, "'channel' must not be null"); this.retriever = new ChannelPollingMessageRetriever(channel); - this.broadcast = channel.isBroadcaster(); + this.publishSubscribe = channel.isPublishSubscribe(); } public DispatcherTask(MessageRetriever retriever) { Assert.notNull(retriever, "'retriever' must not be null"); if (retriever instanceof ChannelPollingMessageRetriever) { - this.broadcast = ((ChannelPollingMessageRetriever) retriever).getChannel().isBroadcaster(); + this.publishSubscribe = ((ChannelPollingMessageRetriever) retriever).getChannel().isPublishSubscribe(); } this.retriever = retriever; } - public void setBroadcast(boolean broadcast) { - this.broadcast = broadcast; + public void setPublishSubscribe(boolean publishSubscribe) { + this.publishSubscribe = publishSubscribe; } public void setSchedule(Schedule schedule) { @@ -161,7 +161,7 @@ public class DispatcherTask implements MessagingTask { MessageHandler handler = iter.next(); try { handler.handle(message); - if (!this.broadcast) { + if (!this.publishSubscribe) { return true; } iter.remove(); 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 bf7adb1bbb..222d39f058 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 @@ -72,7 +72,7 @@ public class DefaultMessageDispatcherTests { TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); SimpleChannel channel = new SimpleChannel(); - channel.setBroadcaster(true); + channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); @@ -117,7 +117,7 @@ public class DefaultMessageDispatcherTests { TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); SimpleChannel channel = new SimpleChannel(); - channel.setBroadcaster(true); + channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); @@ -151,7 +151,7 @@ public class DefaultMessageDispatcherTests { TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); SimpleChannel channel = new SimpleChannel(); - channel.setBroadcaster(true); + channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); @@ -186,7 +186,7 @@ public class DefaultMessageDispatcherTests { TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); SimpleChannel channel = new SimpleChannel(); - channel.setBroadcaster(true); + channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); @@ -343,7 +343,7 @@ public class DefaultMessageDispatcherTests { TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); SimpleChannel channel = new SimpleChannel(); - channel.setBroadcaster(true); + channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(5); @@ -452,7 +452,7 @@ public class DefaultMessageDispatcherTests { TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); SimpleChannel channel = new SimpleChannel(); - channel.setBroadcaster(true); + channel.setPublishSubscribe(true); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 0faa8f3675..338efac7d1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -128,7 +128,7 @@ public class MessageBusTests { @Test public void testBothHandlersReceivePublishSubscribeMessage() { SimpleChannel inputChannel = new SimpleChannel(); - inputChannel.setBroadcaster(true); + inputChannel.setPublishSubscribe(true); SimpleChannel outputChannel1 = new SimpleChannel(); SimpleChannel outputChannel2 = new SimpleChannel(); MessageHandler handler1 = new MessageHandler() {