diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index fe4a3400f0..9e438acf4f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -30,6 +30,7 @@ import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.config.ConfigurableBeanFactory; @@ -64,44 +65,46 @@ import org.springframework.util.StringUtils; */ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, - InitializingBean, SmartLifecycle { + InitializingBean, SmartLifecycle, DisposableBean { - private volatile MessageHandler handler; - - private volatile String beanName; - - private volatile String inputChannelName; - - private volatile PollerMetadata pollerMetadata; - - private volatile Boolean autoStartup; - - private volatile int phase = 0; - - private volatile boolean isPhaseSet; - - private volatile MessageChannel inputChannel; - - private volatile ConfigurableBeanFactory beanFactory; - - private volatile ClassLoader beanClassLoader; - - private volatile AbstractEndpoint endpoint; - - private volatile boolean initialized; + private static final Log logger = LogFactory.getLog(ConsumerEndpointFactoryBean.class); private final Object initializationMonitor = new Object(); private final Object handlerMonitor = new Object(); - private final Log logger = LogFactory.getLog(this.getClass()); + private MessageHandler handler; - private volatile List adviceChain; + private String beanName; - private volatile DestinationResolver channelResolver; + private String inputChannelName; + + private PollerMetadata pollerMetadata; + + private Boolean autoStartup; + + private int phase = 0; + + private boolean isPhaseSet; + + private String role; + + private MessageChannel inputChannel; + + private ConfigurableBeanFactory beanFactory; + + private ClassLoader beanClassLoader; + + private List adviceChain; + + private DestinationResolver channelResolver; private TaskScheduler taskScheduler; + private volatile AbstractEndpoint endpoint; + + private volatile boolean initialized; + public void setHandler(MessageHandler handler) { Assert.notNull(handler, "handler must not be null"); synchronized (this.handlerMonitor) { @@ -147,6 +150,10 @@ public class ConsumerEndpointFactoryBean this.isPhaseSet = true; } + public void setRole(String role) { + this.role = role; + } + @Override public void setBeanName(String beanName) { this.beanName = beanName; @@ -170,7 +177,7 @@ public class ConsumerEndpointFactoryBean @Override public void afterPropertiesSet() throws Exception { if (this.beanName == null) { - this.logger.error("The MessageHandler [" + this.handler + "] will be created without a 'componentName'. " + + logger.error("The MessageHandler [" + this.handler + "] will be created without a 'componentName'. " + "Consider specifying the 'beanName' property on this ConsumerEndpointFactoryBean."); } else { @@ -189,8 +196,8 @@ public class ConsumerEndpointFactoryBean } } catch (Exception e) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Could not set component name for handler " + if (logger.isDebugEnabled()) { + logger.debug("Could not set component name for handler " + this.handler + " for " + this.beanName + " :" + e.getMessage()); } } @@ -267,10 +274,10 @@ public class ConsumerEndpointFactoryBean Assert.isNull(this.pollerMetadata, "A poller should not be specified for endpoint '" + this.beanName + "', since '" + channel + "' is a SubscribableChannel (not pollable)."); this.endpoint = new EventDrivenConsumer((SubscribableChannel) channel, this.handler); - if (this.logger.isWarnEnabled() + if (logger.isWarnEnabled() && Boolean.FALSE.equals(this.autoStartup) && channel instanceof FixedSubscriberChannel) { - this.logger.warn("'autoStartup=\"false\"' has no effect when using a FixedSubscriberChannel"); + logger.warn("'autoStartup=\"false\"' has no effect when using a FixedSubscriberChannel"); } } else if (channel instanceof PollableChannel) { @@ -307,6 +314,7 @@ public class ConsumerEndpointFactoryBean phase = Integer.MAX_VALUE / 2; } this.endpoint.setPhase(phase); + this.endpoint.setRole(this.role); if (this.taskScheduler != null) { this.endpoint.setTaskScheduler(this.taskScheduler); } @@ -356,4 +364,11 @@ public class ConsumerEndpointFactoryBean } } + @Override + public void destroy() throws Exception { + if (this.endpoint != null) { + this.endpoint.destroy(); + } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java index f46a976f14..8d3a3aef77 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -20,6 +20,7 @@ import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.config.ConfigurableBeanFactory; @@ -42,35 +43,38 @@ import org.springframework.util.StringUtils; * @author Artem Bilan */ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean, - BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean, SmartLifecycle { + BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean, SmartLifecycle, DisposableBean { - private volatile MessageSource source; + private final Object initializationMonitor = new Object(); - private volatile MessageChannel outputChannel; + private MessageSource source; - private volatile String outputChannelName; + private MessageChannel outputChannel; - private volatile PollerMetadata pollerMetadata; + private String outputChannelName; - private volatile boolean autoStartup = true; + private PollerMetadata pollerMetadata; - private volatile int phase = Integer.MAX_VALUE / 2; + private boolean autoStartup = true; - private volatile Long sendTimeout; + private int phase = Integer.MAX_VALUE / 2; - private volatile String beanName; + private Long sendTimeout; - private volatile ConfigurableBeanFactory beanFactory; + private String beanName; - private volatile ClassLoader beanClassLoader; + private ConfigurableBeanFactory beanFactory; + + private ClassLoader beanClassLoader; + + private DestinationResolver channelResolver; + + private String role; private volatile SourcePollingChannelAdapter adapter; private volatile boolean initialized; - private volatile DestinationResolver channelResolver; - - private final Object initializationMonitor = new Object(); public void setSource(MessageSource source) { this.source = source; @@ -100,6 +104,10 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean> entry : this.lazyLifecycleRoles.entrySet()) { - roleController.addLifecyclesToRole(entry.getKey(), entry.getValue()); - } - } - catch (NoSuchBeanDefinitionException e) { - this.logger.error("No LifecycleRoleController in the context"); - } - } - @Override public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException { Assert.notNull(this.beanFactory, "BeanFactory must not be null"); @@ -222,16 +202,15 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean } } + Role role = AnnotationUtils.findAnnotation(method, Role.class); + if (role != null) { + endpoint.setRole(role.value()); + } + String endpointBeanName = generateBeanName(beanName, method, annotationType); endpoint.setBeanName(endpointBeanName); getBeanFactory().registerSingleton(endpointBeanName, endpoint); getBeanFactory().initializeBean(endpoint, endpointBeanName); - - Role role = AnnotationUtils.findAnnotation(method, Role.class); - if (role != null) { - MessagingAnnotationPostProcessor.this.lazyLifecycleRoles.add(role.value(), - endpointBeanName); - } } } } @@ -290,8 +269,4 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean return this.postProcessors; } - protected MultiValueMap getLazyLifecycleRoles() { - return this.lazyLifecycleRoles; - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelAdapterParser.java index 68ec2ddc97..e4d375b375 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelAdapterParser.java @@ -52,7 +52,8 @@ public abstract class AbstractChannelAdapterParser extends AbstractBeanDefinitio id = id + ".adapter"; } else if (!StringUtils.hasText(id)) { - id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry(), parserContext.isNested()); + id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry(), + parserContext.isNested()); } return id; } @@ -76,10 +77,7 @@ public abstract class AbstractChannelAdapterParser extends AbstractBeanDefinitio } String role = element.getAttribute(IntegrationNamespaceUtils.ROLE); if (StringUtils.hasText(role)) { - if (!StringUtils.hasText(element.getAttribute(ID_ATTRIBUTE))) { - parserContext.getReaderContext().error("When using 'role', 'id' is required", element); - } - IntegrationNamespaceUtils.putLifecycleInRole(role, element.getAttribute(ID_ATTRIBUTE), parserContext); + propertyValues.add("role", new TypedStringValue(role)); } } return beanDefinition; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java index 2d1b564fdd..52438eaa76 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2017 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. @@ -63,7 +63,8 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit id = element.getAttribute("name"); } if (!StringUtils.hasText(id)) { - id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry(), parserContext.isNested()); + id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry(), + parserContext.isNested()); } return id; } @@ -142,7 +143,9 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit String inputChannelName = element.getAttribute(inputChannelAttributeName); if (!parserContext.getRegistry().containsBeanDefinition(inputChannelName)) { - if (parserContext.getRegistry().containsBeanDefinition(IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME)) { + if (parserContext.getRegistry() + .containsBeanDefinition(IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME)) { + BeanDefinition channelRegistry = parserContext.getRegistry(). getBeanDefinition(IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME); ConstructorArgumentValues caValues = channelRegistry.getConstructorArgumentValues(); @@ -152,12 +155,14 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit } @SuppressWarnings("unchecked") - Collection channelCandidateNames = (Collection) caValues.getArgumentValue(0, Collection.class).getValue(); + Collection channelCandidateNames = + (Collection) caValues.getArgumentValue(0, Collection.class).getValue(); channelCandidateNames.add(inputChannelName); } else { parserContext.getReaderContext().error("Failed to locate '" + - IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME + "'", parserContext.getRegistry()); + IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME + "'", + parserContext.getRegistry()); } } IntegrationNamespaceUtils.checkAndConfigureFixedSubscriberChannel(element, parserContext, inputChannelName, @@ -174,13 +179,8 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit } IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE); - String role = element.getAttribute(IntegrationNamespaceUtils.ROLE); - if (StringUtils.hasText(role)) { - if (!StringUtils.hasText(element.getAttribute(ID_ATTRIBUTE))) { - parserContext.getReaderContext().error("When using 'role', 'id' is required", element); - } - IntegrationNamespaceUtils.putLifecycleInRole(role, element.getAttribute(ID_ATTRIBUTE), parserContext); - } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.ROLE); + AbstractBeanDefinition beanDefinition = builder.getBeanDefinition(); String beanName = this.resolveId(element, beanDefinition, parserContext); parserContext.registerBeanComponent(new BeanComponentDefinition(beanDefinition, beanName)); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java index 3d6b08f049..6c6bd38571 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -26,7 +26,6 @@ import org.w3c.dom.NodeList; import org.springframework.beans.BeanMetadataElement; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.BeanDefinitionHolder; -import org.springframework.beans.factory.config.BeanReference; import org.springframework.beans.factory.config.ConstructorArgumentValues; import org.springframework.beans.factory.config.ConstructorArgumentValues.ValueHolder; import org.springframework.beans.factory.config.RuntimeBeanReference; @@ -613,21 +612,6 @@ public abstract class IntegrationNamespaceUtils { } } - public static void putLifecycleInRole(String role, String beanName, ParserContext parserContext) { - BeanDefinitionRegistry registry = parserContext.getRegistry(); - IntegrationConfigUtils.registerRoleControllerDefinitionIfNecessary(registry); - BeanDefinition controllerDef = registry.getBeanDefinition( - IntegrationContextUtils.INTEGRATION_LIFECYCLE_ROLE_CONTROLLER); - @SuppressWarnings("unchecked") - ManagedList roles = (ManagedList) controllerDef.getConstructorArgumentValues() - .getArgumentValue(0, ManagedList.class).getValue(); - @SuppressWarnings("unchecked") - ManagedList lifecycles = (ManagedList) controllerDef.getConstructorArgumentValues() - .getArgumentValue(1, ManagedList.class).getValue(); - roles.add(role); - lifecycles.add(new RuntimeBeanReference(beanName)); - } - public static void injectPropertyWithAdapter(String beanRefAttribute, String methodRefAttribute, String expressionAttribute, String beanProperty, String adapterClass, Element element, BeanDefinitionBuilder builder, BeanMetadataElement processor, ParserContext parserContext) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java index 765b72a1e2..291b007fd1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java @@ -80,6 +80,12 @@ public abstract class ConsumerEndpointSpec, return _this(); } + @Override + public S role(String role) { + this.endpointFactoryBean.setRole(role); + return _this(); + } + /** * Configure a {@link TaskScheduler} for scheduling tasks, for example in the * Polling Consumer. By default the global {@code ThreadPoolTaskScheduler} bean is used. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java index 3b3012d3fa..6d717da45b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java @@ -115,6 +115,16 @@ public abstract class EndpointSpec, F extends Be */ public abstract S autoStartup(boolean autoStartup); + /** + * Specify the role for the endpoint. + * Such endpoints can be started/stopped as a group. + * @param role the role for this endpoint. + * @return the endpoint spec + * @see SmartLifecycle + * @see org.springframework.integration.support.SmartLifecycleRoleController + */ + public abstract S role(String role); + @Override public Map getComponentsToRegister() { return this.componentsToRegister.isEmpty() diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java index 1cc67d0d7d..b11322c58d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -53,4 +53,10 @@ public final class SourcePollingChannelAdapterSpec extends return _this(); } + @Override + public SourcePollingChannelAdapterSpec role(String role) { + this.endpointFactoryBean.setRole(role); + return this; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index a455495b55..64041fcf56 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -19,11 +19,16 @@ package org.springframework.integration.endpoint; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.context.SmartLifecycle; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.context.IntegrationProperties; +import org.springframework.integration.support.SmartLifecycleRoleController; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.PatternMatchUtils; +import org.springframework.util.StringUtils; /** * The base class for Message Endpoint implementations. @@ -40,7 +45,8 @@ import org.springframework.util.PatternMatchUtils; * @author Gary Russell * @author Artem Bilan */ -public abstract class AbstractEndpoint extends IntegrationObjectSupport implements SmartLifecycle { +public abstract class AbstractEndpoint extends IntegrationObjectSupport + implements SmartLifecycle, DisposableBean { private boolean autoStartupSetExplicitly; @@ -54,6 +60,9 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen protected final Condition lifecycleCondition = this.lifecycleLock.newCondition(); + private String role; + + private SmartLifecycleRoleController roleController; public void setAutoStartup(boolean autoStartup) { this.autoStartup = autoStartup; @@ -64,6 +73,22 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen this.phase = phase; } + /** + * Specify the role for the endpoint. + * Such endpoints can be started/stopped as a group. + * @param role the role for this endpoint. + * @since 5.0 + * @see SmartLifecycle + * @see org.springframework.integration.support.SmartLifecycleRoleController + */ + public void setRole(String role) { + this.role = role; + } + + public String getRole() { + return this.role; + } + @Override public void setTaskScheduler(TaskScheduler taskScheduler) { super.setTaskScheduler(taskScheduler); @@ -84,6 +109,26 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen } } } + + if (StringUtils.hasText(this.role)) { + try { + this.roleController = getBeanFactory() + .getBean(IntegrationContextUtils.INTEGRATION_LIFECYCLE_ROLE_CONTROLLER, + SmartLifecycleRoleController.class); + + this.roleController.addLifecycleToRole(this.role, this); + } + catch (NoSuchBeanDefinitionException e) { + this.logger.trace("No LifecycleRoleController in the context"); + } + } + } + + @Override + public void destroy() throws Exception { + if (this.roleController != null) { + this.roleController.removeLifecycle(this); + } } // SmartLifecycle implementation diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index 64ed69306e..593b007776 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -27,6 +27,7 @@ import org.aopalliance.aop.Advice; import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.beans.factory.BeanInitializationException; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.support.MessagingExceptionWrapper; @@ -178,6 +179,12 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement } this.initialized = true; } + try { + super.onInit(); + } + catch (Exception e) { + throw new BeanInitializationException("Cannot initialize: " + this, e); + } } @SuppressWarnings("unchecked") diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 06bf8203d4..fad396dcf9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -16,6 +16,7 @@ package org.springframework.integration.endpoint; +import org.springframework.beans.factory.BeanInitializationException; import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.core.AttributeAccessor; import org.springframework.integration.core.MessageProducer; @@ -156,9 +157,17 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements @Override protected void onInit() { + try { + super.onInit(); + } + catch (Exception e) { + throw new BeanInitializationException("Cannot initialize: " + this, e); + } + if (this.getBeanFactory() != null) { this.messagingTemplate.setBeanFactory(this.getBeanFactory()); } + } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/SmartLifecycleRoleController.java b/spring-integration-core/src/main/java/org/springframework/integration/support/SmartLifecycleRoleController.java index 2f67255fdb..6903cca591 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/SmartLifecycleRoleController.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/SmartLifecycleRoleController.java @@ -19,11 +19,12 @@ package org.springframework.integration.support; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.Map.Entry; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Predicate; import java.util.stream.Collectors; import org.apache.commons.logging.Log; @@ -35,6 +36,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationListener; import org.springframework.context.Lifecycle; +import org.springframework.context.Phased; import org.springframework.context.SmartLifecycle; import org.springframework.integration.leader.event.AbstractLeaderEvent; import org.springframework.integration.leader.event.OnGrantedEvent; @@ -85,12 +87,7 @@ public class SmartLifecycleRoleController implements ApplicationListener lifcycles) { - for (Entry> lifecyclesInRole : lifcycles.entrySet()) { - String role = lifecyclesInRole.getKey(); - for (SmartLifecycle lifecycle : lifecyclesInRole.getValue()) { - addLifecycleToRole(role, lifecycle); - } - } + lifcycles.forEach((role, values) -> values.forEach(lifecycle -> addLifecycleToRole(role, lifecycle))); } @Override @@ -124,9 +121,7 @@ public class SmartLifecycleRoleController implements ApplicationListener lifecycleBeanNames) { Assert.state(this.applicationContext != null, "An application context is required to use this method"); - for (String lifecycleBeanName : lifecycleBeanNames) { - this.lazyLifecycles.add(role, lifecycleBeanName); - } + lifecycleBeanNames.forEach(lifecycleBeanName -> this.lazyLifecycles.add(role, lifecycleBeanName)); } /** @@ -139,20 +134,20 @@ public class SmartLifecycleRoleController implements ApplicationListener lifecycles = this.lifecycles.get(role); if (lifecycles != null) { - lifecycles = new ArrayList(lifecycles); - Collections.sort(lifecycles, (o1, o2) -> - o1.getPhase() < o2.getPhase() ? -1 : o1.getPhase() > o2.getPhase() ? 1 : 0); + lifecycles = new ArrayList<>(lifecycles); + lifecycles.sort(Comparator.comparingInt(Phased::getPhase)); if (logger.isDebugEnabled()) { logger.debug("Starting " + lifecycles + " in role " + role); } - for (SmartLifecycle lifecycle : lifecycles) { + + lifecycles.forEach(lifecycle -> { try { lifecycle.start(); } catch (Exception e) { logger.error("Failed to start " + lifecycle + " in role " + role, e); } - } + }); } else { if (logger.isDebugEnabled()) { @@ -171,20 +166,20 @@ public class SmartLifecycleRoleController implements ApplicationListener lifecycles = this.lifecycles.get(role); if (lifecycles != null) { - lifecycles = new ArrayList(lifecycles); - Collections.sort(lifecycles, (o1, o2) -> - o1.getPhase() < o2.getPhase() ? 1 : o1.getPhase() > o2.getPhase() ? -1 : 0); + lifecycles = new ArrayList<>(lifecycles); + lifecycles.sort((o1, o2) -> Integer.compare(o2.getPhase(), o1.getPhase())); if (logger.isDebugEnabled()) { logger.debug("Stopping " + lifecycles + " in role " + role); } - for (SmartLifecycle lifecycle : lifecycles) { + + lifecycles.forEach(lifecycle -> { try { lifecycle.stop(); } catch (Exception e) { logger.error("Failed to stop " + lifecycle + " in role " + role, e); } - } + }); } else { if (logger.isDebugEnabled()) { @@ -202,7 +197,7 @@ public class SmartLifecycleRoleController implements ApplicationListener 0) { addLazyLifecycles(); } - return new ArrayList<>(this.lifecycles.keySet()); + return Collections.unmodifiableCollection(this.lifecycles.keySet()); } /** @@ -238,6 +233,7 @@ public class SmartLifecycleRoleController implements ApplicationListener 0) { addLazyLifecycles(); } + if (!this.lifecycles.containsKey(role)) { return Collections.emptyMap(); } @@ -253,14 +249,12 @@ public class SmartLifecycleRoleController implements ApplicationListener> entry : this.lazyLifecycles.entrySet()) { - doAddLifecyclesToRole(entry.getKey(), entry.getValue()); - } + this.lazyLifecycles.forEach(this::doAddLifecyclesToRole); this.lazyLifecycles.clear(); } private void doAddLifecyclesToRole(String role, List lifecycleBeanNames) { - for (String lifecycleBeanName : lifecycleBeanNames) { + lifecycleBeanNames.forEach(lifecycleBeanName -> { try { SmartLifecycle lifecycle = this.applicationContext.getBean(lifecycleBeanName, SmartLifecycle.class); addLifecycleToRole(role, lifecycle); @@ -268,7 +262,7 @@ public class SmartLifecycleRoleController implements ApplicationListener lifecycles : this.lifecycles.values()) { + boolean actualRemoved = lifecycles.removeIf(Predicate.isEqual(lifecycle)); + if (!removed) { + removed = actualRemoved; + } + } + + if (removed) { + this.lifecycles.entrySet() + .removeIf(entry -> entry.getValue().isEmpty()); + } + + return removed; + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java index 5f474560e3..6770f841ca 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java @@ -57,8 +57,10 @@ import org.springframework.integration.dsl.context.IntegrationFlowContext; import org.springframework.integration.dsl.context.IntegrationFlowRegistration; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.support.SmartLifecycleRoleController; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; @@ -85,6 +87,9 @@ public class ManualFlowTests { @Autowired private BeanFactory beanFactory; + @Autowired + private SmartLifecycleRoleController roleController; + @Test public void testWithAnonymousMessageProducerStart() { final AtomicBoolean started = new AtomicBoolean(); @@ -99,8 +104,8 @@ public class ManualFlowTests { }; QueueChannel channel = new QueueChannel(); IntegrationFlow flow = IntegrationFlows.from(producer) - .channel(channel) - .get(); + .channel(channel) + .get(); this.integrationFlowContext.registration(flow).register(); assertTrue(started.get()); } @@ -127,8 +132,8 @@ public class ManualFlowTests { MyProducerSpec spec = new MyProducerSpec(new MyProducer()); QueueChannel channel = new QueueChannel(); IntegrationFlow flow = IntegrationFlows.from(spec.id("foo")) - .channel(channel) - .get(); + .channel(channel) + .get(); this.integrationFlowContext.registration(flow).register(); assertTrue(started.get()); } @@ -306,6 +311,51 @@ public class ManualFlowTests { assertEquals("test", receive.getPayload()); } + @Test + public void testRoleControl() { + String testRole = "bridge"; + + PollableChannel resultChannel = new QueueChannel(); + + IntegrationFlowRegistration flowRegistration = + this.integrationFlowContext + .registration(flow -> flow + .bridge(e -> e.role(testRole)) + .channel(resultChannel)) + .register(); + + MessagingTemplate messagingTemplate = + this.integrationFlowContext.messagingTemplateFor(flowRegistration.getId()); + + messagingTemplate.send(new GenericMessage<>("test")); + + Message receive = resultChannel.receive(1000); + assertNotNull(receive); + assertEquals("test", receive.getPayload()); + + this.roleController.stopLifecyclesInRole(testRole); + + try { + messagingTemplate.send(new GenericMessage<>("test2")); + } + catch (Exception e) { + assertThat(e, instanceOf(MessageDeliveryException.class)); + assertThat(e.getMessage(), containsString("Dispatcher has no subscribers for channel")); + } + + this.roleController.startLifecyclesInRole(testRole); + + messagingTemplate.send(new GenericMessage<>("test2")); + + receive = resultChannel.receive(1000); + assertNotNull(receive); + assertEquals("test2", receive.getPayload()); + + flowRegistration.destroy(); + + assertTrue(this.roleController.getEndpointsRunningStatus(testRole).isEmpty()); + } + @Configuration @EnableIntegration public static class RootConfiguration { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java index 55d4bbc351..b1d62208b6 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -89,6 +89,7 @@ public class JmsInboundGateway extends MessagingGatewaySupport implements Dispos @Override public void destroy() throws Exception { this.endpoint.destroy(); + super.destroy(); } @Override diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java index 69a3e28713..36ad11694e 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -150,6 +150,8 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements @Override protected void onInit() { + super.onInit(); + this.listener.afterPropertiesSet(); if (!this.listenerContainer.isActive()) { this.listenerContainer.afterPropertiesSet(); @@ -161,7 +163,7 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements } Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(sessionAcknowledgeMode); if (acknowledgeMode != null) { - if (acknowledgeMode.intValue() == JmsAdapterUtils.SESSION_TRANSACTED) { + if (JmsAdapterUtils.SESSION_TRANSACTED == acknowledgeMode) { this.listenerContainer.setSessionTransacted(true); } else { @@ -191,16 +193,15 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements this.stop(); } this.listenerContainer.destroy(); + super.destroy(); } - @Override public int beforeShutdown() { this.stop(); return 0; } - @Override public int afterShutdown() { return 0; diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java index 5c7f234a4a..fb2c96cb72 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -110,20 +110,14 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition @Override protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { - String containerBeanName = this.parseMessageListenerContainer(element, parserContext, builder.getRawBeanDefinition()); - String listenerBeanName = this.parseMessageListener(element, parserContext, builder.getRawBeanDefinition()); + String containerBeanName = parseMessageListenerContainer(element, parserContext, builder.getRawBeanDefinition()); + String listenerBeanName = parseMessageListener(element, parserContext, builder.getRawBeanDefinition()); builder.addConstructorArgReference(containerBeanName); builder.addConstructorArgReference(listenerBeanName); builder.addConstructorArgValue(hasExternalContainer(element)); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE); - String role = element.getAttribute(IntegrationNamespaceUtils.ROLE); - if (StringUtils.hasText(role)) { - if (!StringUtils.hasText(element.getAttribute(ID_ATTRIBUTE))) { - parserContext.getReaderContext().error("When using 'role', 'id' is required", element); - } - IntegrationNamespaceUtils.putLifecycleInRole(role, element.getAttribute(ID_ATTRIBUTE), parserContext); - } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.ROLE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge", "sessionAcknowledgeMode"); } diff --git a/src/reference/asciidoc/endpoint.adoc b/src/reference/asciidoc/endpoint.adoc index aecf0dafe9..40dac822cd 100644 --- a/src/reference/asciidoc/endpoint.adoc +++ b/src/reference/asciidoc/endpoint.adoc @@ -653,6 +653,12 @@ private SmartLifecycleRoleController roleController; ... ---- +[source, java] +---- +IntegrationFlow flow -> flow + .handle(..., e -> e.role("cluster")); +---- + Each of these adds the endpoint to the role `cluster`. Invoking `roleController.startLifecyclesInRole("cluster")` (and the corresponding `stop...` method) will start/stop