createRetainedStream is per topic

[behrad]

createRetainedStream is called per subscription topic, for client's with multiple subscriptions it looks weird. e.g. redis persistence currently runs hget retained n times for a client connecting with n offline subscriptions.
Can we

1. change createRetainedStream(topic) to createRetainedStream(topics) in persistence api.
2. move sendRetained calls here to the end of completeSubscribe called with the whole packet.subscriptions.map(s=>s.topic)

[mcollina]

How big is n in your experience? It would be nice to have some numbers before committing.

This would need to be changed in aedes-persistence, and then downstream to all implementors. I'm 👍 if you want to tackle this one.

[behrad]

How big is n in your experience?

1. n >= 3 for each of my clients
2. Bad news is that sendRetained runs at the end of handle subscribe, which is called on each CONNECT (on restore sub). So even if clients don't SUBSCRIBE, retained check will be done on each connect.
   And in our environment we have a ratio of 100 conn/disconn per sec!

and please clarify something for me, I had checked mqtt spec some time ago, but can't remember if we should sendRetained both on CONNECT/SUBSCRIBE, or only as a response to a SUBSCRIBE command? I mean when a client SUBSCRIBEs with qos=1 and clean=false, should we send retained on each subsequent feature connects only?

[behrad]

we had a somehow related disscusion here @mcollina, However I'm not sure my statement there was right!

[mcollina]

had checked mqtt spec some time ago, but can't remember if we should sendRetained both on CONNECT/SUBSCRIBE, or only as a response to a SUBSCRIBE command?

@behrad the retained check is part of the mqtt implementation decisions :(. Mosquitto implements it the same way we do, resending the retained message on connect. However, the spec is unclear to me.
You might want to ask them, maybe.

@behrad go ahead and start making the changes. However, I would implement it in as a non-breaking changes. Adding a new createdRetainedStreamCombi(topics) which in the default implementation calls createRetainedStream(topic), so that we do not have to implement it for all the persistences right now.

[behrad]

handled in #121