diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java index 89828577d1..2aaffa7ef3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java @@ -19,12 +19,10 @@ package org.springframework.integration.adapter; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageMapper; import org.springframework.integration.message.SimplePayloadMessageMapper; -import org.springframework.integration.scheduling.PollingSchedule; -import org.springframework.integration.scheduling.Schedule; import org.springframework.util.Assert; /** @@ -32,35 +30,12 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public abstract class AbstractTargetAdapter implements TargetAdapter { +public abstract class AbstractTargetAdapter implements MessageHandler { protected Log logger = LogFactory.getLog(this.getClass()); - private String name; - - private MessageChannel channel; - private MessageMapper mapper = new SimplePayloadMessageMapper(); - private Schedule schedule = new PollingSchedule(5); - - - public void setName(String name) { - this.name = name; - } - - public String getName() { - return this.name; - } - - public void setChannel(MessageChannel channel) { - Assert.notNull(channel, "'channel' must not be null"); - this.channel = channel; - } - - public MessageChannel getChannel() { - return this.channel; - } public void setMessageMapper(MessageMapper mapper) { Assert.notNull(mapper, "'mapper' must not be null"); @@ -71,15 +46,6 @@ public abstract class AbstractTargetAdapter implements TargetAdapter { return this.mapper; } - public void setSchedule(Schedule schedule) { - Assert.notNull(schedule, "'schedule' must not be null"); - this.schedule = schedule; - } - - public Schedule getSchedule() { - return this.schedule; - } - public final Message handle(Message message) { this.sendToTarget(this.mapper.fromMessage(message)); return null; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java deleted file mode 100644 index 6f7b7a3ae7..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java +++ /dev/null @@ -1,33 +0,0 @@ -/* - * Copyright 2002-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.adapter; - -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.handler.MessageHandler; - -/** - * Base interface for target adapters. - * - * @author Mark Fisher - */ -public interface TargetAdapter extends MessageHandler { - - String getName(); - - void setChannel(MessageChannel channel); - -} 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 9c8ac94ee0..ec1911a8da 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 @@ -28,11 +28,11 @@ import org.springframework.beans.BeansException; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; +import org.springframework.integration.MessagingConfigurationException; import org.springframework.integration.MessagingException; -import org.springframework.integration.adapter.AbstractTargetAdapter; import org.springframework.integration.adapter.SourceAdapter; -import org.springframework.integration.adapter.TargetAdapter; import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; @@ -40,6 +40,7 @@ import org.springframework.integration.dispatcher.DefaultMessageDispatcher; import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.PooledMessageHandler; @@ -50,7 +51,6 @@ import org.springframework.integration.scheduling.Schedule; import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; /** * The messaging bus. Serves as a registry for channels and endpoints, manages their lifecycle, @@ -64,7 +64,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - private Map handlers = new ConcurrentHashMap(); + private Map endpoints = new ConcurrentHashMap(); private Map dispatchers = new ConcurrentHashMap(); @@ -90,8 +90,6 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif this.registerChannels(applicationContext); this.registerEndpoints(applicationContext); this.registerSourceAdapters(applicationContext); - this.registerTargetAdapters(applicationContext); - this.activateSubscriptions(applicationContext); } /** @@ -140,28 +138,6 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } - @SuppressWarnings("unchecked") - private void registerTargetAdapters(ApplicationContext context) { - Map targetAdapterBeans = - (Map) context.getBeansOfType(TargetAdapter.class); - for (Map.Entry entry : targetAdapterBeans.entrySet()) { - this.registerTargetAdapter(entry.getKey(), entry.getValue()); - } - } - - @SuppressWarnings("unchecked") - private void activateSubscriptions(ApplicationContext context) { - Map subscriptionBeans = - (Map) context.getBeansOfType(Subscription.class); - for (Subscription subscription : subscriptionBeans.values()) { - this.activateSubscription(subscription); - if (logger.isInfoEnabled()) { - logger.info("activated subscription to channel '" + subscription.getChannel() + - "' for handler '" + subscription.getHandler() + "'"); - } - } - } - public void initialize() { if (this.getInvalidMessageChannel() == null) { this.setInvalidMessageChannel(new SimpleChannel(Integer.MAX_VALUE)); @@ -212,32 +188,72 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } this.dispatchers.put(channel, dispatcher); this.channelRegistry.registerChannel(name, channel); + if (logger.isInfoEnabled()) { + logger.info("registered channel '" + name + "'"); + } } - public void registerEndpoint(String name, MessageEndpoint endpoint) { + public void registerHandler(String name, MessageHandler handler, Subscription subscription) { + this.registerHandler(name, handler, subscription, null); + } + + public void registerHandler(String name, MessageHandler handler, Subscription subscription, ConcurrencyPolicy concurrencyPolicy) { if (!this.initialized) { this.initialize(); } Assert.notNull(name, "'name' must not be null"); - Assert.notNull(endpoint, "'endpoint' must not be null"); + Assert.notNull(handler, "'handler' must not be null"); + Assert.notNull(subscription, "'subscription' must not be null"); + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); endpoint.setName(name); - this.handlers.put(name, endpoint); - endpoint.setChannelRegistry(this); - Schedule schedule = endpoint.getSchedule(); - if (endpoint.getInputChannelName() != null) { - Subscription subscription = new Subscription(); - subscription.setHandler(name); - subscription.setChannel(endpoint.getInputChannelName()); - if (schedule != null) { - subscription.setSchedule(schedule); - } - this.activateSubscription(subscription); + endpoint.setHandler(handler); + endpoint.setSubscription(subscription); + endpoint.setConcurrencyPolicy(concurrencyPolicy); + this.registerEndpoint(name, endpoint); + } + + public void registerEndpoint(String name, MessageEndpoint endpoint) { + if (endpoint instanceof ChannelRegistryAware) { + ((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry); } - if (this.autoCreateChannels) { - String defaultOutputChannelName = endpoint.getDefaultOutputChannelName(); - if (StringUtils.hasText(defaultOutputChannelName) && this.lookupChannel(defaultOutputChannelName) == null) { - this.registerChannel(defaultOutputChannelName, new SimpleChannel()); + this.endpoints.put(name, endpoint); + if (logger.isInfoEnabled()) { + logger.info("registered endpoint '" + name + "'"); + } + } + + private void activateEndpoints() { + for (MessageEndpoint endpoint : this.endpoints.values()) { + this.activateEndpoint(endpoint); + } + } + + private void activateEndpoint(MessageEndpoint endpoint) { + Subscription subscription = endpoint.getSubscription(); + MessageChannel channel = subscription.getChannel(); + if (channel == null) { + String channelName = subscription.getChannelName(); + if (channelName == null) { + throw new MessagingConfigurationException("endpoint '" + endpoint.getName() + + "' must provide either 'channel' or 'channelName' in its subscription metadata"); } + channel = this.lookupChannel(channelName); + if (channel == null) { + if (this.autoCreateChannels == false) { + throw new MessagingException("Cannot activate subscription, unknown channel '" + channelName + + "'. Consider enabling the 'autoCreateChannels' option for the message bus."); + } + if (this.logger.isInfoEnabled()) { + logger.info("auto-creating channel '" + channel.getName() + "'"); + } + channel = new SimpleChannel(); + this.registerChannel(channelName, channel); + } + } + this.registerWithDispatcher(channel, endpoint, subscription.getSchedule(), endpoint.getConcurrencyPolicy()); + if (logger.isInfoEnabled()) { + logger.info("activated subscription to channel '" + channel.getName() + + "' for endpoint '" + endpoint.getName() + "'"); } } @@ -259,56 +275,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } - public void registerTargetAdapter(String name, TargetAdapter targetAdapter) { - if (targetAdapter instanceof AbstractTargetAdapter) { - AbstractTargetAdapter adapter = (AbstractTargetAdapter) targetAdapter; - adapter.setName(name); - this.handlers.put(name, adapter); - MessageChannel channel = adapter.getChannel(); - Schedule schedule = adapter.getSchedule(); - ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); - concurrencyPolicy.setCoreConcurrency(1); - concurrencyPolicy.setMaxConcurrency(1); - this.doActivate(channel, adapter, schedule, concurrencyPolicy); - } - if (logger.isInfoEnabled()) { - logger.info("registered target adapter '" + name + "'"); - } - } - - public void activateSubscription(Subscription subscription) { - String channelName = subscription.getChannel(); - String handlerName = subscription.getHandler(); - Schedule schedule = subscription.getSchedule(); - ConcurrencyPolicy concurrencyPolicy = subscription.getConcurrencyPolicy(); - MessageHandler handler = this.handlers.get(handlerName); - if (handler == null) { - throw new MessagingException("Cannot activate subscription, unknown handler '" + handlerName + "'"); - } - MessageChannel channel = this.lookupChannel(channelName); - if (channel == null) { - if (this.autoCreateChannels == false) { - throw new MessagingException("Cannot activate subscription, unknown channel '" + channelName + - "'. Consider enabling the 'autoCreateChannels' option for the message bus."); - } - if (this.logger.isInfoEnabled()) { - logger.info("auto-creating channel '" + channelName + "'"); - } - channel = new SimpleChannel(); - this.registerChannel(channelName, channel); - } - this.doActivate(channel, handler, schedule, concurrencyPolicy); - if (logger.isInfoEnabled()) { - logger.info("activated subscription to channel '" + channelName + - "' for handler '" + handlerName + "'"); - } - } - - private void doActivate(MessageChannel channel, MessageHandler handler, Schedule schedule, ConcurrencyPolicy concurrencyPolicy) { + private void registerWithDispatcher(MessageChannel channel, MessageHandler handler, Schedule schedule, ConcurrencyPolicy concurrencyPolicy) { MessageDispatcher dispatcher = dispatchers.get(channel); if (dispatcher == null) { if (logger.isWarnEnabled()) { - logger.warn("no dispatcher available for channel '" + channel + "', be sure to register the channel"); + logger.warn("no dispatcher available for channel '" + channel.getName() + "', be sure to register the channel"); } } if (concurrencyPolicy != null) { @@ -335,8 +306,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } this.starting = true; synchronized (this.lifecycleMonitor) { + this.activateEndpoints(); this.taskScheduler.start(); - this.running = true; for (MessageDispatcher dispatcher : this.dispatchers.values()) { dispatcher.start(); if (logger.isInfoEnabled()) { 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 15b6f7f0b1..37658074d0 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 @@ -16,7 +16,7 @@ package org.springframework.integration.bus; -import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.scheduling.Schedule; /** @@ -26,29 +26,39 @@ import org.springframework.integration.scheduling.Schedule; */ public class Subscription { - private String channel; + private MessageChannel channel; - private String handler; + private String channelName; private Schedule schedule; - private ConcurrencyPolicy concurrencyPolicy; - - public String getChannel() { - return this.channel; + public Subscription() { } - public void setChannel(String channel) { + public Subscription(MessageChannel channel) { this.channel = channel; } - public String getHandler() { - return this.handler; + public Subscription(String channelName) { + this.channelName = channelName; } - public void setHandler(String handler) { - this.handler = handler; + + public MessageChannel getChannel() { + return this.channel; + } + + public void setChannel(MessageChannel channel) { + this.channel = channel; + } + + public String getChannelName() { + return (this.channel != null) ? this.channel.getName() : this.channelName; + } + + public void setChannelName(String channelName) { + this.channelName = channelName; } public Schedule getSchedule() { @@ -59,12 +69,4 @@ public class Subscription { this.schedule = schedule; } - public ConcurrencyPolicy getConcurrencyPolicy() { - return this.concurrencyPolicy; - } - - public void setConcurrencyPolicy(ConcurrencyPolicy concurrencyPolicy) { - this.concurrencyPolicy = concurrencyPolicy; - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java index 4fdbd33996..232276b925 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java @@ -19,9 +19,6 @@ package org.springframework.integration.channel; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - import org.springframework.util.Assert; /** @@ -31,8 +28,6 @@ import org.springframework.util.Assert; */ public class DefaultChannelRegistry implements ChannelRegistry { - private Log logger = LogFactory.getLog(this.getClass()); - private Map channels = new ConcurrentHashMap(); private MessageChannel invalidMessageChannel; @@ -55,9 +50,6 @@ public class DefaultChannelRegistry implements ChannelRegistry { Assert.notNull(channel, "'channel' must not be null"); channel.setName(name); this.channels.put(name, channel); - if (logger.isInfoEnabled()) { - logger.info("registered channel '" + name + "'"); - } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java index 2832700001..6cc1f44613 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java @@ -29,6 +29,8 @@ import org.springframework.integration.adapter.DefaultTargetAdapter; import org.springframework.integration.adapter.MethodInvokingSource; import org.springframework.integration.adapter.MethodInvokingTarget; import org.springframework.integration.adapter.PollingSourceAdapter; +import org.springframework.integration.bus.Subscription; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.util.StringUtils; /** @@ -79,6 +81,7 @@ public class ChannelAdapterParser implements BeanDefinitionParser { if (StringUtils.hasText(period)) { adapterDef.getPropertyValues().addPropertyValue("period", period); } + adapterDef.getPropertyValues().addPropertyValue("channel", new RuntimeBeanReference(channel)); } else { adapterDef = new RootBeanDefinition(DefaultTargetAdapter.class); @@ -89,12 +92,22 @@ public class ChannelAdapterParser implements BeanDefinitionParser { String invokerBeanName = parserContext.getReaderContext().generateBeanName(invokerDef); parserContext.registerBeanComponent(new BeanComponentDefinition(invokerDef, invokerBeanName)); adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName)); - adapterDef.getPropertyValues().addPropertyValue("channel", new RuntimeBeanReference(channel)); adapterDef.setSource(parserContext.extractSource(element)); String beanName = element.getAttribute(ID_ATTRIBUTE); if (!StringUtils.hasText(beanName)) { beanName = parserContext.getReaderContext().generateBeanName(adapterDef); } + if (!this.isInbound) { + RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class); + RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class); + subscriptionDef.getPropertyValues().addPropertyValue("channel", new RuntimeBeanReference(channel)); + String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef); + parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName)); + endpointDef.getPropertyValues().addPropertyValue("subscription", new RuntimeBeanReference(subscriptionBeanName)); + endpointDef.getPropertyValues().addPropertyValue("handler", new RuntimeBeanReference(beanName)); + String endpointBeanName = parserContext.getReaderContext().generateBeanName(endpointDef); + parserContext.registerBeanComponent(new BeanComponentDefinition(endpointDef, endpointBeanName)); + } parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDef, beanName)); return adapterDef; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java index a8550614da..3d5e4de996 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java @@ -31,6 +31,7 @@ import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.BeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.MessagingConfigurationException; +import org.springframework.integration.bus.Subscription; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.DefaultMessageHandlerAdapter; @@ -49,7 +50,9 @@ public class EndpointParser implements BeanDefinitionParser { private static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel"; - private static final String INPUT_CHANNEL_PROPERTY = "inputChannelName"; + private static final String SUBSCRIPTION_PROPERTY = "subscription"; + + private static final String CHANNEL_PROPERTY = "channel"; private static final String DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE = "default-output-channel"; @@ -90,8 +93,9 @@ public class EndpointParser implements BeanDefinitionParser { RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class); endpointDef.setSource(parserContext.extractSource(element)); String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE); + RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class); if (StringUtils.hasText(inputChannel)) { - endpointDef.getPropertyValues().addPropertyValue(INPUT_CHANNEL_PROPERTY, inputChannel); + subscriptionDef.getPropertyValues().addPropertyValue(CHANNEL_PROPERTY, new RuntimeBeanReference(inputChannel)); } String defaultOutputChannel = element.getAttribute(DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE); if (StringUtils.hasText(defaultOutputChannel)) { @@ -104,7 +108,7 @@ public class EndpointParser implements BeanDefinitionParser { if (child.getNodeType() == Node.ELEMENT_NODE) { String localName = child.getLocalName(); if (CONCURRENCY_ELEMENT.equals(localName)) { - parseConcurrencyPolicy((Element) child, endpointDef); + parseConcurrencyPolicy((Element) child, subscriptionDef); } else if (HANDLER_ELEMENT.equals(localName)) { String ref = ((Element) child).getAttribute(REF_ATTRIBUTE); @@ -113,6 +117,9 @@ public class EndpointParser implements BeanDefinitionParser { } } } + String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef); + parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName)); + endpointDef.getPropertyValues().addPropertyValue(SUBSCRIPTION_PROPERTY, new RuntimeBeanReference(subscriptionBeanName)); if (childHandlerRefs.size() > 0) { if (childHandlerRefs.size() == 1) { endpointDef.getPropertyValues().addPropertyValue( @@ -153,7 +160,7 @@ public class EndpointParser implements BeanDefinitionParser { return endpointDef; } - private void parseConcurrencyPolicy(Element concurrencyElement, RootBeanDefinition endpointDefinition) { + private void parseConcurrencyPolicy(Element concurrencyElement, RootBeanDefinition subscriptionDefinition) { ConcurrencyPolicy policy = new ConcurrencyPolicy(); String coreConcurrency = concurrencyElement.getAttribute(CORE_CONCURRENCY_ATTRIBUTE); String maxConcurrency = concurrencyElement.getAttribute(MAX_CONCURRENCY_ATTRIBUTE); @@ -163,7 +170,7 @@ public class EndpointParser implements BeanDefinitionParser { if (StringUtils.hasText(maxConcurrency)) { policy.setMaxConcurrency(Integer.parseInt(maxConcurrency)); } - endpointDefinition.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy); + subscriptionDefinition.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy); } private void parseSchedule(Element scheduleElement, RootBeanDefinition endpointDefinition) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index c2408fc877..5e7fe56c0e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -44,10 +44,10 @@ import org.springframework.integration.annotation.Polled; import org.springframework.integration.annotation.Router; import org.springframework.integration.annotation.Splitter; import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.bus.Subscription; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerChain; @@ -106,24 +106,23 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor return bean; } DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); - this.configureInputChannel(bean, beanName, endpointAnnotation, endpoint); - this.configureDefaultOutputChannel(bean, beanName, endpointAnnotation, endpoint); + this.configureInput(bean, beanName, endpointAnnotation, endpoint); MessageHandlerChain handlerChain = this.createHandlerChain(bean); - if (handlerChain != null) { - endpoint.setHandler(handlerChain); - } - this.messageBus.registerEndpoint(beanName, endpoint); - return endpoint; + endpoint.setHandler(handlerChain); + this.configureDefaultOutput(bean, beanName, endpointAnnotation, endpoint); + this.messageBus.registerEndpoint(bean + "-endpoint", endpoint); + return bean; } - private void configureInputChannel(final Object bean, final String beanName, - MessageEndpoint annotation, final DefaultMessageEndpoint endpoint) { + private void configureInput(final Object bean, final String beanName, MessageEndpoint annotation, + final DefaultMessageEndpoint endpoint) { String channelName = annotation.input(); if (StringUtils.hasText(channelName)) { - endpoint.setInputChannelName(channelName); + Subscription subscription = new Subscription(); + subscription.setChannelName(channelName); Schedule schedule = new PollingSchedule(annotation.pollPeriod()); - endpoint.setSchedule(schedule); - return; + subscription.setSchedule(schedule); + endpoint.setSubscription(subscription); } ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { @@ -140,22 +139,16 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor String channelName = beanName + "-inputChannel"; messageBus.registerChannel(channelName, channel); messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter); - endpoint.setInputChannelName(channelName); + Subscription subscription = new Subscription(channel); Schedule schedule = new PollingSchedule(period); - endpoint.setSchedule(schedule); - if (period > 0) { - ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); - concurrencyPolicy.setCoreConcurrency(1); - concurrencyPolicy.setMaxConcurrency(1); - endpoint.setConcurrencyPolicy(concurrencyPolicy); - } - return; + subscription.setSchedule(schedule); + endpoint.setSubscription(subscription); } } }); } - private void configureDefaultOutputChannel(final Object bean, final String beanName, + private void configureDefaultOutput(final Object bean, final String beanName, final MessageEndpoint annotation, final DefaultMessageEndpoint endpoint) { String channelName = annotation.defaultOutput(); if (StringUtils.hasText(channelName)) { @@ -177,8 +170,9 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor DefaultTargetAdapter adapter = new DefaultTargetAdapter(target); SimpleChannel channel = new SimpleChannel(); String channelName = beanName + "-defaultOutputChannel"; + Subscription subscription = new Subscription(channel); messageBus.registerChannel(channelName, channel); - messageBus.registerTargetAdapter(beanName + "-targetAdapter", adapter); + messageBus.registerHandler(beanName + "-targetAdapter", adapter, subscription); endpoint.setDefaultOutputChannelName(channelName); foundDefaultOutput = true; return; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java index 69cb8fc20d..05d53036ba 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java @@ -28,7 +28,7 @@ import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.annotation.Subscriber; import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.endpoint.DefaultMessageEndpoint; +import org.springframework.integration.bus.Subscription; import org.springframework.integration.handler.DefaultMessageHandlerAdapter; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -85,13 +85,10 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor { adapter.setMethodName(method.getName()); adapter.setObject(bean); adapter.afterPropertiesSet(); - DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); - endpoint.setInputChannelName(channelName); - endpoint.setChannelRegistry(messageBus); - endpoint.setHandler(adapter); - String endpointName = ClassUtils.getShortNameAsProperty(targetClass) + + String adapterName = ClassUtils.getShortNameAsProperty(targetClass) + "-" + method.getName() + "-endpoint"; - messageBus.registerEndpoint(endpointName, endpoint); + Subscription subscription = new Subscription(channelName); + messageBus.registerHandler(adapterName, adapter, subscription); } } }); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index 0b428e1390..60a5262f96 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -19,31 +19,30 @@ 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.Subscription; import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; -import org.springframework.integration.scheduling.Schedule; /** * Default implementation of the {@link MessageEndpoint} interface. * * @author Mark Fisher */ -public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware { +public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryAware, BeanNameAware { private String name; - private String inputChannelName; - - private String defaultOutputChannelName; - private MessageHandler handler; - private Schedule schedule; + private Subscription subscription; private ConcurrencyPolicy concurrencyPolicy; + private String defaultOutputChannelName; + private ChannelRegistry channelRegistry; @@ -59,18 +58,8 @@ public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware { this.setName(beanName); } - /** - * Set the name of the channel from which this endpoint receives messages. - */ - public void setInputChannelName(String inputChannelName) { - this.inputChannelName = inputChannelName; - } - - /** - * Return the name of the channel from which this endpoint receives messages. - */ - public String getInputChannelName() { - return this.inputChannelName; + public String getDefaultOutputChannelName() { + return this.defaultOutputChannelName; } /** @@ -80,8 +69,8 @@ public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware { this.defaultOutputChannelName = defaultOutputChannelName; } - public String getDefaultOutputChannelName() { - return this.defaultOutputChannelName; + public MessageHandler getHandler() { + return this.handler; } /** @@ -91,12 +80,12 @@ public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware { this.handler = handler; } - public Schedule getSchedule() { - return this.schedule; + public Subscription getSubscription() { + return this.subscription; } - public void setSchedule(Schedule schedule) { - this.schedule = schedule; + public void setSubscription(Subscription subscription) { + this.subscription = subscription; } public ConcurrencyPolicy getConcurrencyPolicy() { @@ -114,7 +103,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware { this.channelRegistry = channelRegistry; } - public Message handle(Message message) { + public Message handle(Message message) { if (this.handler == null) { if (this.defaultOutputChannelName == null) { throw new MessagingConfigurationException( diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index 464c3fbe53..f5ecfaf813 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -16,9 +16,8 @@ package org.springframework.integration.endpoint; -import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.bus.Subscription; import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.scheduling.Schedule; /** * Base interface for message endpoints. @@ -27,19 +26,11 @@ import org.springframework.integration.scheduling.Schedule; */ public interface MessageEndpoint extends MessageHandler { - void setName(String name); + String getName(); - void setInputChannelName(String inputChannelName); + MessageHandler getHandler(); - String getInputChannelName(); - - void setDefaultOutputChannelName(String defaultOutputChannelName); - - String getDefaultOutputChannelName(); - - void setChannelRegistry(ChannelRegistry channelRegistry); - - Schedule getSchedule(); + Subscription getSubscription(); ConcurrencyPolicy getConcurrencyPolicy(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java index 93fb8e1e11..fb8ab43e03 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java @@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.bus.Subscription; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; @@ -46,15 +47,16 @@ public class DefaultTargetAdapterTests { target.afterPropertiesSet(); DefaultTargetAdapter adapter = new DefaultTargetAdapter(target); SimpleChannel channel = new SimpleChannel(); - adapter.setChannel(channel); + Subscription subscription = new Subscription(); + subscription.setChannel(channel); Message message = new GenericMessage("123", "testing"); channel.send(message); assertNull(queue.poll()); MessageBus bus = new MessageBus(); bus.registerChannel("channel", channel); - bus.registerTargetAdapter("targetAdapter", adapter); + bus.registerHandler("targetAdapter", adapter, subscription); bus.start(); - String result = queue.poll(100, TimeUnit.MILLISECONDS); + String result = queue.poll(500, TimeUnit.MILLISECONDS); assertNotNull(result); assertEquals("testing", result); bus.stop(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml index 78a1d4d15e..52dffeec47 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml @@ -29,13 +29,25 @@ - + + + + + + + + + - - + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java index ce772ec26a..435f16bb71 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java @@ -26,6 +26,7 @@ import org.junit.Test; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.bus.Subscription; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.message.StringMessage; @@ -46,10 +47,9 @@ public class ApplicationEventTargetAdapterTests { MessageChannel channel = new SimpleChannel(); ApplicationEventTargetAdapter adapter = new ApplicationEventTargetAdapter(); adapter.setApplicationEventPublisher(publisher); - adapter.setChannel(channel); MessageBus bus = new MessageBus(); bus.registerChannel("channel", channel); - bus.registerTargetAdapter("adapter", adapter); + bus.registerHandler("adapter", adapter, new Subscription(channel)); bus.start(); assertEquals(1, latch.getCount()); channel.send(new StringMessage("123", "testing")); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java index 8741873be6..7411102191 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java @@ -39,7 +39,6 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - adapter.setChannel(channel); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); channel.send(new StringMessage("foo")); @@ -54,7 +53,6 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - adapter.setChannel(channel); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); channel.send(new StringMessage("foo")); @@ -72,7 +70,6 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - adapter.setChannel(channel); adapter.setShouldAppendNewLine(true); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); @@ -92,7 +89,6 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - adapter.setChannel(channel); ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); retriever.setMaxMessagesPerTask(2); DispatcherTask dispatcherTask = new DispatcherTask(retriever); @@ -109,7 +105,6 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - adapter.setChannel(channel); adapter.setShouldAppendNewLine(true); ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); retriever.setReceiveTimeout(0); @@ -129,7 +124,6 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - adapter.setChannel(channel); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); TestObject testObject = new TestObject("foo"); @@ -145,7 +139,6 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - adapter.setChannel(channel); ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); retriever.setReceiveTimeout(0); retriever.setMaxMessagesPerTask(2); @@ -165,7 +158,6 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - adapter.setChannel(channel); adapter.setShouldAppendNewLine(true); ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); retriever.setReceiveTimeout(0); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java index 86b38b7f55..bf7adb1bbb 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java @@ -30,7 +30,7 @@ import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.dispatcher.DefaultMessageDispatcher; import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; -import org.springframework.integration.endpoint.DefaultMessageEndpoint; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.PooledMessageHandler; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.Message; @@ -467,19 +467,18 @@ public class DefaultMessageDispatcherTests { assertEquals("endpoint2 should have accepted the message", 1, counter2.get()); } - private static class TestEndpoint extends DefaultMessageEndpoint { + + private static class TestEndpoint implements MessageHandler { private AtomicInteger counter; private CountDownLatch latch; - public TestEndpoint(AtomicInteger counter, CountDownLatch latch) { this.counter = counter; this.latch = latch; } - @Override public Message handle(Message message) { counter.incrementAndGet(); latch.countDown(); 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 d19508630c..9566b4c805 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 @@ -27,8 +27,7 @@ import org.junit.Test; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; -import org.springframework.integration.endpoint.DefaultMessageEndpoint; -import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.scheduling.PollingSchedule; @@ -44,8 +43,7 @@ public class FixedDelayConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); SimpleChannel channel = new SimpleChannel(); - MessageEndpoint endpoint = new DefaultMessageEndpoint() { - @Override + MessageHandler handler = new MessageHandler() { public Message handle(Message message) { counter.incrementAndGet(); latch.countDown(); @@ -55,7 +53,6 @@ public class FixedDelayConsumerTests { MessageBus bus = new MessageBus(); bus.initialize(); bus.registerChannel("testChannel", channel); - bus.registerEndpoint("testEndpoint", endpoint); PollingSchedule schedule = new PollingSchedule(10); schedule.setFixedRate(false); ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); @@ -63,10 +60,8 @@ public class FixedDelayConsumerTests { concurrencyPolicy.setMaxConcurrency(1); Subscription subscription = new Subscription(); subscription.setSchedule(schedule); - subscription.setConcurrencyPolicy(concurrencyPolicy); - subscription.setChannel("testChannel"); - subscription.setHandler("testEndpoint"); - bus.activateSubscription(subscription); + subscription.setChannelName("testChannel"); + bus.registerHandler("testHandler", handler, subscription, concurrencyPolicy); bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new GenericMessage(1, "test " + (i+1))); @@ -81,8 +76,7 @@ public class FixedDelayConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); SimpleChannel channel = new SimpleChannel(); - MessageEndpoint endpoint = new DefaultMessageEndpoint() { - @Override + MessageHandler handler = new MessageHandler() { public Message handle(Message message) { counter.incrementAndGet(); latch.countDown(); @@ -92,14 +86,12 @@ public class FixedDelayConsumerTests { MessageBus bus = new MessageBus(); bus.initialize(); bus.registerChannel("testChannel", channel); - bus.registerEndpoint("testEndpoint", endpoint); PollingSchedule schedule = new PollingSchedule(10); schedule.setFixedRate(false); - Subscription subscription = new Subscription(); - subscription.setChannel("testChannel"); - subscription.setHandler("testEndpoint"); + Subscription subscription = new Subscription(channel); + subscription.setChannelName("testChannel"); subscription.setSchedule(schedule); - bus.activateSubscription(subscription); + bus.registerHandler("testHandler", handler, subscription); for (int i = 0; i < messagesToSend; i++) { channel.send(new GenericMessage(1, "test " + (i+1))); } 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 4f5975d045..782054c2b5 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 @@ -27,8 +27,7 @@ import org.junit.Test; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; -import org.springframework.integration.endpoint.DefaultMessageEndpoint; -import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.scheduling.PollingSchedule; @@ -44,8 +43,7 @@ public class FixedRateConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); SimpleChannel channel = new SimpleChannel(); - MessageEndpoint endpoint = new DefaultMessageEndpoint() { - @Override + MessageHandler handler = new MessageHandler() { public Message handle(Message message) { counter.incrementAndGet(); latch.countDown(); @@ -55,14 +53,12 @@ public class FixedRateConsumerTests { MessageBus bus = new MessageBus(); bus.initialize(); bus.registerChannel("testChannel", channel); - bus.registerEndpoint("testEndpoint", endpoint); PollingSchedule schedule = new PollingSchedule(10); schedule.setFixedRate(true); Subscription subscription = new Subscription(); - subscription.setChannel("testChannel"); - subscription.setHandler("testEndpoint"); + subscription.setChannelName("testChannel"); subscription.setSchedule(schedule); - bus.activateSubscription(subscription); + bus.registerHandler("testHandler", handler, subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new GenericMessage(1, "test " + (i+1))); @@ -77,8 +73,7 @@ public class FixedRateConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); SimpleChannel channel = new SimpleChannel(); - MessageEndpoint endpoint = new DefaultMessageEndpoint() { - @Override + MessageHandler handler = new MessageHandler() { public Message handle(Message message) { counter.incrementAndGet(); latch.countDown(); @@ -86,20 +81,16 @@ public class FixedRateConsumerTests { } }; MessageBus bus = new MessageBus(); - bus.initialize(); - bus.registerChannel("testChannel", channel); - bus.registerEndpoint("testEndpoint", endpoint); PollingSchedule schedule = new PollingSchedule(5); schedule.setFixedRate(true); ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); concurrencyPolicy.setCoreConcurrency(1); concurrencyPolicy.setMaxConcurrency(1); Subscription subscription = new Subscription(); - subscription.setChannel("testChannel"); - subscription.setHandler("testEndpoint"); + subscription.setChannelName("testChannel"); subscription.setSchedule(schedule); - subscription.setConcurrencyPolicy(concurrencyPolicy); - bus.activateSubscription(subscription); + bus.registerChannel("testChannel", channel); + bus.registerHandler("testHandler", handler, subscription, concurrencyPolicy); bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new GenericMessage(1, "test " + (i+1))); 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 ddd660b8c9..0faa8f3675 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 @@ -33,7 +33,7 @@ import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.adapter.SourceAdapter; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.endpoint.DefaultMessageEndpoint; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; @@ -45,25 +45,30 @@ import org.springframework.integration.message.StringMessage; public class MessageBusTests { @Test - public void testChannelsConnectedWithEndpoint() { + public void testOutputChannel() { MessageBus bus = new MessageBus(); MessageChannel sourceChannel = new SimpleChannel(); MessageChannel targetChannel = new SimpleChannel(); bus.registerChannel("sourceChannel", sourceChannel); - sourceChannel.send(new StringMessage("123", "test")); + StringMessage message = new StringMessage("test"); + message.getHeader().setReplyChannelName("targetChannel"); + sourceChannel.send(message); bus.registerChannel("targetChannel", targetChannel); - DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); - endpoint.setInputChannelName("sourceChannel"); - endpoint.setDefaultOutputChannelName("targetChannel"); - bus.registerEndpoint("endpoint", endpoint); + MessageHandler handler = new MessageHandler() { + public Message handle(Message message) { + return message; + } + }; + Subscription subscription = new Subscription(sourceChannel); + bus.registerHandler("handler", handler, subscription); bus.start(); - Message result = targetChannel.receive(100); + Message result = targetChannel.receive(3000); assertEquals("test", result.getPayload()); bus.stop(); } @Test - public void testChannelsWithoutEndpoint() { + public void testChannelsWithoutHandlers() { MessageBus bus = new MessageBus(); MessageChannel sourceChannel = new SimpleChannel(); sourceChannel.send(new StringMessage("123", "test")); @@ -84,31 +89,34 @@ public class MessageBusTests { sourceChannel.send(new GenericMessage("123", "test")); MessageChannel targetChannel = (MessageChannel) context.getBean("targetChannel"); MessageBus bus = (MessageBus) context.getBean("bus"); - Subscription subscription = new Subscription(); - subscription.setChannel("sourceChannel"); - subscription.setHandler("endpoint"); - bus.activateSubscription(subscription); - Message result = targetChannel.receive(100); + bus.start(); + Message result = targetChannel.receive(1000); assertEquals("test", result.getPayload()); } @Test - public void testExactlyOneEndpointReceivesUnicastMessage() { + public void testExactlyOneHandlerReceivesPointToPointMessage() { SimpleChannel inputChannel = new SimpleChannel(); SimpleChannel outputChannel1 = new SimpleChannel(); SimpleChannel outputChannel2 = new SimpleChannel(); - DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(); - endpoint1.setDefaultOutputChannelName("output1"); - endpoint1.setInputChannelName("input"); - DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint(); - endpoint2.setDefaultOutputChannelName("output2"); - endpoint2.setInputChannelName("input"); + MessageHandler handler1 = new MessageHandler() { + public Message handle(Message message) { + message.getHeader().setReplyChannelName("output1"); + return message; + } + }; + MessageHandler handler2 = new MessageHandler() { + public Message handle(Message message) { + message.getHeader().setReplyChannelName("output2"); + return message; + } + }; MessageBus bus = new MessageBus(); bus.registerChannel("input", inputChannel); bus.registerChannel("output1", outputChannel1); bus.registerChannel("output2", outputChannel2); - bus.registerEndpoint("endpoint1", endpoint1); - bus.registerEndpoint("endpoint2", endpoint2); + bus.registerHandler("handler1", handler1, new Subscription(inputChannel)); + bus.registerHandler("handler2", handler2, new Subscription(inputChannel)); bus.start(); inputChannel.send(new StringMessage(1, "testing")); Message message1 = outputChannel1.receive(100); @@ -117,6 +125,39 @@ public class MessageBusTests { assertTrue("exactly one message should be null", message1 == null ^ message2 == null); } + @Test + public void testBothHandlersReceivePublishSubscribeMessage() { + SimpleChannel inputChannel = new SimpleChannel(); + inputChannel.setBroadcaster(true); + SimpleChannel outputChannel1 = new SimpleChannel(); + SimpleChannel outputChannel2 = new SimpleChannel(); + MessageHandler handler1 = new MessageHandler() { + public Message handle(Message message) { + message.getHeader().setReplyChannelName("output1"); + return message; + } + }; + MessageHandler handler2 = new MessageHandler() { + public Message handle(Message message) { + message.getHeader().setReplyChannelName("output2"); + return message; + } + }; + MessageBus bus = new MessageBus(); + bus.registerChannel("input", inputChannel); + bus.registerChannel("output1", outputChannel1); + bus.registerChannel("output2", outputChannel2); + bus.registerHandler("handler1", handler1, new Subscription(inputChannel)); + bus.registerHandler("handler2", handler2, new Subscription(inputChannel)); + bus.start(); + inputChannel.send(new StringMessage(1, "testing")); + Message message1 = outputChannel1.receive(100); + Message message2 = outputChannel2.receive(0); + bus.stop(); + assertTrue("both handlers should have received and replied to the message", + (message1 != null && message2 != null)); + } + @Test public void testInvalidMessageChannelWithFailedDispatch() throws InterruptedException { MessageBus bus = new MessageBus(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml index e1fe6f7f5d..780171753e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml @@ -11,7 +11,11 @@ - + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java index b06f4ecb1d..987ecd6993 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java @@ -21,7 +21,8 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; -import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.handler.MessageHandler; @@ -35,49 +36,40 @@ public class DefaultMessageEndpointTests { @Test public void testDefaultReplyChannel() throws Exception { - MessageChannel channel = new SimpleChannel(); MessageChannel replyChannel = new SimpleChannel(); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("replyChannel", replyChannel); MessageHandler handler = new MessageHandler() { public Message handle(Message message) { return new StringMessage("123", "hello " + message.getPayload()); } }; DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); - endpoint.setInputChannelName("testChannel"); + endpoint.setChannelRegistry(channelRegistry); endpoint.setHandler(handler); endpoint.setDefaultOutputChannelName("replyChannel"); - MessageBus bus = new MessageBus(); - bus.registerChannel("testChannel", channel); - bus.registerEndpoint("testEndpoint", endpoint); - bus.registerChannel("replyChannel", replyChannel); - bus.start(); - StringMessage testMessage = new StringMessage(1, "test"); - channel.send(testMessage); - Message reply = replyChannel.receive(50); + endpoint.handle(new StringMessage(1, "test")); + Message reply = replyChannel.receive(50); assertNotNull(reply); assertEquals("hello test", reply.getPayload()); } @Test public void testExplicitReplyChannel() throws Exception { - MessageChannel channel = new SimpleChannel(); final MessageChannel replyChannel = new SimpleChannel(); + ChannelRegistry channelRegistry = new DefaultChannelRegistry(); + channelRegistry.registerChannel("replyChannel", replyChannel); MessageHandler handler = new MessageHandler() { public Message handle(Message message) { return new StringMessage("123", "hello " + message.getPayload()); } }; DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); - endpoint.setInputChannelName("testChannel"); + endpoint.setChannelRegistry(channelRegistry); endpoint.setHandler(handler); - MessageBus bus = new MessageBus(); - bus.registerChannel("testChannel", channel); - bus.registerEndpoint("testEndpoint", endpoint); - bus.registerChannel("replyChannel", replyChannel); - bus.start(); StringMessage testMessage = new StringMessage(1, "test"); testMessage.getHeader().setReplyChannelName("replyChannel"); - channel.send(testMessage); + endpoint.handle(testMessage); Message reply = replyChannel.receive(50); assertNotNull(reply); assertEquals("hello test", reply.getPayload());