diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java index 2a74ff2f57..68222ec41c 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java @@ -42,7 +42,8 @@ public class ByteStreamSourceTests { ByteArrayInputStream stream = new ByteArrayInputStream(bytes); MessageChannel channel = new QueueChannel(); ByteStreamSource source = new ByteStreamSource(stream); - SourceEndpoint endpoint = new SourceEndpoint(source, channel); + SourceEndpoint endpoint = new SourceEndpoint(source); + endpoint.setOutputChannel(channel); endpoint.afterPropertiesSet(); endpoint.send(new CommandMessage(new PollCommand())); Message message1 = channel.receive(500); @@ -67,7 +68,8 @@ public class ByteStreamSourceTests { source.setBytesPerMessage(4); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - SourceEndpoint endpoint = new SourceEndpoint(source, channel); + SourceEndpoint endpoint = new SourceEndpoint(source); + endpoint.setOutputChannel(channel); endpoint.afterPropertiesSet(); endpoint.send(new CommandMessage(new PollCommand())); Message message1 = channel.receive(0); @@ -89,7 +91,8 @@ public class ByteStreamSourceTests { source.setShouldTruncate(false); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - SourceEndpoint endpoint = new SourceEndpoint(source, channel); + SourceEndpoint endpoint = new SourceEndpoint(source); + endpoint.setOutputChannel(channel); endpoint.afterPropertiesSet(); endpoint.send(new CommandMessage(new PollCommand())); Message message1 = channel.receive(0); diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java index 1a000078fb..007b24ae91 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java @@ -43,7 +43,8 @@ public class CharacterStreamSourceTests { CharacterStreamSource source = new CharacterStreamSource(reader); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - SourceEndpoint endpoint = new SourceEndpoint(source, channel); + SourceEndpoint endpoint = new SourceEndpoint(source); + endpoint.setOutputChannel(channel); endpoint.afterPropertiesSet(); endpoint.send(new CommandMessage(new PollCommand())); Message message1 = channel.receive(0); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java index 2063caf2f8..88e41c208c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -180,7 +180,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Map endpointBeans = (Map) context .getBeansOfType(MessageEndpoint.class); for (Map.Entry entry : endpointBeans.entrySet()) { - this.registerEndpoint(entry.getKey(), entry.getValue()); + this.registerEndpoint(entry.getValue()); } } @@ -241,34 +241,34 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, public void registerHandler(String name, MessageHandler handler, Subscription subscription) { Assert.notNull(handler, "'handler' must not be null"); HandlerEndpoint endpoint = new HandlerEndpoint(handler); - this.doRegisterEndpoint(name, endpoint, subscription); + endpoint.setName(name); + endpoint.setInputChannelName(subscription.getChannelName()); + endpoint.setSchedule(subscription.getSchedule()); + this.registerEndpoint(endpoint); } public void registerTarget(String name, MessageTarget target, Subscription subscription) { Assert.notNull(target, "'target' must not be null"); TargetEndpoint endpoint = new TargetEndpoint(target); - this.doRegisterEndpoint(name, endpoint, subscription); - } - - private void doRegisterEndpoint(String name, TargetEndpoint endpoint, Subscription subscription) { endpoint.setName(name); - endpoint.setSubscription(subscription); - this.registerEndpoint(name, endpoint); + endpoint.setInputChannelName(subscription.getChannelName()); + endpoint.setSchedule(subscription.getSchedule()); + this.registerEndpoint(endpoint); } - public void registerEndpoint(String name, MessageEndpoint endpoint) { + public void registerEndpoint(MessageEndpoint endpoint) { if (!this.initialized) { this.initialize(); } if (endpoint instanceof ChannelRegistryAware) { ((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry); } - this.endpointRegistry.registerEndpoint(name, endpoint); + this.endpointRegistry.registerEndpoint(endpoint); if (this.isRunning()) { this.activateEndpoint(endpoint); } if (logger.isInfoEnabled()) { - logger.info("registered endpoint '" + name + "'"); + logger.info("registered endpoint '" + endpoint + "'"); } } @@ -310,26 +310,19 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, catch (Exception e) { throw new ConfigurationException("failed to initialize endpoint", e); } - Schedule schedule = null; - Subscription subscription = endpoint.getSubscription(); - if (subscription != null) { - schedule = subscription.getSchedule(); - MessageChannel channel = subscription.getChannel(); - if (channel == null) { - channel = this.lookupOrCreateChannel(subscription.getChannelName()); - } - if (channel != null && channel instanceof Subscribable) { - ((Subscribable) channel).subscribe(endpoint); - if (logger.isInfoEnabled()) { - logger.info("activated subscription to channel '" - + channel.getName() + "' for endpoint '" + endpoint + "'"); - } - return; - } + MessageChannel channel = endpoint.getInputChannel(); + if (channel == null) { + channel = this.lookupOrCreateChannel(endpoint.getInputChannelName()); } - if (schedule == null) { - schedule = endpoint.getSchedule(); + if (channel != null && channel instanceof Subscribable) { + ((Subscribable) channel).subscribe(endpoint); + if (logger.isInfoEnabled()) { + logger.info("activated subscription to channel '" + + channel.getName() + "' for endpoint '" + endpoint + "'"); + } + return; } + Schedule schedule = endpoint.getSchedule(); EndpointTrigger trigger = endpoint.getTrigger(); if (trigger == null) { trigger = new EndpointTrigger(schedule != null ? schedule : this.defaultPollerSchedule); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractTargetEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractTargetEndpointParser.java index 7c302097e6..38a572b445 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractTargetEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractTargetEndpointParser.java @@ -31,7 +31,6 @@ import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.ConfigurationException; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Schedule; -import org.springframework.integration.scheduling.Subscription; import org.springframework.util.StringUtils; /** @@ -43,8 +42,6 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef private static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel"; - private static final String SUBSCRIPTION_PROPERTY = "subscription"; - private static final String SELECTOR_ATTRIBUTE = "selector"; private static final String SELECTOR_PROPERTY = "messageSelector"; @@ -79,7 +76,7 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef @Override protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { this.parseTarget(element, this.getTargetAttributeName(), parserContext, builder); - String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE); + String inputChannelName = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE); Schedule schedule = null; NodeList childNodes = element.getChildNodes(); for (int i = 0; i < childNodes.getLength(); i++) { @@ -97,15 +94,11 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef } } } - if (StringUtils.hasText(inputChannel)) { - RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class); - subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(inputChannel); - if (schedule != null) { - subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(schedule); - } - String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef); - parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName)); - builder.addPropertyReference(SUBSCRIPTION_PROPERTY, subscriptionBeanName); + if (StringUtils.hasText(inputChannelName)) { + builder.addPropertyValue("inputChannelName", inputChannelName); + } + if (schedule != null) { + builder.addPropertyValue("schedule", schedule); } String selectorRef = element.getAttribute(SELECTOR_ATTRIBUTE); if (StringUtils.hasText(selectorRef)) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/SourceEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/SourceEndpointParser.java index 02b1ce30a9..c80591fb49 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/SourceEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/SourceEndpointParser.java @@ -63,12 +63,11 @@ public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser { throw new ConfigurationException("'channel' is required"); } builder.addConstructorArgReference(source); - builder.addConstructorArgReference(output); - Element scheduleElement = this.getScheduleElement(element); - if (scheduleElement == null) { - throw new ConfigurationException("The sub-element is required for a ."); + builder.addPropertyValue("outputChannelName", output); + Element scheduleElement = DomUtils.getChildElementByTagName(element, "schedule"); + if (scheduleElement != null) { + builder.addPropertyValue("schedule", this.parseSchedule(scheduleElement)); } - builder.addPropertyValue("schedule", this.parseSchedule(scheduleElement)); Element interceptorsElement = DomUtils.getChildElementByTagName(element, "interceptors"); if (interceptorsElement != null) { EndpointInterceptorParser parser = new EndpointInterceptorParser(); @@ -90,8 +89,4 @@ public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser { return schedule; } - private Element getScheduleElement(Element element) { - return DomUtils.getChildElementByTagName(element, "schedule"); - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java index 47e9218237..dc79ade218 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java @@ -140,7 +140,8 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class); Subscription subscription = this.createSubscription(bean, beanName, endpointAnnotation, polledAnnotation); if (subscription != null) { - endpoint.setSubscription(subscription); + endpoint.setSchedule(subscription.getSchedule()); + endpoint.setInputChannelName(subscription.getChannelName()); } Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Concurrency.class); if (concurrencyAnnotation != null) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java index 485212f385..6b36ab3a9f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java @@ -84,7 +84,8 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init org.springframework.integration.endpoint.MessageEndpoint endpoint = postProcessor.createEndpoint(bean, beanName, beanClass, endpointAnnotation); if (endpoint != null) { - this.messageBus.registerEndpoint(beanName + "." + entry.getKey().getSimpleName() + ".endpoint", endpoint); + endpoint.setName(beanName + "." + entry.getKey().getSimpleName() + ".endpoint"); + this.messageBus.registerEndpoint(endpoint); } } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java index a5349acd12..5f80e1082d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java @@ -61,6 +61,7 @@ public class SourceAnnotationPostProcessor extends AbstractAnnotationMethodPostP public MessageEndpoint createEndpoint(Object bean, String beanName, Class originalBeanClass, org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) { + SourceEndpoint endpoint = new SourceEndpoint((MessageSource) bean); Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class); int period = polledAnnotation.period(); long initialDelay = polledAnnotation.initialDelay(); @@ -68,15 +69,16 @@ public class SourceAnnotationPostProcessor extends AbstractAnnotationMethodPostP PollingSchedule schedule = new PollingSchedule(period); schedule.setInitialDelay(initialDelay); schedule.setFixedRate(fixedRate); - String outputChannelName = endpointAnnotation.output(); - MessageChannel outputChannel = (StringUtils.hasText(outputChannelName)) ? - this.getMessageBus().lookupChannel(outputChannelName) : null; - if (outputChannel == null) { - outputChannel = new DirectChannel(); - this.getMessageBus().registerChannel(beanName + ".output", outputChannel); - } - SourceEndpoint endpoint = new SourceEndpoint((MessageSource) bean, outputChannel); endpoint.setSchedule(schedule); + String outputChannelName = endpointAnnotation.output(); + if (!StringUtils.hasText(outputChannelName)) { + MessageChannel outputChannel = new DirectChannel(); + this.getMessageBus().registerChannel(beanName + ".output", outputChannel); + endpoint.setOutputChannel(outputChannel); + } + else { + endpoint.setOutputChannelName(outputChannelName); + } return endpoint; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java index 4fdfa5079a..af2b3ab216 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java @@ -64,7 +64,8 @@ public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostP TargetEndpoint endpoint = new TargetEndpoint((MessageTarget) bean); Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class); Subscription subscription = this.createSubscription(bean, beanName, endpointAnnotation, polledAnnotation); - endpoint.setSubscription(subscription); + endpoint.setSchedule(subscription.getSchedule()); + endpoint.setInputChannelName(subscription.getChannelName()); Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Concurrency.class); if (concurrencyAnnotation != null) { ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(concurrencyAnnotation.coreSize(), diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index e2338de60d..1e317a458a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -33,7 +33,6 @@ import org.springframework.integration.handler.MessageHandlerNotRunningException import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.scheduling.Schedule; -import org.springframework.integration.scheduling.Subscription; /** * Base class for {@link MessageEndpoint} implementations. @@ -46,22 +45,26 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware private volatile String name; + private volatile String inputChannelName; + + private MessageChannel inputChannel; + + private volatile String outputChannelName; + + private MessageChannel outputChannel; + private final List interceptors = new ArrayList(); - private volatile boolean autoStartup = true; - - private volatile boolean running; - private volatile Schedule schedule; private volatile EndpointTrigger trigger; - private volatile Subscription subscription; - - private volatile String outputChannelName; - private volatile ChannelRegistry channelRegistry; + private volatile boolean autoStartup = true; + + private volatile boolean running; + private final Object lifecycleMonitor = new Object(); @@ -93,12 +96,25 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware return this.trigger; } - public Subscription getSubscription() { - return this.subscription; + public void setInputChannelName(String inputChannelName) { + this.inputChannelName = inputChannelName; } - public void setSubscription(Subscription subscription) { - this.subscription = subscription; + public String getInputChannelName() { + return this.inputChannelName; + } + + public void setInputChannel(MessageChannel channel) { + this.inputChannel = channel; + this.inputChannelName = channel.getName(); + } + + public MessageChannel getInputChannel() { + if (this.inputChannel == null && + (this.inputChannelName != null && this.channelRegistry != null)) { + this.inputChannel = this.channelRegistry.lookupChannel(this.inputChannelName); + } + return this.inputChannel; } /** @@ -109,17 +125,23 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware this.outputChannelName = outputChannelName; } - public MessageChannel getOutputChannel() { - if (this.outputChannelName != null && this.channelRegistry != null) { - return this.channelRegistry.lookupChannel(this.outputChannelName); - } - return null; - } - public String getOutputChannelName() { return this.outputChannelName; } + public void setOutputChannel(MessageChannel outputChannel) { + this.outputChannel = outputChannel; + this.outputChannelName = outputChannel.getName(); + } + + public MessageChannel getOutputChannel() { + if (this.outputChannel == null && + (this.outputChannelName != null && this.channelRegistry != null)) { + this.outputChannel = this.channelRegistry.lookupChannel(this.outputChannelName); + } + return this.outputChannel; + } + /** * Set the channel registry to use for looking up channels by name. */ diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpointRegistry.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpointRegistry.java index 797f52141a..ed0d7de0c4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpointRegistry.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpointRegistry.java @@ -36,10 +36,9 @@ public class DefaultEndpointRegistry implements EndpointRegistry { return this.endpoints.get(endpointName); } - public void registerEndpoint(String name, MessageEndpoint endpoint) { - Assert.notNull(name, "'name' must not be null"); + public void registerEndpoint(MessageEndpoint endpoint) { Assert.notNull(endpoint, "'endpoint' must not be null"); - this.endpoints.put(name, endpoint); + this.endpoints.put(endpoint.getName(), endpoint); } public MessageEndpoint unregisterEndpoint(String name) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java index 081dcaba76..1b65c0dcd4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java @@ -25,7 +25,7 @@ import java.util.Set; */ public interface EndpointRegistry { - void registerEndpoint(String name, MessageEndpoint endpoint); + void registerEndpoint(MessageEndpoint endpoint); MessageEndpoint unregisterEndpoint(String name); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index 8c9a3227c8..d12d0656fb 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -21,7 +21,6 @@ import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.scheduling.Schedule; -import org.springframework.integration.scheduling.Subscription; /** * Base interface for message endpoints. @@ -30,13 +29,17 @@ import org.springframework.integration.scheduling.Subscription; */ public interface MessageEndpoint extends MessageTarget, ChannelRegistryAware, InitializingBean { + void setName(String name); + String getName(); Schedule getSchedule(); EndpointTrigger getTrigger(); - Subscription getSubscription(); + String getInputChannelName(); + + MessageChannel getInputChannel(); String getOutputChannelName(); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java index ed5b50e992..89c1ae0e3e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java @@ -16,14 +16,13 @@ package org.springframework.integration.endpoint; -import org.springframework.integration.channel.DispatcherPolicy; +import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.dispatcher.SimpleDispatcher; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.PollCommand; import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.PollCommand; import org.springframework.util.Assert; /** @@ -36,14 +35,10 @@ public class SourceEndpoint extends AbstractEndpoint { private final MessageSource source; - private final SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy()); - - public SourceEndpoint(MessageSource source, MessageChannel channel) { + public SourceEndpoint(MessageSource source) { Assert.notNull(source, "source must not be null"); - Assert.notNull(channel, "channel must not be null"); this.source = source; - this.dispatcher.addTarget(channel); } @@ -52,11 +47,15 @@ public class SourceEndpoint extends AbstractEndpoint { } public final boolean doInvoke(Message pollCommandMessage) { + if (this.getOutputChannel() == null) { + throw new ConfigurationException( + "no output channel has been configured for source endpoint '" + this.getName() + "'"); + } Message message = this.source.receive(); if (message == null) { return false; } - boolean sent = this.dispatcher.send(message); + boolean sent = this.getOutputChannel().send(message); if (this.source instanceof MessageDeliveryAware) { if (sent) { ((MessageDeliveryAware) this.source).onSend(message); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java index ce91d01062..e5814595e1 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java @@ -22,7 +22,6 @@ import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.PollCommand; import org.springframework.integration.message.selector.MessageSelector; -import org.springframework.integration.scheduling.Subscription; import org.springframework.util.Assert; /** @@ -34,8 +33,6 @@ public class TargetEndpoint extends AbstractEndpoint { private volatile MessageTarget target; - private volatile Subscription subscription; - private volatile MessageSelector selector; private volatile boolean initialized; @@ -65,14 +62,6 @@ public class TargetEndpoint extends AbstractEndpoint { this.selector = selector; } - public Subscription getSubscription() { - return this.subscription; - } - - public void setSubscription(Subscription subscription) { - this.subscription = subscription; - } - protected void initialize() { synchronized (this.initializationMonitor) { if (this.initialized) { @@ -88,10 +77,7 @@ public class TargetEndpoint extends AbstractEndpoint { @Override protected final boolean doInvoke(Message message) { if (message.getPayload() instanceof PollCommand) { - MessageChannel channel = this.getSubscription().getChannel(); - if (channel == null && this.getSubscription().getChannelName() != null) { - channel = this.getChannelRegistry().lookupChannel(this.getSubscription().getChannelName()); - } + MessageChannel channel = this.getInputChannel(); if (channel != null) { Message receivedMessage = channel.receive(5000); if (receivedMessage != null) { @@ -99,8 +85,7 @@ public class TargetEndpoint extends AbstractEndpoint { } } else if (logger.isDebugEnabled()) { - logger.debug("TargetEndpoint unable to resolve channel '" - + this.getSubscription().getChannelName() + "'"); + logger.debug("TargetEndpoint unable to resolve channel '" + this.getInputChannelName() + "'"); } return false; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java index ae2095e70c..fbf37d19f3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java @@ -32,7 +32,6 @@ import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.selector.MessageSelector; -import org.springframework.integration.scheduling.Subscription; /** * A template that facilitates the implementation of request-reply usage @@ -215,8 +214,9 @@ public class RequestReplyTemplate implements MessageBusAware { } ReplyMessageCorrelator correlator = new ReplyMessageCorrelator(10); HandlerEndpoint endpoint = new HandlerEndpoint(correlator); - endpoint.setSubscription(new Subscription(this.replyChannel)); - this.endpointRegistry.registerEndpoint("internal.correlator." + this, endpoint); + endpoint.setInputChannel(this.replyChannel); + endpoint.setName("internal.correlator." + this); + this.endpointRegistry.registerEndpoint(endpoint); this.replyMessageCorrelator = correlator; } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java index 473aa96b05..3042de94d3 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java @@ -33,7 +33,6 @@ import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.StringMessage; -import org.springframework.integration.scheduling.Subscription; /** * @author Mark Fisher @@ -57,9 +56,10 @@ public class DirectChannelSubscriptionTests { @Test public void testSendAndReceiveForRegisteredEndpoint() { HandlerEndpoint endpoint = new HandlerEndpoint(new TestHandler()); - endpoint.setSubscription(new Subscription("sourceChannel")); + endpoint.setInputChannelName("sourceChannel"); endpoint.setOutputChannelName("targetChannel"); - bus.registerEndpoint("testEndpoint", endpoint); + endpoint.setName("testEndpoint"); + bus.registerEndpoint(endpoint); bus.start(); this.sourceChannel.send(new StringMessage("foo")); Message response = this.targetChannel.receive(); @@ -89,9 +89,10 @@ public class DirectChannelSubscriptionTests { throw new RuntimeException("intentional test failure"); } }); - endpoint.setSubscription(new Subscription("sourceChannel")); + endpoint.setInputChannelName("sourceChannel"); endpoint.setOutputChannelName("targetChannel"); - bus.registerEndpoint("testEndpoint", endpoint); + endpoint.setName("testEndpoint"); + bus.registerEndpoint(endpoint); bus.start(); this.sourceChannel.send(new StringMessage("foo")); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 20df684935..247501afe6 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -28,7 +28,6 @@ import org.junit.Test; import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; @@ -169,9 +168,11 @@ public class MessageBusTests { public void testErrorChannelWithFailedDispatch() throws InterruptedException { MessageBus bus = new MessageBus(); CountDownLatch latch = new CountDownLatch(1); - SourceEndpoint sourceEndpoint = new SourceEndpoint(new FailingSource(latch), new QueueChannel()); + SourceEndpoint sourceEndpoint = new SourceEndpoint(new FailingSource(latch)); + sourceEndpoint.setOutputChannel(new QueueChannel()); sourceEndpoint.setSchedule(new PollingSchedule(1000)); - bus.registerEndpoint("testEndpoint", sourceEndpoint); + sourceEndpoint.setName("testEndpoint"); + bus.registerEndpoint(sourceEndpoint); bus.start(); latch.await(2000, TimeUnit.MILLISECONDS); Message message = bus.getErrorChannel().receive(100); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml index 4f3328ee31..4577d9ed87 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml @@ -12,11 +12,7 @@ - - - - - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java index 0f12c8286d..6a5ad0e5e5 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java @@ -361,7 +361,7 @@ public class MessagingAnnotationPostProcessorTests { AnnotatedEndpointWithPolledAnnotation endpoint = new AnnotatedEndpointWithPolledAnnotation(); postProcessor.postProcessAfterInitialization(endpoint, "testBean"); HandlerEndpoint processedEndpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean.MessageHandler.endpoint"); - Schedule schedule = processedEndpoint.getSubscription().getSchedule(); + Schedule schedule = processedEndpoint.getSchedule(); assertEquals(PollingSchedule.class, schedule.getClass()); PollingSchedule pollingSchedule = (PollingSchedule) schedule; assertEquals(1234, pollingSchedule.getPeriod()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java index 9be2c154f5..41a4cb5ba1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java @@ -40,7 +40,8 @@ public class SourceEndpointTests { public void testPolledSourceSendsToChannel() { TestSource source = new TestSource("testing", 1); QueueChannel channel = new QueueChannel(); - SourceEndpoint endpoint = new SourceEndpoint(source, channel); + SourceEndpoint endpoint = new SourceEndpoint(source); + endpoint.setOutputChannel(channel); endpoint.afterPropertiesSet(); endpoint.send(new CommandMessage(new PollCommand())); Message message = channel.receive(1000); @@ -52,7 +53,8 @@ public class SourceEndpointTests { public void testAutoStartupDisabled() { TestSource source = new TestSource("testing", 1); QueueChannel channel = new QueueChannel(); - SourceEndpoint endpoint = new SourceEndpoint(source, channel); + SourceEndpoint endpoint = new SourceEndpoint(source); + endpoint.setOutputChannel(channel); endpoint.setAutoStartup(false); endpoint.afterPropertiesSet(); endpoint.send(new CommandMessage(new PollCommand())); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml index d6c155a086..76413563c9 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml @@ -19,7 +19,7 @@ - + @@ -34,11 +34,7 @@ - - - - - +