()
| 6 | from websockets.asyncio.client import connect |
| 7 | |
| 8 | async def client(): |
| 9 | async with connect("ws://localhost:8765") as websocket: |
| 10 | # Close the connection when receiving SIGTERM. |
| 11 | loop = asyncio.get_running_loop() |
| 12 | def close(): |
| 13 | return loop.create_task(websocket.close()) |
| 14 | loop.add_signal_handler(signal.SIGTERM, close) |
| 15 | |
| 16 | # Process messages received on the connection. |
| 17 | async for message in websocket: |
| 18 | ... |
| 19 | |
| 20 | asyncio.run(client()) |
no test coverage detected
searching dependent graphs…