How to override the ping method in connection class to send a custome ping frame

[river7816]

I am using websockets==13.1 and have reviewed the documentation. The websockets library includes a keep-alive feature that sends an empty ByteFrame as a ping message. How can I modify the following code to send a custom message like {"op": "ping"} instead? I attempted to override the Connection class, but it didn’t work for me.

class CustomConnection(Connection):
    async def ping(self, data: bytes | None = None) -> Awaitable[float]:
        ping_message = json.dumps({"custome": "ping"})
        self.send(ping_message)
        return await super().ping(data)

async for websocket in websockets.connect(
            uri=self._base_url,
            ping_interval=self._ping_interval,
            ping_timeout=self._ping_timeout,
            close_timeout=self._close_timeout,
            max_queue=self._max_queue,
            create_connection=CustomConnection,
        ):
            try:
                payload = json.dumps(payload)
                await websocket.send(payload)
                async for msg in websocket:
                    msg = orjson.loads(msg)
                    print(msg)
            except websockets.ConnectionClosed:
                self._log.error("Connection closed, reconnecting...")

[aaugustin]

You're referring to an application-level keepalive, not protocol-level, and websockets doesn't provide that.

This question comes up regularly; I'll explain in the documentation.

[river7816]

You're referring to an application-level keepalive, not protocol-level, and websockets doesn't provide that.

This question comes up regularly; I'll explain in the documentation.

I'm not sure if I need to create a coroutine, and then implement a while True in that coroutine to send this custom ping using the send method. Is it right?

[aaugustin]

That's correct.

[river7816]

That's correct.

Is there a more elegant way to achieve this? I'm also unsure whether to use await websocket.send(ping_message) or await websocket.ping(ping_message) in the ping method.

async def ping(self, websocket: websockets.asyncio.client.ClientConnection):
    ping_message = json.dumps({"custome": "ping"})
    await websocket.send(ping_message)

async for websocket in websockets.connect(
            uri=self._base_url,
            ping_interval=self._ping_interval,
            ping_timeout=self._ping_timeout,
            close_timeout=self._close_timeout,
            max_queue=self._max_queue,
            create_connection=CustomConnection,
        ):
            try:
                payload = json.dumps(payload)
                await websocket.send(payload)

                asyncio.create_task(ping(websocket))

                async for msg in websocket:
                    msg = orjson.loads(msg)
                    print(msg)
            except websockets.ConnectionClosed:
                self._log.error("Connection closed, reconnecting...")

[aaugustin]

Which API are you connecting to?