Async Firehose Client: block on make message handler call, add on error callback

[DXsmiley]

The original _AsyncWebsocketClient._process_message_frame spawns a new task and then returns immediately, allowing the client to enter a very fast client.recv() -> create_task loop.

    def _process_message_frame(self, frame: 'MessageFrame') -> None:
        task: asyncio.Task = self._loop.create_task(self._on_message_callback(frame))
        self._on_message_tasks.add(task)
        task.add_done_callback(self._on_message_callback_done)

This isn't a problem when consuming the head of the firehose in real-time, however when replaying past events, if message processing is at all non-trivial, it causes tasks to be spawned far faster than they can be completed. A glut of in-flight tasks creates a huge amount of scheduling overhead, and a high amount of contention on any shared resources such as databases, or internal locks. There's no way to slow down the frequency of client.recv() so you end up with a memory-leak-esque situation.

I think the original behaviour is unintuitive and quite an unexpected pit-fall, so I've replaced it with a normal (async) callback. The client waits for the callback to finish before receiving the next message. This means there's no "concurrency" out of the box, but provides a much more predictable interface for people to build their own systems on top of; I think it's valuable that the websocket client is as unopinionated about the message processing side of things as possible.

For example, the callback I'm currently using in my own code is simply this:

messages_to_process: 'asyncio.Queue[MessageFrame]' = asyncio.Queue(maxsize=20)

async def on_message_handler(message: 'MessageFrame') -> None:
    await messages_to_process.put(message)

messages_to_process.put blocks while the queue is full, which means that client.recv() is not called faster we can handle, and I'm able to easily configure whatever concurrency I want on the queue-consumer side. This still allows for message receiving and processing to occur simultaneously, but prevents the former from out-pacing the latter.

I've also made some changes to _WebsocketClient and _WebsocketClientBase to account for the differences between _WebsocketClient and _AsyncWebsocketClient.

[MarshalX]

Thank you! I'll take a look as soon as done with my main job

[MarshalX]

pls make sure that the difference between async and sync version _process_message_frame method only in "await" statements. for now, it is not. and probably we don't need _print_exception anymore? because now we are in the catch block

[MarshalX]

could we move this part to the base class?

    def _process_raw_frame(self, data: bytes) -> None:
        frame = Frame.from_bytes(data)
        if isinstance(frame, ErrorFrame):
            raise FirehoseError(XrpcError(frame.body.error, frame.body.message))
        if isinstance(frame, MessageFrame):
            self._process_message_frame(frame)
        else:
            raise FirehoseDecodingError('Unknown frame type')

idk smth like this:

    def _pre_process_raw_frame(self, data: bytes) -> Frame:   # in base
        frame = Frame.from_bytes(data)
        if isinstance(frame, ErrorFrame):
            raise FirehoseError(XrpcError(frame.body.error, frame.body.message))
        if not isinstance(frame, MessageFrame):
            raise FirehoseDecodingError('Unknown frame type')

        return frame
    
    def _process_raw_frame(self, data: bytes) -> None:  # in sync
        frame = self._pre_process_raw_frame(data)
        self._process_message_frame(frame)

    async def _process_raw_frame(self, data: bytes) -> None:  # in async
        frame = self._pre_process_raw_frame(data)
        await self._process_message_frame(frame)

[MarshalX]

btw dotn we want to have async version of _on_callback_error_callback in async client?

[MarshalX]

i like the use of wait_for :)

[DXsmiley]

I've allowed the stop event to interrupt the client.recv but it's uh... the code's not great I'm gonna be honest.

[MarshalX]

uh..

getting towards scope-creep for this PR.

let's not include wait_for tricks in this pr at all. as i understood that will not be ported to sync client. let's merge callback improvements

[DXsmiley]

Alright, I've rolled back a few changes and cleaned things up a bit.

[MarshalX]

@DXsmiley thank you so much for your contribution! and sorry for a long time