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

Further refactoring of subscription activation and dispatcher creation. Simplifying to provide a foundation for broadcasting channel dispatchers.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-10 12:47:19 +00:00
parent 58b3a7a82c
commit aca1f28e07
7 changed files with 53 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -149,7 +149,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
this.activateSubscription(subscription);
if (logger.isInfoEnabled()) {
logger.info("activated subscription to channel '" + subscription.getChannel() +
"' for endpoint '" + subscription.getEndpoint() + "'");
"' for endpoint '" + subscription.getReceiver() + "'");
}
}
}
@@ -187,6 +187,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
public void registerEndpoint(String name, MessageEndpoint<?> endpoint) {
Assert.notNull(name, "'name' must not be null");
Assert.notNull(endpoint, "'endpoint' must not be null");
endpoint.setName(name);
this.endpoints.put(name, endpoint);
if (logger.isInfoEnabled()) {
logger.info("registered endpoint '" + name + "'");
@@ -195,7 +196,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
if (endpoint.getInputChannelName() != null && endpoint.getConsumerPolicy() != null) {
Subscription subscription = new Subscription();
subscription.setChannel(endpoint.getInputChannelName());
subscription.setEndpoint(name);
subscription.setReceiver(name);
subscription.setPolicy(endpoint.getConsumerPolicy());
this.activateSubscription(subscription);
}
@@ -229,27 +230,23 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
this.targetAdapters.put(name, targetAdapter);
MessageChannel channel = adapter.getChannel();
ConsumerPolicy policy = adapter.getConsumerPolicy();
MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy);
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever);
dispatcher.setRejectionLimit(policy.getRejectionLimit());
dispatcher.setRetryInterval(policy.getRetryInterval());
MessageReceivingExecutor executor = new MessageReceivingExecutor(adapter, policy.getConcurrency(), policy.getMaxConcurrency());
dispatcher.addExecutor(executor);
this.addLifecycleComponent(name + "-executor", executor);
this.addDispatcherTask(new DispatcherTask(dispatcher, policy));
if (logger.isInfoEnabled()) {
logger.info("registered target adapter '" + name + "'");
}
this.doActivate(channel, adapter, policy);
}
if (logger.isInfoEnabled()) {
logger.info("registered target adapter '" + name + "'");
}
}
public void activateSubscription(Subscription subscription) {
String channelName = subscription.getChannel();
String endpointName = subscription.getEndpoint();
String receiverName = subscription.getReceiver();
ConsumerPolicy policy = subscription.getPolicy();
MessageEndpoint<?> endpoint = this.endpoints.get(endpointName);
if (endpoint == null) {
throw new MessagingException("Cannot activate subscription, unknown endpoint '" + endpointName + "'");
MessageReceiver<?> receiver = this.endpoints.get(receiverName);
if (receiver == null) {
receiver = this.targetAdapters.get(receiverName);
if (receiver == null) {
throw new MessagingException("Cannot activate subscription, unknown receiver '" + receiverName + "'");
}
}
MessageChannel channel = this.lookupChannel(channelName);
if (channel == null) {
@@ -263,13 +260,17 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
channel = new SimpleChannel();
this.registerChannel(channelName, channel);
}
this.doActivate(channel, receiver, policy);
if (logger.isInfoEnabled()) {
logger.info("activated subscription to channel '" + channelName +
"' for endpoint '" + endpointName + "'");
"' for receiver '" + receiverName + "'");
}
MessageReceivingExecutor executor = new MessageReceivingExecutor(endpoint, policy.getConcurrency(), policy.getMaxConcurrency());
this.receiverExecutors.put(endpoint, executor);
this.lifecycleComponents.put(endpointName + "-executor", executor);
}
private void doActivate(MessageChannel channel, MessageReceiver<?> receiver, ConsumerPolicy policy) {
MessageReceivingExecutor executor = new MessageReceivingExecutor(receiver, policy.getConcurrency(), policy.getMaxConcurrency());
this.receiverExecutors.put(receiver, executor);
this.addLifecycleComponent(receiver.getName() + "-executor", executor);
MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy);
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever);
dispatcher.setRejectionLimit(policy.getRejectionLimit());
@@ -282,7 +283,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
this.addDispatcherTask(dispatcherTask);
if (this.logger.isInfoEnabled()) {
logger.info("registered dispatcher task: channel='" +
channelName + "' receiver='" + endpointName + "'");
channel.getName() + "' receiver='" + receiver.getName() + "'");
}
}

10
spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java
View File

@@ -25,7 +25,7 @@ public class Subscription {
private String channel;
private String endpoint;
private String receiver;
private ConsumerPolicy policy = new ConsumerPolicy();
@@ -38,12 +38,12 @@ public class Subscription {
this.channel = channel;
}
public String getEndpoint() {
return this.endpoint;
public String getReceiver() {
return this.receiver;
}
public void setEndpoint(String endpoint) {
this.endpoint = endpoint;
public void setReceiver(String receiver) {
this.receiver = receiver;
}
public ConsumerPolicy getPolicy() {

17
spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.endpoint;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.bus.ConsumerPolicy;
@@ -37,7 +38,9 @@ import org.springframework.integration.message.Message;
*
* @author Mark Fisher
*/
public class GenericMessageEndpoint<T> implements MessageEndpoint<T> {
public class GenericMessageEndpoint<T> implements MessageEndpoint<T>, BeanNameAware {
private String name;
private String inputChannelName;
@@ -50,6 +53,18 @@ public class GenericMessageEndpoint<T> implements MessageEndpoint<T> {
private ConsumerPolicy consumerPolicy = new ConsumerPolicy();
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
public void setBeanName(String beanName) {
this.setName(beanName);
}
/**
* Set the name of the channel from which this endpoint receives messages.
*/

4
spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java
View File

@@ -24,6 +24,10 @@ package org.springframework.integration.message;
*/
public interface MessageReceiver<T> {
String getName();
void setName(String name);
void messageReceived(Message<T> message);
}

4
spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java
View File

@@ -58,7 +58,7 @@ public class FixedDelayConsumerTests {
policy.setPeriod(10);
Subscription subscription = new Subscription();
subscription.setChannel("testChannel");
subscription.setEndpoint("testEndpoint");
subscription.setReceiver("testEndpoint");
subscription.setPolicy(policy);
bus.activateSubscription(subscription);
bus.start();
@@ -92,7 +92,7 @@ public class FixedDelayConsumerTests {
policy.setPeriod(10);
Subscription subscription = new Subscription();
subscription.setChannel("testChannel");
subscription.setEndpoint("testEndpoint");
subscription.setReceiver("testEndpoint");
subscription.setPolicy(policy);
bus.activateSubscription(subscription);
for (int i = 0; i < messagesToSend; i++) {

4
spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java
View File

@@ -56,7 +56,7 @@ public class FixedRateConsumerTests {
policy.setPeriod(10);
Subscription subscription = new Subscription();
subscription.setChannel("testChannel");
subscription.setEndpoint("testEndpoint");
subscription.setReceiver("testEndpoint");
subscription.setPolicy(policy);
bus.activateSubscription(subscription);
bus.start();
@@ -90,7 +90,7 @@ public class FixedRateConsumerTests {
policy.setPeriod(20);
Subscription subscription = new Subscription();
subscription.setChannel("testChannel");
subscription.setEndpoint("testEndpoint");
subscription.setReceiver("testEndpoint");
subscription.setPolicy(policy);
bus.activateSubscription(subscription);
bus.start();

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

@@ -81,7 +81,7 @@ public class MessageBusTests {
ConsumerPolicy policy = new ConsumerPolicy();
Subscription subscription = new Subscription();
subscription.setChannel("sourceChannel");
subscription.setEndpoint("endpoint");
subscription.setReceiver("endpoint");
subscription.setPolicy(policy);
bus.activateSubscription(subscription);
Message<?> result = targetChannel.receive(100);