Postgres backend: handle connection loss

[asafc]

This bugfix handle a situation where broadcaster does not detect the Postgres backend terminating the connection.

A user of Broadcaster will not be able to handle disconnects in their code.

i.e: the next loop will not exit:

async def chatroom_ws_sender(websocket):
    async with broadcast.subscribe(channel="chatroom") as subscriber:
        async for event in subscriber:
            await websocket.send_text(event.message)

when either:

1. the Postgres db is down
2. Postgres terminated an idle session (long lived session) due to either idle_session_timeout or idle_in_transaction_session_timeout (see here)

[dariosky]

Hi @asafc - this was a good start.
It is missing the handling of the disconnection though. I ended up playing with this for a bit and change both the base and the postgres backend to reconnect and resubscribe to the channels so it works for my use-case.
Let me know if you want me to add to this merge-request.

[roekatz]

Hi @dariosky,
I was involved in this fix on our fork https://github.com/permitio/broadcaster.

I don't believe automatically reconnecting is a good idea: when connection is loss messages could be missed,
therefor the caller should be notified - any initialization required to catch up with potentially missing updates.

On our use case - disconnecting (marked by ending the iterator) is more than enough (then the caller can reconnect).
It's also possible to have some on_reconnect callback, so caller could be notified while connection is re-established - but IMO it doesn't really add a lot of value :)

WDYT?

[dariosky]

Having an on_reconnect parameter is a good idea - so the caller can handle the eventual gap-filling...
In my case I didn't need it - I had just a reconnect on-error loop and was keeping the memory of the subscriptions to restore them when the reconnection happened...

[roekatz]

Not sure I understand - have you pushed that fix? (or plan on pushing it?)

[dariosky]

I didn't (I asked above if it can be useful) - let me add it here :)

[dariosky]

See here: permitio#4

[logankaser]

# Unsubscribe all
for queue_set in self._subscribers.values():
    for queue in queue_set:
        await queue.put(None)

Seems like a lot less wasteful / clear what's going on.
As far as I know, you can loop over sets directly in python, no need
to convert to a list.

[alex-oleshkevich]

This requires extra testing. Will not include in 0.3