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

INT-3457 Fix stop(Runnable) in AbstractEndpoint

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

INT-3486 changed stop(Runnable) so that it could be
overridden by subclasses, to allow separation of the
`stop()` and `callback` invocation.

However, the refactoring changed the logic such that
the `running` field is not reset, causing `doStop()` to
be called even when the component was not running.

Reset the running field.

Also, the MQTT inbound channel adapter did not test
for a `null` `client` in `doStop()`.
This commit is contained in:
Gary Russell
2014-11-04 10:01:43 -05:00
parent f2ca6cdd06
commit 7dcb6ce58f
2 changed files with 40 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -35,6 +35,7 @@ import org.springframework.scheduling.TaskScheduler;
*
* @author Mark Fisher
* @author Kris Jacyna
* @author Gary Russell
*/
public abstract class AbstractEndpoint extends IntegrationObjectSupport implements SmartLifecycle {
@@ -57,20 +58,24 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
this.phase = phase;
}
@Override
public void setTaskScheduler(TaskScheduler taskScheduler) {
super.setTaskScheduler(taskScheduler);
}
// SmartLifecycle implementation
@Override
public final boolean isAutoStartup() {
return this.autoStartup;
}
@Override
public final int getPhase() {
return this.phase;
}
@Override
public final boolean isRunning() {
this.lifecycleLock.lock();
try {
@@ -81,6 +86,7 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
}
}
@Override
public final void start() {
this.lifecycleLock.lock();
try {
@@ -97,6 +103,7 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
}
}
@Override
public final void stop() {
this.lifecycleLock.lock();
try {
@@ -113,10 +120,17 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
}
}
@Override
public final void stop(Runnable callback) {
this.lifecycleLock.lock();
try {
doStop(callback);
if (this.running) {
doStop(callback);
this.running = false;
if (logger.isInfoEnabled()) {
logger.info("stopped " + this);
}
}
}
finally {
this.lifecycleLock.unlock();
@@ -128,8 +142,8 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
* @param callback the Runnable to invoke.
*/
protected void doStop(Runnable callback) {
doStop();
callback.run();
doStop();
callback.run();
}
/**

44
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
View File

@@ -119,28 +119,30 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
protected void doStop() {
this.cancelReconnect();
super.doStop();
try {
this.client.unsubscribe(this.getTopic())
.waitForCompletion(this.completionTimeout);
if (this.client != null) {
try {
this.client.unsubscribe(this.getTopic())
.waitForCompletion(this.completionTimeout);
}
catch (MqttException e) {
logger.error("Exception while unsubscribing", e);
}
try {
this.client.disconnect()
.waitForCompletion(this.completionTimeout);
}
catch (MqttException e) {
logger.error("Exception while disconnecting", e);
}
try {
this.client.close();
}
catch (MqttException e) {
logger.error("Exception while closing", e);
}
this.connected = false;
this.client = null;
}
catch (MqttException e) {
logger.error("Exception while unsubscribing", e);
}
try {
this.client.disconnect()
.waitForCompletion(this.completionTimeout);
}
catch (MqttException e) {
logger.error("Exception while disconnecting", e);
}
try {
this.client.close();
}
catch (MqttException e) {
logger.error("Exception while closing", e);
}
this.connected = false;
this.client = null;
}
@Override