spring-projectsGH-3822: Take pull request comments into account

Author: Lucas Bowler

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

@@ -172,7 +172,7 @@ protected void doStart() {
 					this.mqttClient.reconnect();
 				}
 				catch (MqttException e) {
-					logger.error(ex, "MQTT client failed to connect. Will retry.");
+					logger.error(ex, "MQTT client failed to connect. Never happens.");
 				}
 			}
 			else {

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

@@ -166,21 +166,7 @@ protected void doStart() {
 			this.mqttClient.connect(this.connectionOptions).waitForCompletion(getCompletionTimeout());
 		}
 		catch (MqttException ex) {
-			if (this.connectionOptions.isAutomaticReconnect()) {
-				try {
-					this.mqttClient.reconnect();
-				}
-				catch (MqttException e) {
-					logger.error(ex, "MQTT client failed to connect. Will retry.");
-				}
-			}
-			else {
-				ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();
-				if (applicationEventPublisher != null) {
-					applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, ex));
-				}
 				logger.error(ex, "MQTT client failed to connect.");
-			}
 		}
 	}
 
@@ -261,6 +247,9 @@ else if (payload instanceof String) {
 	protected void publish(String topic, Object mqttMessage, Message<?> message) {
 		Assert.isInstanceOf(MqttMessage.class, mqttMessage, "The 'mqttMessage' must be an instance of 'MqttMessage'");
 		try {
+			if (!this.mqttClient.isConnected()) {
+				this.mqttClient.connect(this.connectionOptions).waitForCompletion(getCompletionTimeout());
+			}
 			IMqttToken token = this.mqttClient.publish(topic, (MqttMessage) mqttMessage);
 			ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();
 			if (!this.async) {

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackAutomaticReconnectTests.java

@@ -18,12 +18,15 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.net.UnknownHostException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 
 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.Assertions;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -47,6 +50,7 @@
 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.MessageHeaders;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.converter.AbstractMessageConverter;
@@ -55,7 +59,6 @@
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 
-
 /**
  * @author Lucas Bowler
  * @since 5.5.x
@@ -80,11 +83,22 @@ public class Mqttv5BackToBackAutomaticReconnectTests implements MosquittoContain
 
 	@Test //GH-3822
 	public void testReconnectionWhenFirstConnectionFails() throws InterruptedException {
+		String testPayload = "foo";
+
+		MessageHandlingException messageHandlingException = Assertions.assertThrows(MessageHandlingException.class, () ->
+				this.mqttOutFlowInput.send(
+						MessageBuilder.withPayload(testPayload)
+								.setHeader(MqttHeaders.TOPIC, "siTest")
+								.setHeader("foo", "bar")
+								.setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
+								.build())
+		);
+		Assertions.assertInstanceOf(MqttException.class, messageHandlingException.getCause());
+		Assertions.assertInstanceOf(UnknownHostException.class, messageHandlingException.getRootCause());
+
 		connectionOptions.setServerURIs(new String[] {MosquittoContainerTest.mqttUrl()});
 		Thread.sleep(2_500);
 
-		String testPayload = "foo";
-
 		this.mqttOutFlowInput.send(
 				MessageBuilder.withPayload(testPayload)
 						.setHeader(MqttHeaders.TOPIC, "siTest")