[MQTT] Reestablish subscriptions after reconnect

[ffendt]

Hi there,

I am a developer of Eclipse Ditto which uses akka-stream-alpakka-mqtt and think I found a little bug concerning its reconnecting functionality.

The issue seems to be that after a reconnect (because e.g. the broker was unreachable) the subscriptions aren't reestablished. So after a reconnect I am not able to receive messages any longer (publishing will work though).

When debugging my application, I can see that MqttFlowStage#connectComplete callback is called by the paho client after a successful reconnect.
The paho team suggests to resubscribe in the callback. As far as I understand (with my very little Scala knowledge) the subscriptions are only established once in MqttFlowStage#onConnect which is called after the initial connect (see MqttFlowStage#preStart ).

In the gitter chat, @2m suggested to move the subscription logic to connectComplete. As my company has relatively strict rules on contributing code, I don't know yet if or when I will be allowed to provide a PR for this.

Best regards
Florian

[2m]

Thanks for the ticket. As a workaround one could try disabling Paho reconnect logic and instead use RestartSource from akka-stream.

Another workaround is to use Alpakka Mqtt Streaming connector, however that connector provides quite a low level API (different API is being worked on in #1565)

[huntc]

Should you have to re-subscribe given the continuation of a session? I don’t recall the spec stating that you should need to as the previous subscriptions should be continued.

[2m]

@ffendt have you tried MqttConnectionSettings.withCleanSession(session)?

[ffendt]

The mqtt streaming connector currently is no real option for us, since as you said it only provides a very low level API which we don't want to use as of now.

I also stumbled upon the cleanSession setting which I also tried out and failed. But while writing I think that might have been caused by how I'm reproducing the problem. Currently I am testing against docker.io/eclipse-mosquitto:1.4.12 by simply stopping and starting the container. But that will probably destroy the session on the broker, which would be a reason why the subscriptions aren't available after a reconnect. I'll try to find a way to test this without killing the broker. Thanks for now, i'll get back to you as soon as I know more.

[ffendt]

Thanks @2m and thanks @huntc,

setting the client to not create a new session worked when testing without killing the broker (AWS IoT).

I'll close the issue then.