|
| 1 | +""" |
| 2 | +Cyphal/Serial-over-TCP broker. |
| 3 | +
|
| 4 | +Cyphal/Serial uses COBS-encoded frames with a zero byte as frame delimiter. When |
| 5 | +brokering a byte-stream ncat --broker does know about the frame delimiter and |
| 6 | +might interleave frames from different clients. |
| 7 | +This broker is similar in functionality to :code:`ncat --broker`, but reads the |
| 8 | +whole frame before passing it on to other clients, avoiding interleaved frames |
| 9 | +and potential frame/data loss. |
| 10 | +""" |
| 11 | + |
| 12 | +import argparse |
| 13 | +import asyncio |
| 14 | +import logging |
| 15 | +import socket |
| 16 | +import typing as t |
| 17 | + |
| 18 | + |
| 19 | +class Client: |
| 20 | + """ |
| 21 | + Represents a client connected to the broker, wrapping StreamReader and |
| 22 | + StreamWriter to conveniently read/write zero-terminated frames. |
| 23 | + """ |
| 24 | + |
| 25 | + def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: |
| 26 | + self._buffer = bytearray() |
| 27 | + self._reader = reader |
| 28 | + self._writer = writer |
| 29 | + |
| 30 | + async def __aenter__(self) -> "Client": |
| 31 | + return self |
| 32 | + |
| 33 | + async def __aexit__(self, *_: t.Any) -> bool: |
| 34 | + self._writer.close() |
| 35 | + await self._writer.wait_closed() |
| 36 | + return True |
| 37 | + |
| 38 | + async def read(self) -> t.AsyncGenerator[bytes, None]: |
| 39 | + """ |
| 40 | + async generator yielding complete frames, including terminating \x00. |
| 41 | + """ |
| 42 | + buffer = bytearray() |
| 43 | + while not self._reader.at_eof(): |
| 44 | + buffer += await self._reader.readuntil(separator=b"\x00") |
| 45 | + # don't pass on a leading zero-byte on its own. |
| 46 | + if len(buffer) == 1: |
| 47 | + continue |
| 48 | + yield buffer |
| 49 | + buffer = bytearray() |
| 50 | + |
| 51 | + def write(self, frame: bytes) -> None: |
| 52 | + """ |
| 53 | + Writes a frame to the stream, unless the stream is closing. |
| 54 | +
|
| 55 | + :param frame: Frame to send to this client. |
| 56 | + """ |
| 57 | + if self._writer.is_closing(): |
| 58 | + return |
| 59 | + self._writer.write(frame) |
| 60 | + |
| 61 | + async def drain(self) -> None: |
| 62 | + """ |
| 63 | + Flushes the stream. |
| 64 | + """ |
| 65 | + if self._writer.is_closing(): |
| 66 | + return |
| 67 | + await self._writer.drain() |
| 68 | + |
| 69 | + |
| 70 | +async def serve_forever(host: str, port: int) -> None: |
| 71 | + """ |
| 72 | + pybroker core server loop. |
| 73 | +
|
| 74 | + Accept clients on :code:`host`::code:`port` and broadcast any frame |
| 75 | + received from any client to all other clients. |
| 76 | +
|
| 77 | + :param host: IP, where the broker will be reachable on. |
| 78 | + :param port: port, on which the broker will listen on. |
| 79 | + """ |
| 80 | + clients: list[Client] = [] |
| 81 | + list_lock = asyncio.Lock() |
| 82 | + |
| 83 | + async def _run_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: |
| 84 | + async with Client(reader, writer) as client: |
| 85 | + async with list_lock: |
| 86 | + logging.info("Client connected.") |
| 87 | + clients.append(client) |
| 88 | + try: |
| 89 | + async for frame in client.read(): |
| 90 | + logging.debug("Received frame %s", frame) |
| 91 | + for c in clients: |
| 92 | + if c != client: |
| 93 | + c.write(frame) |
| 94 | + async with list_lock: |
| 95 | + # not sure if flushing is required. |
| 96 | + for c in clients: |
| 97 | + await c.drain() |
| 98 | + |
| 99 | + finally: |
| 100 | + async with list_lock: |
| 101 | + clients.remove(client) |
| 102 | + logging.info("Client disconnected.") |
| 103 | + |
| 104 | + logging.info("Broker started on %s:%s", host, port) |
| 105 | + reuse_port = hasattr(socket, "SO_REUSEPORT") and socket.SO_REUSEPORT |
| 106 | + await asyncio.start_server( |
| 107 | + _run_client, |
| 108 | + host, |
| 109 | + port, |
| 110 | + family=socket.AF_INET, |
| 111 | + reuse_address=True, |
| 112 | + reuse_port=reuse_port, |
| 113 | + ) |
| 114 | + |
| 115 | + |
| 116 | +def main() -> None: |
| 117 | + """ |
| 118 | + TCP-broker which forwards complete, zero-terminated frames/datagrams among |
| 119 | + all connected clients. |
| 120 | + """ |
| 121 | + |
| 122 | + parser = argparse.ArgumentParser() |
| 123 | + parser.add_argument("-i", "--host", default="127.0.0.1", help="Interface to listen on for incoming connections.") |
| 124 | + parser.add_argument("-p", "--port", default=50905, help="Clients connect to this port.") |
| 125 | + parser.add_argument("--verbose", default=False, action="store_true", help="Increase logging verbosity.") |
| 126 | + |
| 127 | + args = parser.parse_args() |
| 128 | + |
| 129 | + logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO) |
| 130 | + |
| 131 | + try: |
| 132 | + loop = asyncio.get_running_loop() |
| 133 | + except RuntimeError: |
| 134 | + loop = asyncio.new_event_loop() |
| 135 | + asyncio.set_event_loop(loop) |
| 136 | + loop.run_until_complete(serve_forever(args.host, args.port)) |
| 137 | + loop.run_forever() |
0 commit comments