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

GH-3822: Reconnection for MQTTv5 channel adapters

Browse Source 
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`**
This commit is contained in:
Lucas Bowler
2022-06-09 18:08:04 -04:00
committed by Artem Bilan
parent cf6ce961a2
commit 8cf5f9083b
5 changed files with 198 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
View File

@@ -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

26
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java
View File

@@ -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(

13
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java
View File

@@ -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;
}
}

141
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackAutomaticReconnectTests.java Normal file
View File

@@ -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<String> 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();
}
}
}