Only retain non-closed subscriptions on reconnect

[caspervonb]

If a subscriber is dropped during reconnect, it likely it never fired the unsubscribe command. But we know it closed so we can just not ever re-create the subscription.

[Jarema]

Are you sure?
We first remove the subscription, then try to talk to the socket.
Should be dropped, right?

[Jarema]

                if let Some(subscription) = self.subscriptions.get_mut(&sid) {
                    subscription.max = max;
                    match subscription.max {
                        Some(n) => {
                            if subscription.delivered >= n {
                                self.subscriptions.remove(&sid);
                            }
                        }
                        None => {
                            self.subscriptions.remove(&sid);
                        }
                    }

                    if let Err(err) = self
                        .connection
                        .write_op(ClientOp::Unsubscribe { sid, max })
                        .await
                    {
                        println!("Send failed with {:?}", err);
                    }
                }

or I'm tired and missing the point? 😄

[caspervonb]

Are you sure?
We first remove the subscription, then try to talk to the socket.
Should be dropped, right?

Some cases where it will fail, e.g if we're reconnecting when its being dropped then the command can just never be dispatched if the command buffer is full.

That's fine, if we just don't do the initial re-subscription.

or I'm tired and missing the point?

Short circuit, when reconnecting, there's no point in re-subscribing to a subscription which has no receiver as it would go no-where. We can just not subscribe as we can easily check and it's a safe spot to do so.

Otherwise a sub on reconnect would be:

SUB 1
MSG 1
UNSUB 1

TL,DR; drop isn't infallible, but there is no reason to re-subscribe causing a chain of 3 protocol messages to a thing we know is already dropped.

[Jarema]

LGTM!