From 2d3d37ecbab649982a80bebc33fdc2a85dc3771a Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 27 Feb 2020 16:34:15 -0500 Subject: [PATCH] Upgrade to MQTT Paho 1.2.2 - 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 --- build.gradle | 2 +- .../MqttPahoMessageDrivenChannelAdapter.java | 24 +++++++++++--- .../integration/mqtt/MqttAdapterTests.java | 33 ++++++++++++------- 3 files changed, 41 insertions(+), 18 deletions(-) diff --git a/build.gradle b/build.gradle index bd8b2c45e8..0b03345e20 100644 --- a/build.gradle +++ b/build.gradle @@ -82,7 +82,7 @@ ext { mockitoVersion = '3.2.4' mongoDriverVersion = '4.0.0-beta1' mysqlVersion = '8.0.19' - pahoMqttClientVersion = '1.2.0' + pahoMqttClientVersion = '1.2.2' postgresVersion = '42.2.10' reactorVersion = 'Dysprosium-SR4' resilience4jVersion = '1.3.1' diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java index 26b1e098db..3485b7c94d 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java @@ -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); diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java index 20716f9b3c..557fcb4fc3 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java @@ -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. @@ -17,10 +17,12 @@ package org.springframework.integration.mqtt; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.assertj.core.api.Assertions.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; @@ -48,8 +50,10 @@ import javax.net.SocketFactory; import org.aopalliance.intercept.MethodInterceptor; import org.apache.commons.logging.Log; +import org.assertj.core.api.Condition; 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; @@ -435,11 +439,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"); @@ -449,15 +455,12 @@ public class MqttAdapterTests { method.set(m); }, m -> m.getName().equals("connectAndSubscribe")); assertThat(method.get()).isNotNull(); - try { - method.get().invoke(adapter); - fail("Expected InvocationTargetException"); - } - catch (InvocationTargetException e) { - assertThat(e.getCause()).isInstanceOf(MqttException.class); - assertThat(((MqttException) e.getCause()).getReasonCode()) - .isEqualTo((int) MqttException.REASON_CODE_SUBSCRIBE_FAILED); - } + Condition subscribeFailed = new Condition<>(ex -> + ((MqttException) ex.getCause()).getReasonCode() == MqttException.REASON_CODE_SUBSCRIBE_FAILED, + "expected the reason code to be REASON_CODE_SUBSCRIBE_FAILED"); + assertThatExceptionOfType(InvocationTargetException.class).isThrownBy(() -> method.get().invoke(adapter)) + .withCauseInstanceOf(MqttException.class) + .is(subscribeFailed); } @Test @@ -485,11 +488,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"); @@ -506,6 +511,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,