GH-3685: Add docs for shared MQTT client feature

Related to https://github.com/spring-projects/spring-integration/issues/3685

* Add documentation for a new MQTT shared client feature

Add an overview with reason for the feature as well as basic
capabilities listing. Give an example with Java DSL usage for several
adapters.

* Fill `whats-new.adoc` with MQTT changes

Add a reference to MQTT documentation with info about shared MQTT client

* Couple of code review changes
This commit is contained in:
Artem Vozhdayenko
2022-08-16 20:00:18 +03:00
committed by GitHub
parent 2bfcb32628
commit 20cebfca11
2 changed files with 62 additions and 1 deletions

View File

@@ -500,4 +500,62 @@ 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 `Mqttv5PahoMessageDrivenChannelAdapter` can handle reconnects, e.g. via `MqttConnectionFailedEvent` handling on disconnection.
Otherwise, only the manual restart of `Mqttv5PahoMessageDrivenChannelAdapter` can handle reconnects, e.g. via `MqttConnectionFailedEvent` handling on disconnection.
[[mqtt-shared-client]]
=== Shared MQTT Client Support
If a single MQTT ClientID is required for several integrations, multiple MQTT client instances cannot be used because MQTT brokers may have a limitation on a number of connections per ClientID (typically, a single connection is allowed).
For having a single client reused for different channel adapters, a `org.springframework.integration.mqtt.core.ClientManager` component may be used and passed to any channel adapter needed.
It will manage MQTT connection lifecycle and do automatic reconnects if needed.
Also, a custom connection options and `MqttClientPersistence` may be provided to the client manager just as currently it can be done for channel adapter components.
Note that both MQTT v5 and v3 channel adapters are supported.
The following Java DSL configuration sample demonstrates how to use this client manager in the integration flow:
====
[source,java]
----
@Bean
public ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager() {
MqttConnectionOptions connectionOptions = new MqttConnectionOptions();
connectionOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
connectionOptions.setConnectionTimeout(30000);
connectionOptions.setMaxReconnectDelay(1000);
connectionOptions.setAutomaticReconnect(true);
Mqttv5ClientManager clientManager = new Mqttv5ClientManager(connectionOptions, "client-manager-client-id-v5");
clientManager.setPersistence(new MqttDefaultFilePersistence());
return clientManager;
}
@Bean
public IntegrationFlow mqttInFlowTopic1(
ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager) {
Mqttv5PahoMessageDrivenChannelAdapter messageProducer =
new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, "topic1");
return IntegrationFlow.from(messageProducer)
.channel(c -> c.queue("fromMqttChannel"))
.get();
}
@Bean
public IntegrationFlow mqttInFlowTopic2(
ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager) {
Mqttv5PahoMessageDrivenChannelAdapter messageProducer =
new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, "topic2");
return IntegrationFlow.from(messageProducer)
.channel(c -> c.queue("fromMqttChannel"))
.get();
}
@Bean
public IntegrationFlow mqttOutFlow(
ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager) {
return f -> f.handle(new Mqttv5PahoMessageHandler(clientManager));
}
----
====

View File

@@ -17,6 +17,9 @@ In general the project has been moved to Java 17 baseline and migrated from Java
[[x6.0-new-components]]
=== New Components
A new MQTT `ClientManager` has been added to support a reusable MQTT connection across different channel adapters.
See <<./mqtt.adoc#mqtt-shared-client,Shared MQTT Client Support>> for more information.
[[x6.0-graphql]]
=== GraphQL Support