Share MQTT connection between inbound & outbound `IntegrationFlow`s

[durimkryeziu]

Expected Behavior
If there are two IntegrationFlows that use the same ClientFactory, to be able to share the same MQTT connection, hence being able to publish and subscribe with the same client id

Current Behavior
When I create two IntegartionFlows to publish & subscribe to an MQTT Broker, first the subscribe (inbound) flow connects and subscribes successfully to the given topic. Then when a message is published to a topic, the previous connection drops so the publishing (outbound) flow connects!

Relevant logs:

DEBUG 8950 --- [           main] .m.i.MqttPahoMessageDrivenChannelAdapter : Connected and subscribed to [notified]
 INFO 8950 --- [           main] c.e.m.MqttSharedConnectionApplication    : Publishing a message to MQTT Broker...
DEBUG 8950 --- [           main] o.s.i.m.outbound.MqttPahoMessageHandler  : Client connected
ERROR 8950 --- [c: sameClientId] .m.i.MqttPahoMessageDrivenChannelAdapter : Lost connection: Connection lost; retrying...
DEBUG 8950 --- [   scheduling-1] .m.i.MqttPahoMessageDrivenChannelAdapter : Attempting reconnect
ERROR 8950 --- [c: sameClientId] o.s.i.m.outbound.MqttPahoMessageHandler  : Lost connection; will attempt reconnect on next request
DEBUG 8950 --- [   scheduling-1] .m.i.MqttPahoMessageDrivenChannelAdapter : Connected and subscribed to [notified]

So basically they ping-pong with each other as MQTT allows only one connection per client id

Context
My use-case is that I have to use the same client id both for subscribing and publishing messages to the MQTT broker

Sample project: https://github.com/durimkryeziu/mqtt-shared-connection

[artembilan]

This request is not OK in the context of those channel adapters.
They do this:

this.client = this.clientFactory.getClientInstance(getUrl(), getClientId());
this.client.setCallback(this);

And that callback is really unique for each channel adapter.
Even if we could have a shared IMqttClient instance between channel adapters, we still cannot have a single MqttCallback according an internal logic of MqttPahoMessageDrivenChannelAdapter and MqttPahoMessageHandler.
And now imaging if you would like to share the same client between many similar channel adapters: the callback behavior would become more messy...

So, it is probably OK to have the same client for publishing an subscribing logic in the same place: https://stackoverflow.com/questions/49484629/mqtt-publish-subscribe-in-the-same-client.
But only when we do everything manually.
If we try to rely on some components from some library, it is really hard to be consistent with the shared objects.

How about having a client id as prefixed: sameClientId.subscriber and sameClientId.publisher?
So, you still will be able to distinguish connected client on your broker.

[durimkryeziu]

Hi @artembilan, thanks for your fast response :)

Thanks, I checked that StackOverflow answer but I'd like to achieve it using the APIs of Spring Integration, as we have other integrations implemented with it on that project!

Having client id prefixed... well then it means 2 MQTT connections for each client, which is a waste of resources, isn't it?

[artembilan]

As I said: it was designed like that to keep the proper track of MqttCallback.
We may revise this in the upcoming 6.0 generation: I see there is an API like this:

 void subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) throws MqttException;

So, for every new subscriber we can have its own listening context while keeping the rest of MqttCallback in the central place, e.g. MqttClientManager.
Probably something similar what we have so far with the StompSessionManager or ClientWebSocketContainer.
As far as I'm concerned I see that on publishing side the MqttClient doesn't have a state, but just gives us a DeliveryToken to track publication.

So, yeah: let's see what we can do so far in the next version!

For now you only have a choice or use different clients or implement a logic manually.

[Dacesilian]

Hello, I would like to ask if there is any solution to this, is it? MQTT allows only one connection per clientId, so my client gets disconnected. Thank you.

[artembilan]

@Dacesilian ,

we definitely are looking into this, but since there are expected some breaking changes, the fix is likely would make it only to the next 6.0 version which GA is only due this fall.

As a workaround consider to do such an MQTT logic manually:

1. Use subscribe() in your code and call some @MessagingGateway from the listener callback.
2. The publish() could be used from any POJO method which then can be just referenced from the service activator.