diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 86805cc4e1..bb0de3d7ee 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -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() + "'"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java index e11e2977c1..9904a0ae06 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java @@ -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() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java index 950ba69f59..62d815d8ff 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java @@ -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 implements MessageEndpoint { +public class GenericMessageEndpoint implements MessageEndpoint, BeanNameAware { + + private String name; private String inputChannelName; @@ -50,6 +53,18 @@ public class GenericMessageEndpoint implements MessageEndpoint { 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. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java index 931e9ed253..edbb8891f0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java @@ -24,6 +24,10 @@ package org.springframework.integration.message; */ public interface MessageReceiver { + String getName(); + + void setName(String name); + void messageReceived(Message message); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java index a7114327a1..b520f3c839 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java @@ -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++) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java index ee978b2cc2..24f88a43f1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java @@ -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(); 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 470d5c4a6a..6f8cf950a6 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 @@ -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);