Rewrite WebSocketWriter to use StreamWriter

[asvetlov]

Now websocket writer has custom flow control.
It's maybe not bad but overcomplicates things.
Since #2836 is done websocket writer uses coroutines everywhere, the transition should be smooth.

No Public API changes but limit=DEFAUTLT_LIMIT and MSG_SIZE will be dropped.
I see related micro-optimizations made by @fafhrd91 and @pfreixes but I believe the performance impact will be negligible.

Talking about performance using bytearray as websocket message structure and adding Cython optimization for parsing/building websocket header can get better speed boost.

[ljluestc]

import asyncio
import struct
import aiohttp
from aiohttp import web

class WebSocketWriter:
    def __init__(self, ws):
        self.ws = ws
        self.buffer = bytearray()  # Using bytearray for message structure

    async def send_message(self, data):
        """Send a message over WebSocket, ensuring coroutine-based flow control."""
        header = self.build_websocket_header(len(data))
        self.buffer.extend(header + data)  # Using bytearray for faster appends

        # Write message and flush
        await self.ws.send_bytes(self.buffer)
        self.buffer.clear()  # Clear buffer after sending to reduce memory footprint

    def build_websocket_header(self, message_length):
        """Build WebSocket header for a given message length."""
        if message_length < 126:
            return struct.pack("!BB", 0x81, message_length)
        elif message_length < (1 << 16):
            return struct.pack("!BBH", 0x81, 126, message_length)
        else:
            return struct.pack("!BBQ", 0x81, 127, message_length)

async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    writer = WebSocketWriter(ws)

    try:
        while True:
            msg = await ws.receive()
            if msg.type == aiohttp.WSMsgType.TEXT:
                print(f"Received: {msg.data}")
                await writer.send_message(msg.data.encode('utf-8'))  # Echo message back
            elif msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.ERROR):
                break
    except Exception as e:
        print(f"WebSocket error: {e}")
    finally:
        await ws.close()
        print("WebSocket closed")

app = web.Application()
app.router.add_get('/ws', websocket_handler)

if __name__ == '__main__':
    web.run_app(app, port=8080)

[Dreamsorcerer]

@ljluestc If you think that's an improvement, please submit a PR.

@bdraco Is this task basically done now with your most recent changes?

[bdraco]

I think everything that would still make sense to do with the current codebase is done. There isn't anything to be gained by making WebSocketWriter and AbstractStreamWriter since AbstractStreamWriter knows about compression, chunking, and writing headers which the WebSocketWriter will never do.

We could probably simplify the writer some more by removing ping and pong as it seems like they don't really belong there as they are simple wrappers around send_frame, and it would be nicer to keep all the higher level details out of WebSocketWriter edit: cleanup PR #9566

[bdraco]

Also limit is used downstream so we don't want to drop that. Right now it has to be set via a protected method. Let me also look into exposing that to close up that downstream