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

GH-3215: MQTT Event for failed connection outbound

Browse Source 
Resolves https://github.com/spring-projects/spring-integration/issues/3215

* Add docs; publish an event for an initial connection failure too.

**Cherry-pick to 5.2.x**
This commit is contained in:
Gary Russell
2020-03-17 15:09:22 -04:00
committed by Artem Bilan
parent 438e2215b3
commit 070667adc4
3 changed files with 28 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java
View File

@@ -27,6 +27,7 @@ import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
import org.springframework.integration.mqtt.event.MqttMessageDeliveredEvent;
import org.springframework.integration.mqtt.event.MqttMessageSentEvent;
import org.springframework.integration.mqtt.support.MqttMessageConverter;
@@ -201,6 +202,9 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
this.client.close();
this.client = null;
}
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, e));
}
throw new MessagingException("Failed to connect", e);
}
}
@@ -247,6 +251,9 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
// NOSONAR
}
this.client = null;
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, cause));
}
}
}

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

@@ -198,6 +198,11 @@ public class MqttAdapterTests {
verify(client, times(1)).connect(any(MqttConnectOptions.class));
assertThat(connectCalled.get()).isTrue();
AtomicReference<Object> failed = new AtomicReference<>();
handler.setApplicationEventPublisher(event -> failed.set(event));
handler.connectionLost(new IllegalStateException());
assertThat(failed.get()).isInstanceOf(MqttConnectionFailedEvent.class);
handler.stop();
}
@Test
@@ -410,6 +415,10 @@ public class MqttAdapterTests {
Thread.sleep(1000);
// the following assertion should be equalTo, but leq to protect against a slow CI server
assertThat(attemptingReconnectCount.get()).isLessThanOrEqualTo(2);
AtomicReference<Object> failed = new AtomicReference<>();
adapter.setApplicationEventPublisher(event -> failed.set(event));
adapter.connectionLost(new IllegalStateException());
assertThat(failed.get()).isInstanceOf(MqttConnectionFailedEvent.class);
adapter.stop();
taskScheduler.destroy();
}

13
src/reference/asciidoc/mqtt.adoc
View File

@@ -278,7 +278,7 @@ The default is `headers['mqtt_topic']`.
<11> When `true`, the caller does not block.
Rather, it waits for delivery confirmation when a message is sent.
The default is `false` (the send blocks until delivery is confirmed).
<12> When `async` and `async-events` are both `true`, an `MqttMessageSentEvent` is emitted.
<12> When `async` and `async-events` are both `true`, an `MqttMessageSentEvent` is emitted (See <<events>>).
It contains the message, the topic, the `messageId` generated by the client library, the `clientId`, and the `clientInstance` (incremented each time the client is connected).
When the delivery is confirmed by the client library, an `MqttMessageDeliveredEvent` is emitted.
It contains the the `messageId`, the `clientId`, and the `clientInstance`, enabling delivery to be correlated with the send.
@@ -372,3 +372,14 @@ public class MqttJavaApplication {
}
----
====
[[events]]
=== Events
Certain application events are published by the adapters.
* `MqttConnectionFailedEvent` - published by both adapters if we fail to connect or a connection is subsequently lost.
* `MqttMessageSentEvent` - published by the outbound adapter when a message has been sent, if running in asynchronous mode.
* `MqttMessageDeliveredEvent` - published by the outbound adapter when the client indicates that a message has been delivered, if running in asynchronous mode.
These events can be received by an `ApplicationListener<MqttIntegrationEvent>` or with an `@EventListener` method.