INT-4486: Properly implement stop(Runnable)
JIRA: https://jira.spring.io/browse/INT-4486 The `SmartLifecycle.stop(Runnable callback)` must always call the `callback` in the end independently of the internal state * Revise all the `SmartLifecycle` implementations for the proper `callback` handling **Cherry-pick to 5.0.x and 4.3.x**
This commit is contained in:
committed by
Gary Russell
parent
125cc861d4
commit
1ce656e4fe
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2017 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -234,6 +234,9 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
|
||||
this.container.stop(callback);
|
||||
this.declared = false;
|
||||
}
|
||||
else {
|
||||
callback.run();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2017 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -66,10 +66,11 @@ import org.springframework.util.StringUtils;
|
||||
* @author Mark Fisher
|
||||
* @author Gary Russell
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 2.1
|
||||
*/
|
||||
public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChannel> implements SmartLifecycle,
|
||||
DisposableBean, BeanNameAware {
|
||||
public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChannel>
|
||||
implements SmartLifecycle, DisposableBean, BeanNameAware {
|
||||
|
||||
private volatile AbstractAmqpChannel channel;
|
||||
|
||||
@@ -516,13 +517,14 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
|
||||
if (this.channel instanceof SmartLifecycle) {
|
||||
((SmartLifecycle) this.channel).stop(callback);
|
||||
}
|
||||
else {
|
||||
callback.run();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void destroyInstance(AbstractAmqpChannel instance) throws Exception {
|
||||
if (instance instanceof DisposableBean) {
|
||||
((DisposableBean) this.channel).destroy();
|
||||
}
|
||||
this.channel.destroy();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -362,6 +362,9 @@ public class ConsumerEndpointFactoryBean
|
||||
if (this.endpoint != null) {
|
||||
this.endpoint.stop(callback);
|
||||
}
|
||||
else {
|
||||
callback.run();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2017 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -247,6 +247,9 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
|
||||
if (this.adapter != null) {
|
||||
this.adapter.stop(callback);
|
||||
}
|
||||
else {
|
||||
callback.run();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -95,6 +95,9 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, SmartLi
|
||||
if (this.running.getAndSet(false)) {
|
||||
this.targetIntegrationFlow.stop(callback);
|
||||
}
|
||||
else {
|
||||
callback.run();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2017 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -199,6 +199,9 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport
|
||||
logger.info("stopped " + this);
|
||||
}
|
||||
}
|
||||
else {
|
||||
callback.run();
|
||||
}
|
||||
}
|
||||
finally {
|
||||
this.lifecycleLock.unlock();
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2014 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -36,6 +36,7 @@ import org.springframework.util.StringUtils;
|
||||
* @author Gary Russell
|
||||
* @author Artem Bilan
|
||||
* @author Ali Shahbour
|
||||
*
|
||||
* @since 3.0
|
||||
*
|
||||
*/
|
||||
@@ -192,6 +193,9 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
|
||||
if (this.adapter != null) {
|
||||
this.adapter.stop(callback);
|
||||
}
|
||||
else {
|
||||
callback.run();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2016 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -43,9 +43,12 @@ import org.springframework.util.Assert;
|
||||
/**
|
||||
* @author Mark Fisher
|
||||
* @author Gary Russell
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 2.0
|
||||
*/
|
||||
public class SubscribableJmsChannel extends AbstractJmsChannel implements SubscribableChannel, SmartLifecycle, DisposableBean {
|
||||
public class SubscribableJmsChannel extends AbstractJmsChannel
|
||||
implements SubscribableChannel, SmartLifecycle, DisposableBean {
|
||||
|
||||
private final AbstractMessageListenerContainer container;
|
||||
|
||||
@@ -72,13 +75,15 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
|
||||
|
||||
@Override
|
||||
public boolean subscribe(MessageHandler handler) {
|
||||
Assert.state(this.dispatcher != null, "'MessageDispatcher' must not be null. This channel might not have been initialized");
|
||||
Assert.state(this.dispatcher != null,
|
||||
"'MessageDispatcher' must not be null. This channel might not have been initialized");
|
||||
return this.dispatcher.addHandler(handler);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean unsubscribe(MessageHandler handler) {
|
||||
Assert.state(this.dispatcher != null, "'MessageDispatcher' must not be null. This channel might not have been initialized");
|
||||
Assert.state(this.dispatcher != null,
|
||||
"'MessageDispatcher' must not be null. This channel might not have been initialized");
|
||||
return this.dispatcher.removeHandler(handler);
|
||||
}
|
||||
|
||||
@@ -126,7 +131,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
|
||||
|
||||
@Override
|
||||
public boolean isAutoStartup() {
|
||||
return (this.container != null) ? this.container.isAutoStartup() : false;
|
||||
return (this.container != null) && this.container.isAutoStartup();
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -136,7 +141,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return (this.container != null) ? this.container.isRunning() : false;
|
||||
return (this.container != null) && this.container.isRunning();
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -158,6 +163,9 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
|
||||
if (this.container != null) {
|
||||
this.container.stop(callback);
|
||||
}
|
||||
else {
|
||||
callback.run();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -226,6 +234,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
|
||||
throw new MessagingException("failed to handle incoming JMS Message", e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2016 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -548,6 +548,9 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
|
||||
if (this.channel instanceof SubscribableJmsChannel) {
|
||||
((SubscribableJmsChannel) this.channel).stop(callback);
|
||||
}
|
||||
else {
|
||||
callback.run();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2016 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -51,6 +51,7 @@ import org.springframework.util.StringUtils;
|
||||
* @author Oleg Zhurakousky
|
||||
* @author Gary Russell
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 2.0
|
||||
*/
|
||||
@SuppressWarnings("rawtypes")
|
||||
@@ -105,7 +106,6 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
|
||||
/**
|
||||
* Specify the maximum number of subscribers supported by the
|
||||
* channel's dispatcher.
|
||||
*
|
||||
* @param maxSubscribers The maximum number of subscribers allowed.
|
||||
*/
|
||||
public void setMaxSubscribers(int maxSubscribers) {
|
||||
@@ -168,45 +168,37 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
|
||||
|
||||
@Override
|
||||
public boolean isAutoStartup() {
|
||||
return (this.container != null) && this.container.isAutoStartup();
|
||||
return this.container.isAutoStartup();
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getPhase() {
|
||||
return (this.container != null) ? this.container.getPhase() : 0;
|
||||
return this.container.getPhase();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return (this.container != null) && this.container.isRunning();
|
||||
return this.container.isRunning();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
if (this.container != null) {
|
||||
this.container.start();
|
||||
}
|
||||
this.container.start();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
if (this.container != null) {
|
||||
this.container.stop();
|
||||
}
|
||||
this.container.stop();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop(Runnable callback) {
|
||||
if (this.container != null) {
|
||||
this.container.stop(callback);
|
||||
}
|
||||
this.container.stop(callback);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroy() throws Exception {
|
||||
if (this.container != null) {
|
||||
this.container.destroy();
|
||||
}
|
||||
this.container.destroy();
|
||||
}
|
||||
|
||||
private class MessageListenerDelegate {
|
||||
@@ -230,6 +222,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
|
||||
+ "' (" + SubscribableRedisChannel.this.getFullChannelName() + ").", e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2016 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -32,7 +32,10 @@ import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Factory bean to create syslog inbound adapters (UDP or TCP).
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 3.0
|
||||
*
|
||||
*/
|
||||
@@ -163,6 +166,9 @@ public class SyslogReceivingChannelAdapterFactoryBean extends AbstractFactoryBea
|
||||
if (this.adapter != null) {
|
||||
this.adapter.stop(callback);
|
||||
}
|
||||
else {
|
||||
callback.run();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
Reference in New Issue
Block a user