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

Upgrade to MQTT Paho 1.2.2

Browse Source 
- fix mock tests for internal client changes
- reduce stop wait for completion time

**cherry-pick to 5.2.x**

* Remove stack trace from test and convert to assertJ
This commit is contained in:
Gary Russell
2020-02-27 16:34:15 -05:00
parent 35fd383399
commit 4174bfd6ca
3 changed files with 33 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -127,7 +127,7 @@ subprojects { subproject ->
micrometerVersion = '1.1.8'
mockitoVersion = '2.25.1'
mysqlVersion = '8.0.16'
pahoMqttClientVersion = '1.2.0'
pahoMqttClientVersion = '1.2.2'
postgresVersion = '42.2.5'
reactorNettyVersion = '0.8.13.RELEASE'
reactorVersion = '3.2.12.RELEASE'

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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,15 +51,19 @@ import org.springframework.util.Assert;
public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDrivenChannelAdapter
implements MqttCallback, ApplicationEventPublisherAware {
public static final long DEFAULT_COMPLETION_TIMEOUT = 30000L;
public static final long DEFAULT_COMPLETION_TIMEOUT = 30_000L;
private static final int DEFAULT_RECOVERY_INTERVAL = 10000;
public static final long STOP_COMPLETION_TIMEOUT = 5_000L;
private static final int DEFAULT_RECOVERY_INTERVAL = 10_000;
private final MqttPahoClientFactory clientFactory;
private int recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
private volatile long completionTimeout = DEFAULT_COMPLETION_TIMEOUT;
private long completionTimeout = DEFAULT_COMPLETION_TIMEOUT;
private long stopCompletionTimeout = STOP_COMPLETION_TIMEOUT;
private volatile IMqttClient client;
@@ -122,6 +126,16 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
this.completionTimeout = completionTimeout;
}
/**
* Set the completion timeout wnen stopping. Not settable using the namespace.
* Default {@value #STOP_COMPLETION_TIMEOUT} milliseconds.
* @param completionTimeout The timeout.
* @since 5.2.5
*/
public void setStopCompletionTimeout(long completionTimeout) {
this.stopCompletionTimeout = completionTimeout;
}
/**
* The time (ms) to wait between reconnection attempts.
* Default {@value #DEFAULT_RECOVERY_INTERVAL}.
@@ -170,7 +184,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
logger.error("Exception while unsubscribing", e);
}
try {
this.client.disconnectForcibly(this.completionTimeout);
this.client.disconnectForcibly(this.stopCompletionTimeout);
}
catch (MqttException e) {
logger.error("Exception while disconnecting", e);

16
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -28,6 +28,7 @@ import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.BDDMockito.willReturn;
@@ -57,6 +58,7 @@ import org.aopalliance.intercept.MethodInterceptor;
import org.apache.commons.logging.Log;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
@@ -446,11 +448,13 @@ public class MqttAdapterTests {
new DirectFieldAccessor(client).setPropertyValue("aClient", aClient);
willAnswer(new CallsRealMethods()).given(client).connect(any(MqttConnectOptions.class));
willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class));
willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class),
(IMqttMessageListener[]) isNull());
willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
IMqttToken token = mock(IMqttToken.class);
given(token.getGrantedQos()).willReturn(new int[] { 0x80 });
willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), any(), any());
willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any());
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
"baz", "fix");
@@ -496,11 +500,13 @@ public class MqttAdapterTests {
new DirectFieldAccessor(client).setPropertyValue("aClient", aClient);
willAnswer(new CallsRealMethods()).given(client).connect(any(MqttConnectOptions.class));
willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class));
willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class),
(IMqttMessageListener[]) isNull());
willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
IMqttToken token = mock(IMqttToken.class);
given(token.getGrantedQos()).willReturn(new int[] { 2, 0 });
willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), any(), any());
willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any());
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
"baz", "fix");
@@ -517,6 +523,10 @@ public class MqttAdapterTests {
verify(logger, atLeastOnce())
.warn("Granted QOS different to Requested QOS; topics: [baz, fix] requested: [1, 1] granted: [2, 0]");
verify(client).setTimeToWait(30_000L);
new DirectFieldAccessor(adapter).setPropertyValue("running", Boolean.TRUE);
adapter.stop();
verify(client).disconnectForcibly(5_000L);
}
private MqttPahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttClient client, Boolean cleanSession,