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

INT-4361: Add a role() hook to Java DSL

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4361

* To get a gain of the method chain code flow and avoid extra annotation
parsing, add `.role(String)` hook to the `EndpointSpec`
* Delegate the provided `role` property to the `AbstractEndpoint`
* Register `AbstractEndpoint` as itself `SmartLifecycle` in the
`SmartLifecycleRoleController`
* Add `destroy()` to the `AbstractEndpoint` and remove it from the
`SmartLifecycleRoleController`
* Provide some Java 8 code style refactoring

* Rework XML parsers and Annotation processors to populate `role`
property on the `AbstractEndpoint`
* Wrap `roleController` bean extraction to the `NoSuchBeanDefinitionException` catch
* Fix several `AbstractEndpoint` implementation to properly call
`super.onInit()` which has been missed before
This commit is contained in:
Artem Bilan
2017-11-07 11:04:33 -05:00
committed by Gary Russell
parent a61327766e
commit f7e75223c7
18 changed files with 303 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -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<AbstractEndpoint>, 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<Advice> adviceChain;
private String beanName;
private volatile DestinationResolver<MessageChannel> 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<Advice> adviceChain;
private DestinationResolver<MessageChannel> 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();
}
}
}

46
spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
View File

@@ -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<SourcePollingChannelAdapter>,
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<MessageChannel> channelResolver;
private String role;
private volatile SourcePollingChannelAdapter adapter;
private volatile boolean initialized;
private volatile DestinationResolver<MessageChannel> channelResolver;
private final Object initializationMonitor = new Object();
public void setSource(MessageSource<?> source) {
this.source = source;
@@ -100,6 +104,10 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
this.phase = phase;
}
public void setRole(String role) {
this.role = role;
}
/**
* Specify the {@link DestinationResolver} strategy to use.
* The default is a BeanFactoryChannelResolver.
@@ -191,6 +199,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
spca.setBeanClassLoader(this.beanClassLoader);
spca.setAutoStartup(this.autoStartup);
spca.setPhase(this.phase);
spca.setRole(this.role);
spca.setBeanName(this.beanName);
spca.setBeanFactory(this.beanFactory);
spca.setTransactionSynchronizationFactory(this.pollerMetadata.getTransactionSynchronizationFactory());
@@ -240,4 +249,11 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
}
}
@Override
public void destroy() throws Exception {
if (this.adapter != null) {
this.adapter.destroy();
}
}
}

37
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java
View File

@@ -36,8 +36,6 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.core.annotation.AnnotatedElementUtils;
@@ -52,9 +50,7 @@ import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.support.SmartLifecycleRoleController;
import org.springframework.integration.util.MessagingAnnotationUtils;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -74,8 +70,7 @@ import org.springframework.util.StringUtils;
* @author Gary Russell
* @author Rick Hogge
*/
public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware,
InitializingBean, SmartInitializingSingleton {
public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, InitializingBean {
protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
@@ -134,21 +129,6 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
return bean;
}
@Override
public void afterSingletonsInstantiated() {
SmartLifecycleRoleController roleController;
try {
roleController = this.beanFactory.getBean(IntegrationContextUtils.INTEGRATION_LIFECYCLE_ROLE_CONTROLLER,
SmartLifecycleRoleController.class);
for (Entry<String, List<String>> 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<String, String> getLazyLifecycleRoles() {
return this.lazyLifecycleRoles;
}
}

8
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelAdapterParser.java
View File

@@ -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;

24
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java
View File

@@ -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<String> channelCandidateNames = (Collection<String>) caValues.getArgumentValue(0, Collection.class).getValue();
Collection<String> channelCandidateNames =
(Collection<String>) 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));

18
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -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<String> roles = (ManagedList<String>) controllerDef.getConstructorArgumentValues()
.getArgumentValue(0, ManagedList.class).getValue();
@SuppressWarnings("unchecked")
ManagedList<BeanReference> lifecycles = (ManagedList<BeanReference>) 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) {

6
spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java
View File

@@ -80,6 +80,12 @@ public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>,
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.

10
spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java
View File

@@ -115,6 +115,16 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, 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<Object, String> getComponentsToRegister() {
return this.componentsToRegister.isEmpty()

8
spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java
View File

@@ -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;
}
}

47
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
View File

@@ -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

7
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
View File

@@ -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")

9
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

@@ -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());
}
}
/**

73
spring-integration-core/src/main/java/org/springframework/integration/support/SmartLifecycleRoleController.java
View File

@@ -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<Abstrac
* @param lifcycles the {@link MultiValueMap} of beans in roles.
*/
public SmartLifecycleRoleController(MultiValueMap<String, SmartLifecycle> lifcycles) {
for (Entry<String, List<SmartLifecycle>> 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<Abstrac
*/
public void addLifecyclesToRole(String role, List<String> 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<Abstrac
}
List<SmartLifecycle> lifecycles = this.lifecycles.get(role);
if (lifecycles != null) {
lifecycles = new ArrayList<SmartLifecycle>(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<Abstrac
}
List<SmartLifecycle> lifecycles = this.lifecycles.get(role);
if (lifecycles != null) {
lifecycles = new ArrayList<SmartLifecycle>(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<Abstrac
if (this.lazyLifecycles.size() > 0) {
addLazyLifecycles();
}
return new ArrayList<>(this.lifecycles.keySet());
return Collections.unmodifiableCollection(this.lifecycles.keySet());
}
/**
@@ -238,6 +233,7 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
if (this.lazyLifecycles.size() > 0) {
addLazyLifecycles();
}
if (!this.lifecycles.containsKey(role)) {
return Collections.emptyMap();
}
@@ -253,14 +249,12 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
}
private synchronized void addLazyLifecycles() {
for (Entry<String, List<String>> entry : this.lazyLifecycles.entrySet()) {
doAddLifecyclesToRole(entry.getKey(), entry.getValue());
}
this.lazyLifecycles.forEach(this::doAddLifecyclesToRole);
this.lazyLifecycles.clear();
}
private void doAddLifecyclesToRole(String role, List<String> 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<Abstrac
catch (NoSuchBeanDefinitionException e) {
logger.warn("Skipped; no such bean: " + lifecycleBeanName);
}
}
});
}
@Override
@@ -281,4 +275,31 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
}
}
/**
* Remove the provided SmartLifecycle from all the roles,
* for example when a SmartLifecycle bean is destroyed.
* The role entry in the lifecycles map is cleared as well
* if its value list is empty after SmartLifecycle removal.
* @param lifecycle the SmartLifecycle to remove.
* @return the removal status
* @since 5.0
*/
public boolean removeLifecycle(SmartLifecycle lifecycle) {
boolean removed = false;
for (List<SmartLifecycle> 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;
}
}

58
spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java
View File

@@ -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 {

3
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java
View File

@@ -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

9
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java
View File

@@ -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;

14
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java
View File

@@ -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");
}

6
src/reference/asciidoc/endpoint.adoc
View File

@@ -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