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

Renamed 'broadcaster' property to 'publishSubscribe'.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-16 03:39:52 +00:00
parent 45d1332659
commit 2121dd5f7f
7 changed files with 30 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java
View File

@@ -36,11 +36,11 @@ public interface MessageChannel {
String getName(); String getName();
/** /**
* Return whether this channel has been designated as a broadcaster. If so, * Return whether this channel has been designated as a publish-subscribe channel.
* any dispatcher retrieving messages from this channel should send each * If so, any dispatcher retrieving messages from this channel should send each
* message to each of its receivers. * message to each of the dispatcher's handlers.
*/ */
boolean isBroadcaster(); boolean isPublishSubscribe();
/** /**
* Send a message, blocking indefinitely if necessary. * Send a message, blocking indefinitely if necessary.

10
spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java
View File

@@ -36,7 +36,7 @@ public class SimpleChannel implements MessageChannel, BeanNameAware {
private String name; private String name;
private boolean broadcaster = false; private boolean publishSubscribe = false;
private BlockingQueue<Message<?>> queue; private BlockingQueue<Message<?>> queue;
@@ -70,12 +70,12 @@ public class SimpleChannel implements MessageChannel, BeanNameAware {
return this.name; return this.name;
} }
public boolean isBroadcaster() { public boolean isPublishSubscribe() {
return this.broadcaster; return this.publishSubscribe;
} }
public void setBroadcaster(boolean broadcaster) { public void setPublishSubscribe(boolean publishSubscribe) {
this.broadcaster = broadcaster; this.publishSubscribe = publishSubscribe;
} }
/** /**

4
spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java
View File

@@ -39,6 +39,8 @@ public class ChannelParser implements BeanDefinitionParser {
private static final String PUBLISH_SUBSCRIBE_ATTRIBUTE = "publish-subscribe"; private static final String PUBLISH_SUBSCRIBE_ATTRIBUTE = "publish-subscribe";
private static final String PUBLISH_SUBSCRIBE_PROPERTY = "publishSubscribe";
public BeanDefinition parse(Element element, ParserContext parserContext) { public BeanDefinition parse(Element element, ParserContext parserContext) {
RootBeanDefinition channelDef = new RootBeanDefinition(SimpleChannel.class); RootBeanDefinition channelDef = new RootBeanDefinition(SimpleChannel.class);
@@ -49,7 +51,7 @@ public class ChannelParser implements BeanDefinitionParser {
} }
String isPublishSubscribe = element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE); String isPublishSubscribe = element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE);
if ("true".equals(isPublishSubscribe)) { if ("true".equals(isPublishSubscribe)) {
channelDef.getPropertyValues().addPropertyValue("broadcaster", Boolean.TRUE); channelDef.getPropertyValues().addPropertyValue(PUBLISH_SUBSCRIBE_PROPERTY, Boolean.TRUE);
} }
String beanName = element.getAttribute(ID_ATTRIBUTE); String beanName = element.getAttribute(ID_ATTRIBUTE);
if (!StringUtils.hasText(beanName)) { if (!StringUtils.hasText(beanName)) {

9
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java
View File

@@ -121,10 +121,11 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
if (schedule == null) { if (schedule == null) {
schedule = this.defaultSchedule; schedule = this.defaultSchedule;
} }
else if (this.channel.isBroadcaster()) { else if (this.channel.isPublishSubscribe()) {
if (logger.isInfoEnabled()) { if (logger.isInfoEnabled()) {
logger.info("This dispatcher's channel is a broadcaster, and therefore all handlers are " + logger.info("This dispatcher broadcasts messages for a publish-subscribe channel. " +
"scheduled with its 'defaultSchedule'. The provided schedule will be ignored."); "Therefore all handlers are scheduled with its 'defaultSchedule', " +
"and the provided schedule will be ignored.");
} }
schedule = this.defaultSchedule; schedule = this.defaultSchedule;
} }
@@ -167,7 +168,7 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
task.setSchedule(schedule); task.setSchedule(schedule);
task.setRejectionLimit(this.rejectionLimit); task.setRejectionLimit(this.rejectionLimit);
task.setRetryInterval(this.retryInterval); task.setRetryInterval(this.retryInterval);
task.setBroadcast(channel.isBroadcaster()); task.setPublishSubscribe(channel.isPublishSubscribe());
task.setShouldFailOnRejectionLimit(this.shouldFailOnRejectionLimit); task.setShouldFailOnRejectionLimit(this.shouldFailOnRejectionLimit);
for (MessageHandler handler : handlers) { for (MessageHandler handler : handlers) {
if (handler instanceof Lifecycle) { if (handler instanceof Lifecycle) {

12
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java
View File

@@ -46,7 +46,7 @@ public class DispatcherTask implements MessagingTask {
private Log logger = LogFactory.getLog(this.getClass()); private Log logger = LogFactory.getLog(this.getClass());
private boolean broadcast = false; private boolean publishSubscribe = false;
private Schedule schedule; private Schedule schedule;
@@ -64,20 +64,20 @@ public class DispatcherTask implements MessagingTask {
public DispatcherTask(MessageChannel channel) { public DispatcherTask(MessageChannel channel) {
Assert.notNull(channel, "'channel' must not be null"); Assert.notNull(channel, "'channel' must not be null");
this.retriever = new ChannelPollingMessageRetriever(channel); this.retriever = new ChannelPollingMessageRetriever(channel);
this.broadcast = channel.isBroadcaster(); this.publishSubscribe = channel.isPublishSubscribe();
} }
public DispatcherTask(MessageRetriever retriever) { public DispatcherTask(MessageRetriever retriever) {
Assert.notNull(retriever, "'retriever' must not be null"); Assert.notNull(retriever, "'retriever' must not be null");
if (retriever instanceof ChannelPollingMessageRetriever) { if (retriever instanceof ChannelPollingMessageRetriever) {
this.broadcast = ((ChannelPollingMessageRetriever) retriever).getChannel().isBroadcaster(); this.publishSubscribe = ((ChannelPollingMessageRetriever) retriever).getChannel().isPublishSubscribe();
} }
this.retriever = retriever; this.retriever = retriever;
} }
public void setBroadcast(boolean broadcast) { public void setPublishSubscribe(boolean publishSubscribe) {
this.broadcast = broadcast; this.publishSubscribe = publishSubscribe;
} }
public void setSchedule(Schedule schedule) { public void setSchedule(Schedule schedule) {
@@ -161,7 +161,7 @@ public class DispatcherTask implements MessagingTask {
MessageHandler handler = iter.next(); MessageHandler handler = iter.next();
try { try {
handler.handle(message); handler.handle(message);
if (!this.broadcast) { if (!this.publishSubscribe) {
return true; return true;
} }
iter.remove(); iter.remove();

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

@@ -72,7 +72,7 @@ public class DefaultMessageDispatcherTests {
TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
SimpleChannel channel = new SimpleChannel(); SimpleChannel channel = new SimpleChannel();
channel.setBroadcaster(true); channel.setPublishSubscribe(true);
channel.send(new StringMessage(1, "test")); channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1));
@@ -117,7 +117,7 @@ public class DefaultMessageDispatcherTests {
TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch);
SimpleChannel channel = new SimpleChannel(); SimpleChannel channel = new SimpleChannel();
channel.setBroadcaster(true); channel.setPublishSubscribe(true);
channel.send(new StringMessage(1, "test")); channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1));
@@ -151,7 +151,7 @@ public class DefaultMessageDispatcherTests {
TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch);
SimpleChannel channel = new SimpleChannel(); SimpleChannel channel = new SimpleChannel();
channel.setBroadcaster(true); channel.setPublishSubscribe(true);
channel.send(new StringMessage(1, "test")); channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.setRejectionLimit(2); dispatcher.setRejectionLimit(2);
@@ -186,7 +186,7 @@ public class DefaultMessageDispatcherTests {
TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch);
SimpleChannel channel = new SimpleChannel(); SimpleChannel channel = new SimpleChannel();
channel.setBroadcaster(true); channel.setPublishSubscribe(true);
channel.send(new StringMessage(1, "test")); channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.setRejectionLimit(2); dispatcher.setRejectionLimit(2);
@@ -343,7 +343,7 @@ public class DefaultMessageDispatcherTests {
TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
SimpleChannel channel = new SimpleChannel(); SimpleChannel channel = new SimpleChannel();
channel.setBroadcaster(true); channel.setPublishSubscribe(true);
channel.send(new StringMessage(1, "test")); channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.setRejectionLimit(5); dispatcher.setRejectionLimit(5);
@@ -452,7 +452,7 @@ public class DefaultMessageDispatcherTests {
TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
SimpleChannel channel = new SimpleChannel(); SimpleChannel channel = new SimpleChannel();
channel.setBroadcaster(true); channel.setPublishSubscribe(true);
channel.send(new StringMessage(1, "test")); channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1); PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1);

2
spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
View File

@@ -128,7 +128,7 @@ public class MessageBusTests {
@Test @Test
public void testBothHandlersReceivePublishSubscribeMessage() { public void testBothHandlersReceivePublishSubscribeMessage() {
SimpleChannel inputChannel = new SimpleChannel(); SimpleChannel inputChannel = new SimpleChannel();
inputChannel.setBroadcaster(true); inputChannel.setPublishSubscribe(true);
SimpleChannel outputChannel1 = new SimpleChannel(); SimpleChannel outputChannel1 = new SimpleChannel();
SimpleChannel outputChannel2 = new SimpleChannel(); SimpleChannel outputChannel2 = new SimpleChannel();
MessageHandler handler1 = new MessageHandler() { MessageHandler handler1 = new MessageHandler() {