feat(realtime): realtime major overhaul

[o-santi]

What kind of change does this PR introduce?

Rewrite realtime from scratch, aiming at removing most callbacks from the core library, by instead offering an event iterator based library, upon which callbacks can more safely be built upon.

The current library suffers from 3 problems that hinder usability:

1. There is no simple way to know which events arrived happened first, as the callbacks can be called at any time. When the user wants to ensure some particular order he needs to shove a locking mechanism into the callback and wait for it on the outside. This can get quite complicated.
2. All exceptions that happen on background tasks are ignored. This includes exceptions thrown from callbacks, as well as from parsing and sending messages in behalf of the user.
3. Automatic re connection logic happens in background threads with no stack to bubble up errors. This can cause mysterious behavior if the background connection reaches max_retries and silently dies after hours or days of normal use. Automatic reconnection in the background also has weird semantic problems, like not knowing if a message may have been sent twice, or

This PR proposes to get rid of all of these behaviors, by doing as much as possible outside of background threads. I believe this can be mostly solved by offering fewer and simpler building blocks using iterators; ie. instead of

channel.on_broadcast(my_broadcast_callback)
channel.on_presence(my_presence_callback)
subscribe_event = asyncio.Event()
channel.on_subscribe(lambda _: subscribe.set())

await subscribe_event.wait()

await channel.send_broadcast("my-event", "i'm subscribed")

I propose doing:

subscribe_ack = await channel.subscribe()

send_broadcast_ack = await channel.send_broadcast("my-event", "i'm subscribed")

async for message in channel.messages():
    if isinstance(message, BroadcastMessage):
         my_broadcast_callback(message)
    elif isinstance(message, PresenceMessage):
         my_presence_callback(message)

which is both easier to understand and has explicit ordering between events

Core design decisions

Listen and heartbeat still happen in the background

The idea is to offer per-channel message listeners, but internally we only hold one websocket connection. This means that the listen task needs to happen in the background in order to multiplex the messages and route them to the correct channel object. Because of this, we still need a way to bubble up errors in these background tasks.

The proposed solution is to have a signaling asyncio.Future called self._connection_failure, that signals if an error happened in either the listen or the heartbeat task. Then, each message that reads from the websocket connection (be it waiting for an ack message or listening for messages on a channel) must poll both futures (the receive one and the failure future) at once, in order to bubble up the exception if it happens.

Exceptions may bubble up on all await points

In order to not let connection failures be silently ignored, the library intends to raises any connection errors (in fact, any un-checked exception that may happen at all) to the user. As it may happen in the background at any point, it needs to have a stack where it can raise them. This means that, if the user is not currently polling the library, errors may already have happened and will be thrown the next time he tries to interact with it, be it by listening to messages or by trying to send one.

This also means that re-connection now needs to be handled with more care, as it is not being handled implicitly anymore; which brings up the next point:

Automatic re connection is not entirely solved

I'm not sure that the current behavior of automatically reconnecting when a connection occurs is the best behavior to have as default in the library. The problem here is largely that, when an error occurs and the client automatically reconnects, it is not in the same state as before: all channels are gone.

The way the library works today is to pretend that nothing happened and try to reconnect in the background to all channels were marked as joined before the disconnect occurred, and continue from where it left off. This has some weird edge cases, where some messages may be dropped entirely if the connection error was raised inside a send_* function. Furthermore, the more pressing issue is that it pretends that all channel connections are entirely stateless, which might not be the case for all applications.

If a user wants to ensure that a special method is called or a particular message is sent whenever a user joins or leaves a channel, it is not possible today. This is because there is no way to know from the outside when a channel is being rejoined in the background, or was left due to a connection error. All of this implies that trying to build a stateful workflow on top of the current library is very hard, and I'm not sure there is a way to reconcile this.

automatically_reconnect is a way to try to solve this, by using a blocking function that expects a callback, and calls it with the client whenever a connection failure happened and it was able to reconnect. The intended use cases is the user knowing when a re-connection occured, by the callback being called again, and having tools for dealing with it in his own way; for instance, he may choose to track the current open channels and re-connect as if nothing happend, if the old behavior was okay for him.

A callback based design is still possible

The idea with offering simpler building blocks is that bulding callback based abstractions is still possible; in fact it should be a lot easier now, as it should only worry about calling the callbacks and dealing with the acks, all the messaging parsing and sending logic should be dealt with already, and the user may treat each channel as it's own connection entirely.

In fact, the callback based design is so straightforward to write that I'm unsure if we should offer it as one of the building blocks of the library, or let the user write the one he wants. It should be as simple as:

class ChannelCallbackManager:
    async def run(self):
        async for message in self.channel.messages():
            if isinstance(message, BroadcastMessage):
                for callback in self.broadcast_callback:
                    await callback(message)
            ...

    def on_broadcast(self, callback: ...):
        self.broadcast_callbacks.append(callback)

[grdsdev]

subscribe() never marks the channel joined → set_auth() is a no-op, subscribe() sends the join, checks the reply, returns — but never sets self.joined = True or self.state = ChannelStates.JOINED. They stay at the __init__ defaults (False / CLOSED).

Consequences:

• client.set_auth() (client.py) filters on channel.joined and channel.state == ChannelStates.JOINED → always False → token refresh is never pushed to any channel.

• _can_push() (uses self.joined) is always False.
  Fix: set self.joined = True / self.state = JOINED on a successful subscribe reply (and reset on unsubscribe/error).

[grdsdev]

KeyError if key is absent, should catch this?

[grdsdev]

No delay here?

[grdsdev]

self._ws_connection is never cleaned, so this is always True.

[grdsdev]

self.join_ref only exists after subscribe(), consider assigning None to it on __init__.

[grdsdev]

should this be added to dev dependencies instead?

[grdsdev]

if this call raises then del self.message_refs[message.ref] is not called.

[grdsdev]

Suggested change

	hb_interval: int = DEFAULT_HEARTBEAT_INTERVAL,
	heartbeat_interval: int = DEFAULT_HEARTBEAT_INTERVAL,

[grdsdev]

we had discussed about naming this connect, I'm not used to use Python, but it seems calling connect is the right thing, considering the websocket also names it connect.

[grdsdev]

what is this replay? it is not used.