From eea5089d692545cb58d876f846bf4f336f9d92c2 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 21 Jun 2019 16:50:38 -0400 Subject: [PATCH] Fix `SmartLifecycle.stop(Runnable)` usage (#2973) * Fix `SmartLifecycle.stop(Runnable)` usage We always have to call `callback` in the `SmartLifecycle.stop(Runnable)` implementation independently of component state * Fix `StandardIntegrationFlow.stop(Runnable)` for a logic when we don't have any `this.lifecycles` * Remove those `stop(Runnable)` which are fully equivalent of the `default` on in the `SmartLifecycle` * Some other simple polishing for the affected classes, e.g. `isSingleton()` is `default` with `true` in the `InitializingBean` **Cherry-pick to 5.1.x** * * Fix checkstyle violation --- .../config/ConsumerEndpointFactoryBean.java | 6 -- ...ourcePollingChannelAdapterFactoryBean.java | 5 -- .../dsl/StandardIntegrationFlow.java | 23 +++-- .../history/MessageHistoryConfigurer.java | 8 +- .../store/MessageGroupStoreReaper.java | 27 ++---- .../leader/LockRegistryLeaderInitiator.java | 6 -- .../jms/config/JmsChannelFactoryBean.java | 83 +++++++++---------- .../stomp/AbstractStompSessionManager.java | 10 --- .../config/XmppConnectionFactoryBean.java | 6 -- .../config/CuratorFrameworkFactoryBean.java | 11 --- .../config/LeaderInitiatorFactoryBean.java | 5 -- .../zookeeper/leader/LeaderInitiator.java | 23 ++--- .../metadata/ZookeeperMetadataStore.java | 9 +- 13 files changed, 73 insertions(+), 149 deletions(-) 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 5ef404899e..ed7ab50f24 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 @@ -242,11 +242,6 @@ public class ConsumerEndpointFactoryBean initializeEndpoint(); } - @Override - public boolean isSingleton() { - return true; - } - @Override public AbstractEndpoint getObject() throws Exception { if (!this.initialized) { @@ -336,7 +331,6 @@ public class ConsumerEndpointFactoryBean } } - /* * SmartLifecycle implementation (delegates to the created endpoint) */ 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 554b532ac3..ca3f93161c 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 @@ -157,11 +157,6 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean iterator = this.lifecycles.listIterator(this.lifecycles.size()); - while (iterator.hasPrevious()) { - SmartLifecycle lifecycle = iterator.previous(); - if (lifecycle.isRunning()) { - lifecycle.stop(aggregatingCallback); - } - else { - aggregatingCallback.run(); + if (this.lifecycles.size() > 0) { + AggregatingCallback aggregatingCallback = new AggregatingCallback(this.lifecycles.size(), callback); + ListIterator iterator = this.lifecycles.listIterator(this.lifecycles.size()); + while (iterator.hasPrevious()) { + SmartLifecycle lifecycle = iterator.previous(); + if (lifecycle.isRunning()) { + lifecycle.stop(aggregatingCallback); + } + else { + aggregatingCallback.run(); + } } } + else { + callback.run(); + } this.running = false; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java index 72a8c51227..575478a287 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryConfigurer.java @@ -52,7 +52,7 @@ import org.springframework.util.StringUtils; @IntegrationManagedResource public class MessageHistoryConfigurer implements SmartLifecycle, BeanFactoryAware, DestructionAwareBeanPostProcessor { - private final Log logger = LogFactory.getLog(this.getClass()); + private static final Log logger = LogFactory.getLog(MessageHistoryConfigurer.class); private final Set currentlyTrackedComponents = ConcurrentHashMap.newKeySet(); @@ -230,12 +230,6 @@ public class MessageHistoryConfigurer implements SmartLifecycle, BeanFactoryAwar } } - @Override - public void stop(Runnable callback) { - this.stop(); - callback.run(); - } - private static Collection getTrackableComponents(ListableBeanFactory beanFactory) { return BeanFactoryUtils.beansOfTypeIncludingAncestors(beanFactory, TrackableComponent.class).values(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStoreReaper.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStoreReaper.java index 3902f7b20f..48654ff859 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStoreReaper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStoreReaper.java @@ -39,20 +39,20 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia private static Log logger = LogFactory.getLog(MessageGroupStoreReaper.class); + private final ReentrantLock lifecycleLock = new ReentrantLock(); + private MessageGroupStore messageGroupStore; private boolean expireOnDestroy = false; private long timeout = -1; + private int phase = 0; + + private boolean autoStartup = true; + private volatile boolean running; - private final ReentrantLock lifecycleLock = new ReentrantLock(); - - private volatile int phase = 0; - - private volatile boolean autoStartup = true; - public MessageGroupStoreReaper(MessageGroupStore messageGroupStore) { this.messageGroupStore = messageGroupStore; } @@ -63,7 +63,6 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia /** * Flag to indicate that the stores should be expired when this component is destroyed (i.e. usually when its * enclosing {@link org.springframework.context.ApplicationContext} is closed). - * * @param expireOnDestroy the flag value to set */ public void setExpireOnDestroy(boolean expireOnDestroy) { @@ -73,7 +72,6 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia /** * Timeout in milliseconds (default -1). If negative then no groups ever time out. If greater than zero then all * groups older than that value are expired when this component is {@link #run()}. - * * @param timeout the timeout to set */ public void setTimeout(long timeout) { @@ -82,7 +80,6 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia /** * A message group store to expire according the other configurations. - * * @param messageGroupStore the {@link MessageGroupStore} to set */ public void setMessageGroupStore(MessageGroupStore messageGroupStore) { @@ -187,16 +184,4 @@ public class MessageGroupStoreReaper implements Runnable, DisposableBean, Initia this.autoStartup = autoStartup; } - @Override - public void stop(Runnable callback) { - this.lifecycleLock.lock(); - try { - this.stop(); - callback.run(); - } - finally { - this.lifecycleLock.unlock(); - } - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java b/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java index 07de87505f..1206070204 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java @@ -308,12 +308,6 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe } } - @Override - public void stop(Runnable runnable) { - stop(); - runnable.run(); - } - /** * Stop the registration of the {@link #candidate} for leader election. If the * candidate is currently leader, its leadership will be revoked. diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java index 6adcb0509b..af485b5724 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java @@ -51,6 +51,7 @@ import org.springframework.util.StringUtils; * @author Oleg Zhurakousky * @author Gary Russell * @author Artem Bilan + * * @since 2.0 */ public class JmsChannelFactoryBean extends AbstractFactoryBean @@ -64,83 +65,83 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean containerType; + private Class containerType; - private volatile boolean acceptMessagesWhileStopping; + private boolean acceptMessagesWhileStopping; - private volatile boolean autoStartup = true; + private boolean autoStartup = true; - private volatile String cacheLevelName; + private String cacheLevelName; - private volatile Integer cacheLevel; + private Integer cacheLevel; - private volatile String clientId; + private String clientId; - private volatile String concurrency; + private String concurrency; - private volatile Integer concurrentConsumers; + private Integer concurrentConsumers; - private volatile ConnectionFactory connectionFactory; + private ConnectionFactory connectionFactory; - private volatile Destination destination; + private Destination destination; - private volatile String destinationName; + private String destinationName; - private volatile DestinationResolver destinationResolver; + private DestinationResolver destinationResolver; - private volatile String durableSubscriptionName; + private String durableSubscriptionName; - private volatile ErrorHandler errorHandler; + private ErrorHandler errorHandler; - private volatile ExceptionListener exceptionListener; + private ExceptionListener exceptionListener; - private volatile Boolean exposeListenerSession; + private Boolean exposeListenerSession; - private volatile Integer idleTaskExecutionLimit; + private Integer idleTaskExecutionLimit; - private volatile Integer maxConcurrentConsumers; + private Integer maxConcurrentConsumers; - private volatile Integer maxMessagesPerTask; + private Integer maxMessagesPerTask; - private volatile String messageSelector; + private String messageSelector; - private volatile Integer phase; + private Integer phase; - private volatile Boolean pubSubDomain; + private Boolean pubSubDomain; - private volatile boolean pubSubNoLocal; + private boolean pubSubNoLocal; - private volatile Long receiveTimeout; + private Long receiveTimeout; - private volatile Long recoveryInterval; + private Long recoveryInterval; - private volatile String beanName; + private String beanName; - private volatile boolean subscriptionShared; + private boolean subscriptionShared; /** * This value differs from the container implementations' default (which is AUTO_ACKNOWLEDGE) */ - private volatile int sessionAcknowledgeMode = Session.SESSION_TRANSACTED; + private int sessionAcknowledgeMode = Session.SESSION_TRANSACTED; /** * This value differs from the container implementations' default (which is false). */ - private volatile boolean sessionTransacted = true; + private boolean sessionTransacted = true; - private volatile boolean subscriptionDurable; + private boolean subscriptionDurable; - private volatile Executor taskExecutor; + private Executor taskExecutor; - private volatile PlatformTransactionManager transactionManager; + private PlatformTransactionManager transactionManager; - private volatile String transactionName; + private String transactionName; - private volatile Integer transactionTimeout; + private Integer transactionTimeout; - private volatile int maxSubscribers = Integer.MAX_VALUE; + private int maxSubscribers = Integer.MAX_VALUE; public JmsChannelFactoryBean() { @@ -204,8 +205,7 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean