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

Subscribable is now SubscribableChannel, and MessageDispatcher no longer implements Subscribable.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-02 17:50:18 +00:00
parent 3f9230d73b
commit e5a8972a97
11 changed files with 124 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java
View File

@@ -19,7 +19,7 @@ package org.springframework.integration.channel;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.message.SubscribableChannel;
import org.springframework.util.Assert;
/**
@@ -28,7 +28,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class AbstractSubscribableChannel<T extends MessageDispatcher> extends AbstractMessageChannel implements Subscribable {
public class AbstractSubscribableChannel<T extends MessageDispatcher> extends AbstractMessageChannel implements SubscribableChannel {
private final T dispatcher;
@@ -44,11 +44,11 @@ public class AbstractSubscribableChannel<T extends MessageDispatcher> extends Ab
}
public boolean subscribe(MessageConsumer consumer) {
return this.dispatcher.subscribe(consumer);
return this.dispatcher.addConsumer(consumer);
}
public boolean unsubscribe(MessageConsumer consumer) {
return this.dispatcher.unsubscribe(consumer);
return this.dispatcher.removeConsumer(consumer);
}
@Override

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

@@ -36,17 +36,17 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
protected final Log logger = LogFactory.getLog(this.getClass());
protected final Set<MessageConsumer> subscribers = new CopyOnWriteArraySet<MessageConsumer>();
protected final Set<MessageConsumer> consumers = new CopyOnWriteArraySet<MessageConsumer>();
private volatile TaskExecutor taskExecutor;
public boolean subscribe(MessageConsumer consumer) {
return this.subscribers.add(consumer);
public boolean addConsumer(MessageConsumer consumer) {
return this.consumers.add(consumer);
}
public boolean unsubscribe(MessageConsumer consumer) {
return this.subscribers.remove(consumer);
public boolean removeConsumer(MessageConsumer consumer) {
return this.consumers.remove(consumer);
}
/**
@@ -63,7 +63,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
}
public String toString() {
return this.getClass().getSimpleName() + " with subscribers: " + this.subscribers;
return this.getClass().getSimpleName() + " with consumers: " + this.consumers;
}
/**
@@ -77,7 +77,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
}
catch (MessageRejectedException e) {
if (logger.isDebugEnabled()) {
logger.debug("Consumer '" + consumer + "' rejected Message, continuing with other subscribers if available.", e);
logger.debug("Consumer '" + consumer + "' rejected Message, continuing with other consumers if available.", e);
}
return false;
}

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

@@ -23,9 +23,9 @@ import org.springframework.integration.message.MessageConsumer;
/**
* A broadcasting dispatcher implementation. It makes a best effort to
* send the message to each of its endpoints. If it fails to send to any
* one endpoints, it will log a warn-level message but continue to send
* to the other endpoints.
* send the message to each of its consumers. If it fails to send to any
* one consumer, it will log a warn-level message but continue to send
* to the other consumers.
*
* @author Mark Fisher
*/
@@ -45,8 +45,8 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
public boolean dispatch(Message<?> message) {
int sequenceNumber = 1;
int sequenceSize = this.subscribers.size();
for (final MessageConsumer consumer : this.subscribers) {
int sequenceSize = this.consumers.size();
for (final MessageConsumer consumer : this.consumers) {
final Message<?> messageToSend = (!this.applySequence) ? message
: MessageBuilder.fromMessage(message)
.setSequenceNumber(sequenceNumber++)

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

@@ -17,14 +17,18 @@
package org.springframework.integration.dispatcher;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.message.MessageConsumer;
/**
* Strategy interface for dispatching messages.
* Strategy interface for dispatching messages to consumers.
*
* @author Mark Fisher
*/
public interface MessageDispatcher extends Subscribable {
public interface MessageDispatcher {
boolean addConsumer(MessageConsumer consumer);
boolean removeConsumer(MessageConsumer consumer);
boolean dispatch(Message<?> message);

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

@@ -23,11 +23,11 @@ import org.springframework.integration.message.MessageRejectedException;
/**
* Basic implementation of {@link MessageDispatcher} that will attempt
* to send a {@link Message} to one of its subscribers. As soon as <em>one</em>
* of the subscribers accepts the Message, the dispatcher will return 'true'.
* to send a {@link Message} to one of its consumers. As soon as <em>one</em>
* of the consumers accepts the Message, the dispatcher will return 'true'.
* <p>
* If the dispatcher has no subscribers, a {@link MessageDeliveryException}
* will be thrown. If all subscribers reject the Message, the dispatcher will
* If the dispatcher has no consumers, a {@link MessageDeliveryException}
* will be thrown. If all consumers reject the Message, the dispatcher will
* throw a MessageRejectedException.
*
* @author Mark Fisher
@@ -35,12 +35,12 @@ import org.springframework.integration.message.MessageRejectedException;
public class SimpleDispatcher extends AbstractDispatcher {
public boolean dispatch(Message<?> message) {
if (this.subscribers.size() == 0) {
if (this.consumers.size() == 0) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
}
int count = 0;
int rejectedExceptionCount = 0;
for (MessageConsumer consumer : this.subscribers) {
for (MessageConsumer consumer : this.consumers) {
count++;
if (this.sendMessageToConsumer(message, consumer)) {
return true;

12
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageConsumingEndpoint.java
View File

@@ -26,7 +26,7 @@ import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.message.SubscribableChannel;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.util.Assert;
@@ -90,7 +90,7 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
if (this.taskExecutor != null) {
this.poller.setTaskExecutor(this.taskExecutor);
}
this.poller.subscribe(this);
this.poller.setConsumer(this);
}
this.initialized = true;
}
@@ -105,8 +105,8 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
this.afterPropertiesSet();
}
Assert.notNull(this.inputChannel, "failed to start endpoint, inputChannel is required");
if (this.inputChannel instanceof Subscribable) {
((Subscribable) inputChannel).subscribe(this);
if (this.inputChannel instanceof SubscribableChannel) {
((SubscribableChannel) inputChannel).subscribe(this);
}
else if (this.inputChannel instanceof PollableChannel) {
Assert.notNull(this.getTaskScheduler(),
@@ -122,8 +122,8 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
if (!this.running) {
return;
}
if (this.inputChannel instanceof Subscribable) {
((Subscribable) inputChannel).unsubscribe(this);
if (this.inputChannel instanceof SubscribableChannel) {
((SubscribableChannel) inputChannel).unsubscribe(this);
}
else if (this.pollerFuture != null) {
this.pollerFuture.cancel(true);

20
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ChannelPoller.java
View File

@@ -17,23 +17,21 @@
package org.springframework.integration.endpoint;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.dispatcher.SimpleDispatcher;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
*/
public class ChannelPoller extends AbstractPoller implements Subscribable {
public class ChannelPoller extends AbstractPoller {
private final PollableChannel channel;
private volatile long receiveTimeout = 1000;
private volatile MessageConsumer consumer;
private final SimpleDispatcher dispatcher = new SimpleDispatcher();
private volatile long receiveTimeout = 1000;
public ChannelPoller(PollableChannel channel, Trigger trigger) {
@@ -51,23 +49,21 @@ public class ChannelPoller extends AbstractPoller implements Subscribable {
this.receiveTimeout = receiveTimeout;
}
public boolean subscribe(MessageConsumer consumer) {
return this.dispatcher.subscribe(consumer);
}
public boolean unsubscribe(MessageConsumer consumer) {
return this.dispatcher.unsubscribe(consumer);
public void setConsumer(MessageConsumer consumer) {
this.consumer = consumer;
}
@Override
protected boolean doPoll() {
Assert.notNull(this.consumer, "consumer must not be null");
Message<?> message = (this.receiveTimeout >= 0)
? this.channel.receive(this.receiveTimeout)
: this.channel.receive();
if (message == null) {
return false;
}
return this.dispatcher.dispatch(message);
this.consumer.onMessage(message);
return true;
}
}

10
org.springframework.integration/src/main/java/org/springframework/integration/message/Subscribable.java → org.springframework.integration/src/main/java/org/springframework/integration/message/SubscribableChannel.java
View File

@@ -16,20 +16,22 @@
package org.springframework.integration.message;
import org.springframework.integration.channel.MessageChannel;
/**
* Interface for any source of messages that accepts subscribers.
* Interface for any MessageChannel implementation that accepts subscribers.
*
* @author Mark Fisher
*/
public interface Subscribable {
public interface SubscribableChannel extends MessageChannel {
/**
* Register a {@link MessageConsumer} as a subscriber to this source.
* Register a {@link MessageConsumer} as a subscriber to this channel.
*/
boolean subscribe(MessageConsumer consumer);
/**
* Remove a {@link MessageConsumer} from the subscribers of this source.
* Remove a {@link MessageConsumer} from the subscribers of this channel.
*/
boolean unsubscribe(MessageConsumer consumer);