MQTT integration: Possiblity to set the consumerStopAction when using a ClientManager

[t-lintanf]

Expected Behavior
In the MQTT integration when using a client manager, I need to be able to set the ConsumerStopAction. This could be achieved by adding a constructor with the possibility to provide both a ClientManager and a MqttPahoClientFactory.

Current Behavior
The ConsumerStopAction are defined on the MqttPahoClientFactory which is private. It is not possible to define the value when using a ClientManager. As it use a DefaultMqttPahoClientFactory which set it to UNSUBSCRIBE_CLEAN.

Context
I need to be able to have the messages queue up in the broker while the spring boot application restart for maintenance purposes. And i also can only use one client but I need several channelAdaptors. Currently restarting the spring application causes the loss of message that are critical for my use case.

A possible alternative would be to make a custom channelAdaptor but I feel it would be appropriate to implement it on the base class. I am willing to submit a pull request to implement those changes if they are approved.

[artembilan]

I think this property has nothing to do with the connection factory at all.
I'd like to deprecate this ConsumerStopAction and remove eventually, since cleanSession should be enough: not sure who would do cleanSession = true and then say UNSUBSCRIBE_ALWAYS.
But the cat is out of bag and we can go now only baby steps:

1. Deprecate ConsumerStopAction on the MqttPahoClientFactory
2. Expose consumerStopAction property on the MqttPahoMessageDrivenChannelAdapter
3. This way we won't need to do anything with the ClientManager API.

Let me know if that works for you!

@garyrussell , what is your thought about this ConsumerStopAction removal eventually?
The v5 channel adapter impl does not take it into account if that...

Thanks

[garyrussell]

Unfortunately, we should have deprecated this long ago and removed it in 6.0 (or even 5.0); I have no issues with doing that now (with removal in 6.1).

UNSUBSCRIBE_ALWAYS was only added to provide a path for backwards compatibility since the "fix" was introduced in a patch release...

[NOTE]
====
Prior to version 4.2.3, the client always unsubscribed when the adapter was stopped.
This was incorrect because, if the client QOS is greater than 0, we need to keep the subscription active so that messages arriving
while the adapter is stopped are delivered on the next start.
This also requires setting the `cleanSession` property on the client factory to `false`.
It defaults to `true`.

Starting with version 4.2.3, the adapter does not unsubscribe (by default) if the `cleanSession` property is `false`.

This behavior can be overridden by setting the `consumerCloseAction` property on the factory.
It can have values: `UNSUBSCRIBE_ALWAYS`, `UNSUBSCRIBE_NEVER`, and `UNSUBSCRIBE_CLEAN`.
The latter (the default) unsubscribes only if the `cleanSession` property is `true`.

To revert to the pre-4.2.3 behavior, use `UNSUBSCRIBE_ALWAYS`.
====

[artembilan]

@t-lintanf ,

What is wrong in your case to set cleanSession = false into MqttConnectOptions of the Mqttv3ClientManager?
So, you won't need to worry about this ConsumerStopAction default value.
I'm not sure what is the bug then, since we definitely can control a behavior with cleanSession even on the common ClientManager.
I'll the issue in the bug-waiting-feedback state until you come back to us with more clarifications.

I think right now we just deprecate it as is and keep the behavior until we remove it in the next 6.1.
And I guess it even would be better to deprecate it down to 5.5.x, so migration path to next version would be more comfort.

[t-lintanf]

@artembilan
Thanks for the quick response. Indeed, I do have cleanSession set to false. So after further investigation it seems that was not the source of my issue. I think the deprecation would be indeed useful to avoid someone making the same mistake that I did.

So no bug here just me who got confused. Thanks again for your time.