From 8cf5f9083bd7834ca4d7f4956338a445b4d76ca3 Mon Sep 17 00:00:00 2001 From: Lucas Bowler Date: Thu, 9 Jun 2022 18:08:04 -0400 Subject: [PATCH] GH-3822: Reconnection for MQTTv5 channel adapters Fixes: https://github.com/spring-projects/spring-integration/issues/3822 * Apply spring-framework code style on modified class * Remove unwanted formatting * Take pull request comments into account * Code and JavaDocs clean up * Improve `Mqttv5BackToBackAutomaticReconnectTests` removing non-related code * Improve `mqtt.adoc` for this new manual reconnection feature **Cherry-pick to `5.5.x`** --- ...Mqttv5PahoMessageDrivenChannelAdapter.java | 57 ++++--- .../outbound/Mqttv5PahoMessageHandler.java | 26 +--- .../mqtt/BackToBackAdapterTests.java | 13 +- ...ttv5BackToBackAutomaticReconnectTests.java | 141 ++++++++++++++++++ src/reference/asciidoc/mqtt.adoc | 6 +- 5 files changed, 198 insertions(+), 45 deletions(-) create mode 100644 spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackAutomaticReconnectTests.java diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java index 84fd94b709..aaab5792f0 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java @@ -66,6 +66,7 @@ import org.springframework.util.Assert; * * @author Artem Bilan * @author Mikhail Polivakha + * @author Lucas Bowler * * @since 5.5.5 * @@ -162,28 +163,24 @@ public class Mqttv5PahoMessageDrivenChannelAdapter extends AbstractMqttMessageDr @Override protected void doStart() { ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); - this.topicLock.lock(); - String[] topics = getTopic(); try { this.mqttClient.connect(this.connectionOptions).waitForCompletion(getCompletionTimeout()); - if (topics.length > 0) { - int[] requestedQos = getQos(); - this.mqttClient.subscribe(topics, requestedQos).waitForCompletion(getCompletionTimeout()); - String message = "Connected and subscribed to " + Arrays.toString(topics); - logger.debug(message); - if (applicationEventPublisher != null) { - applicationEventPublisher.publishEvent(new MqttSubscribedEvent(this, message)); - } - } } catch (MqttException ex) { - if (applicationEventPublisher != null) { - applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, ex)); + if (this.connectionOptions.isAutomaticReconnect()) { + try { + this.mqttClient.reconnect(); + } + catch (MqttException e) { + logger.error(ex, "MQTT client failed to connect. Never happens."); + } + } + else { + if (applicationEventPublisher != null) { + applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, ex)); + } + logger.error(ex, "MQTT client failed to connect."); } - logger.error(ex, () -> "Error connecting or subscribing to " + Arrays.toString(topics)); - } - finally { - this.topicLock.unlock(); } } @@ -312,7 +309,31 @@ public class Mqttv5PahoMessageDrivenChannelAdapter extends AbstractMqttMessageDr @Override public void connectComplete(boolean reconnect, String serverURI) { - + if (!reconnect) { + ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); + String[] topics = getTopic(); + this.topicLock.lock(); + try { + if (topics.length > 0) { + int[] requestedQos = getQos(); + this.mqttClient.subscribe(topics, requestedQos).waitForCompletion(getCompletionTimeout()); + String message = "Connected and subscribed to " + Arrays.toString(topics); + logger.debug(message); + if (applicationEventPublisher != null) { + applicationEventPublisher.publishEvent(new MqttSubscribedEvent(this, message)); + } + } + } + catch (MqttException ex) { + if (applicationEventPublisher != null) { + applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, ex)); + } + logger.error(ex, () -> "Error subscribing to " + Arrays.toString(topics)); + } + finally { + this.topicLock.unlock(); + } + } } @Override diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java index 10318d98c5..46908724b6 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2021 the original author or authors. + * Copyright 2021-2022 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. @@ -50,13 +50,8 @@ import org.springframework.util.Assert; /** * The {@link AbstractMqttMessageHandler} implementation for MQTT v5. * - * It is recommended to have the {@link MqttConnectionOptions#setAutomaticReconnect(boolean)} - * set to true to let an internal {@link IMqttAsyncClient} instance to handle reconnects. - * Otherwise, only the manual restart of this component can handle reconnects, e.g. via - * {@link MqttConnectionFailedEvent} handling on disconnection. - * - * * @author Artem Bilan + * @author Lucas Bowler * * @since 5.5.5 */ @@ -86,11 +81,6 @@ public class Mqttv5PahoMessageHandler extends AbstractMqttMessageHandler public Mqttv5PahoMessageHandler(MqttConnectionOptions connectionOptions, String clientId) { super(obtainServerUrlFromOptions(connectionOptions), clientId); this.connectionOptions = connectionOptions; - if (!this.connectionOptions.isAutomaticReconnect()) { - logger.warn("It is recommended to set 'automaticReconnect' MQTT client option. " + - "Otherwise the current channel adapter restart should be used explicitly, " + - "e.g. via handling 'MqttConnectionFailedEvent' on client disconnection."); - } } @@ -165,11 +155,7 @@ public class Mqttv5PahoMessageHandler extends AbstractMqttMessageHandler this.mqttClient.connect(this.connectionOptions).waitForCompletion(getCompletionTimeout()); } catch (MqttException ex) { - ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); - if (applicationEventPublisher != null) { - applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, ex)); - } - logger.error(ex, "MQTT client failed to connect. Will retry if 'ConnectionOptions.isAutomaticReconnect()'."); + logger.error(ex, "MQTT client failed to connect."); } } @@ -249,11 +235,15 @@ public class Mqttv5PahoMessageHandler extends AbstractMqttMessageHandler @Override protected void publish(String topic, Object mqttMessage, Message message) { Assert.isInstanceOf(MqttMessage.class, mqttMessage, "The 'mqttMessage' must be an instance of 'MqttMessage'"); + long completionTimeout = getCompletionTimeout(); try { + if (!this.mqttClient.isConnected()) { + this.mqttClient.connect(this.connectionOptions).waitForCompletion(completionTimeout); + } IMqttToken token = this.mqttClient.publish(topic, (MqttMessage) mqttMessage); ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); if (!this.async) { - token.waitForCompletion(getCompletionTimeout()); // NOSONAR (sync) + token.waitForCompletion(completionTimeout); // NOSONAR (sync) } else if (this.asyncEvents && applicationEventPublisher != null) { applicationEventPublisher.publishEvent( diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java index 1d24296c7a..fb6e78be7d 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java @@ -105,8 +105,8 @@ public class BackToBackAdapterTests implements MosquittoContainerTest { adapter.setBeanFactory(mock(BeanFactory.class)); adapter.afterPropertiesSet(); adapter.start(); - MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(), - "si-test-in", "mqtt-foo"); + MqttPahoMessageDrivenChannelAdapter inbound = + new MqttPahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(), "si-test-in", "mqtt-foo"); QueueChannel outputChannel = new QueueChannel(); inbound.setOutputChannel(outputChannel); inbound.setTaskScheduler(taskScheduler); @@ -467,14 +467,11 @@ public class BackToBackAdapterTests implements MosquittoContainerTest { } Foo other = (Foo) obj; if (this.bar == null) { - if (other.bar != null) { - return false; - } + return other.bar == null; } - else if (!this.bar.equals(other.bar)) { - return false; + else { + return this.bar.equals(other.bar); } - return true; } } diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackAutomaticReconnectTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackAutomaticReconnectTests.java new file mode 100644 index 0000000000..c5a741f5e2 --- /dev/null +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackAutomaticReconnectTests.java @@ -0,0 +1,141 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.mqtt; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; + +import java.net.UnknownHostException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.eclipse.paho.mqttv5.client.MqttConnectionOptions; +import org.eclipse.paho.mqttv5.client.MqttConnectionOptionsBuilder; +import org.eclipse.paho.mqttv5.common.MqttException; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.EventListener; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.mqtt.event.MqttSubscribedEvent; +import org.springframework.integration.mqtt.inbound.Mqttv5PahoMessageDrivenChannelAdapter; +import org.springframework.integration.mqtt.outbound.Mqttv5PahoMessageHandler; +import org.springframework.integration.mqtt.support.MqttHeaders; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + + +/** + * @author Lucas Bowler + * @author Artem Bilan + * + * @since 5.5.13 + */ +@SpringJUnitConfig +@DirtiesContext +public class Mqttv5BackToBackAutomaticReconnectTests implements MosquittoContainerTest { + + @Autowired + @Qualifier("mqttOutFlow.input") + private MessageChannel mqttOutFlowInput; + + @Autowired + private PollableChannel fromMqttChannel; + + @Autowired + private MqttConnectionOptions connectionOptions; + + @Autowired + Config config; + + @Test + public void testReconnectionWhenFirstConnectionFails() throws InterruptedException { + Message testMessage = + MessageBuilder.withPayload("testPayload") + .setHeader(MqttHeaders.TOPIC, "siTest") + .build(); + + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> this.mqttOutFlowInput.send(testMessage)) + .withCauseExactlyInstanceOf(MqttException.class) + .withRootCauseExactlyInstanceOf(UnknownHostException.class); + + connectionOptions.setServerURIs(new String[]{ MosquittoContainerTest.mqttUrl() }); + + assertThat(this.config.subscribeLatch.await(10, TimeUnit.SECONDS)).isTrue(); + + this.mqttOutFlowInput.send(testMessage); + + Message receive = this.fromMqttChannel.receive(10_000); + + assertThat(receive).isNotNull(); + } + + + @Configuration + @EnableIntegration + public static class Config { + + CountDownLatch subscribeLatch = new CountDownLatch(1); + + @Bean + public MqttConnectionOptions mqttConnectOptions() { + return new MqttConnectionOptionsBuilder() + .serverURI("wss://badMqttUrl") + .automaticReconnect(true) + .connectionTimeout(1) + .build(); + } + + @Bean + public IntegrationFlow mqttOutFlow() { + Mqttv5PahoMessageHandler messageHandler = + new Mqttv5PahoMessageHandler(mqttConnectOptions(), "mqttv5SIout"); + + return f -> f.handle(messageHandler); + } + + + @Bean + public IntegrationFlow mqttInFlow() { + Mqttv5PahoMessageDrivenChannelAdapter messageProducer = + new Mqttv5PahoMessageDrivenChannelAdapter(mqttConnectOptions(), "mqttv5SIin", "siTest"); + messageProducer.setPayloadType(String.class); + + return IntegrationFlows.from(messageProducer) + .channel(c -> c.queue("fromMqttChannel")) + .get(); + } + + @EventListener(MqttSubscribedEvent.class) + void mqttEvents() { + this.subscribeLatch.countDown(); + } + + } + +} diff --git a/src/reference/asciidoc/mqtt.adoc b/src/reference/asciidoc/mqtt.adoc index 012e4f4e53..d0bffc87c1 100644 --- a/src/reference/asciidoc/mqtt.adoc +++ b/src/reference/asciidoc/mqtt.adoc @@ -460,6 +460,10 @@ public IntegrationFlow mqttOutFlow() { IMPORTANT: The `org.springframework.integration.mqtt.support.MqttMessageConverter` cannot be used with the `Mqttv5PahoMessageHandler` since its contract is aimed only for the MQTT v3 protocol. +If connection fails on start up or at runtime, the `Mqttv5PahoMessageHandler` tries to reconnect on the next message produced to this handler. +If this manual reconnection fails, the connection is exception is thrown back to the caller. +In this case the standard Spring Integration error handling procedure is applied, including request handler advices, e.g. retry or circuit breaker. + See more information in the `Mqttv5PahoMessageHandler` javadocs and its superclass. The inbound channel adapter for the MQTT v5 protocol is present as an `Mqttv5PahoMessageDrivenChannelAdapter`. @@ -496,4 +500,4 @@ IMPORTANT: The `org.springframework.integration.mqtt.support.MqttMessageConverte See more information in the `Mqttv5PahoMessageDrivenChannelAdapter` javadocs and its superclass. IMPORTANT: It is recommended to have the `MqttConnectionOptions#setAutomaticReconnect(boolean)` set to true to let an internal `IMqttAsyncClient` instance to handle reconnects. -Otherwise, only the manual restart of these channel adapters can handle reconnects, e.g. via `MqttConnectionFailedEvent` handling on disconnection. \ No newline at end of file +Otherwise, only the manual restart of `Mqttv5PahoMessageDrivenChannelAdapter` can handle reconnects, e.g. via `MqttConnectionFailedEvent` handling on disconnection. \ No newline at end of file