src/genro_asgi/channel/client.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Channel client — the child side of the parent↔child channel.16 17 Knowing how to BE a child is part of what a server IS (SPECIFICATION.md18 ◆D10): the minimal package ships the frame protocol and this client; the hub19 (parent side) lives in the orchestration package and imports the protocol20 from below, never the reverse.21 22 ``connect()`` retries with short backoff until ``connect_timeout`` (boot23 race: the hub socket may not be bound yet) and presents the child with a24 REGISTER frame (``data={"name", "pid"}``). Steady state is fire-and-forget25 frames in both directions. There is no steady-state reconnection: when the26 hub side goes away (EOF — the connection-loss signal),27 ``on_orphan(client)`` fires and the child is expected to terminate cleanly.28 A deliberate ``close()`` fires no orphan signal.29 30 Callbacks (``on_message(frame)``, ``on_orphan(client)``) may be sync or31 async; an exception raised by a callback is logged and never severs the32 channel — a consumer bug must not fake a member death.33 34 Addresses::35 36 uds:/path/to/hub.sock Unix domain socket (default)37 tcp:127.0.0.1:8731 TCP (multi-host door)38 """39 40 from __future__ import annotations41 42 import asyncio43 import inspect44 import logging45 import os46 from typing import Any, Callable47 48 from .control import ControlPayload49 from .frame import REGISTER_METHOD, REGISTER_PATH, Frame, FrameStream50 51 __all__ = ["ChannelClient"]52 53 54 class ChannelClient:55 """Child-side endpoint: connect to the hub, present itself, relay frames."""56 57 def __init__(58 self,59 address: str,60 name: str,61 *,62 on_message: Callable[..., Any] | None = None,63 on_orphan: Callable[..., Any] | None = None,64 connect_timeout: float = 10.0,65 max_size: int | None = None,66 ) -> None:67 self.address = address68 self.name = name69 self.on_message = on_message70 self.on_orphan = on_orphan71 self.connect_timeout = connect_timeout72 self.max_size = max_size73 self.control_payload = ControlPayload()74 transport, _, rest = address.partition(":")75 self._uds_path: str | None = None76 self._tcp: tuple[str, int] | None = None77 if transport == "uds" and rest:78 self._uds_path = rest79 elif transport == "tcp" and rest:80 host, _, port = rest.rpartition(":")81 if not host or not port.isdigit():82 raise ValueError(f"invalid tcp address: {address!r}")83 self._tcp = (host, int(port))84 else:85 raise ValueError(86 f"invalid channel address: {address!r} (uds:<path> | tcp:<host>:<port>)"87 )88 self._logger = logging.getLogger(__name__)89 self._stream: FrameStream | None = None90 self._receive_task: asyncio.Task[None] | None = None91 self._connected = False92 self._closing = False93 self._closed_event = asyncio.Event()94 95 @property96 def connected(self) -> bool:97 """Whether the channel is up (REGISTER sent, receive loop running)."""98 return self._connected99 100 @property101 def closed(self) -> bool:102 """Whether the channel ended (either side; ``False`` before connect)."""103 return self._closed_event.is_set()104 105 async def connect(self) -> None:106 """Connect with boot-time retry/backoff, present the REGISTER frame."""107 if self.connected:108 raise RuntimeError("channel client is already connected")109 loop = asyncio.get_running_loop()110 deadline = loop.time() + self.connect_timeout111 interval = 0.05112 while True:113 try:114 reader, writer = await self._open_connection()115 break116 except OSError:117 if loop.time() + interval >= deadline:118 raise ConnectionError(119 f"hub not reachable at {self.address} within {self.connect_timeout}s"120 ) from None121 await asyncio.sleep(interval)122 interval = min(interval * 2, 0.5)123 self._stream = FrameStream(124 reader, writer, max_size=self.max_size125 )126 register = Frame(127 method=REGISTER_METHOD,128 path=REGISTER_PATH,129 payload=self.control_payload.encode({"name": self.name, "pid": os.getpid()}),130 )131 await self._stream.write(register)132 self._connected = True133 self._closed_event.clear()134 self._receive_task = asyncio.create_task(self._receive_loop(self._stream))135 self._logger.info("Connected to hub at %s as %s", self.address, self.name)136 137 async def close(self) -> None:138 """Deliberate local close: no orphan signal."""139 self._closing = True140 if self._receive_task is not None and not self._receive_task.done():141 self._receive_task.cancel()142 try:143 await self._receive_task144 except asyncio.CancelledError:145 pass146 if self._stream is not None:147 await self._stream.close()148 self._stream = None149 self._connected = False150 self._closed_event.set()151 152 async def wait_closed(self) -> None:153 """Block until the channel ends (either side); the child's main wait."""154 await self._closed_event.wait()155 156 async def send(self, *, method: str = "POST", path: str = "/", data: Any = None) -> str:157 """Send one frame to the hub (fire-and-forget); returns the frame id.158 159 A connection dropping mid-send is a dying hub: the frame is lost by160 design and the orphan signal follows on the receive side.161 """162 if self._stream is None or not self.connected:163 raise ConnectionError("not connected")164 frame = Frame(method=method, path=path, payload=self.control_payload.encode(data))165 return await self.send_frame(frame)166 167 async def send_frame(self, frame: Frame) -> str:168 """Send an already encoded frame without opening its payload."""169 if self._stream is None or not self.connected:170 raise ConnectionError("not connected")171 await self._stream.write(frame)172 return frame.id173 174 async def _open_connection(self) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:175 """Open the transport for the parsed address (uds or tcp)."""176 if self._uds_path is not None:177 return await asyncio.open_unix_connection(self._uds_path)178 host, port = self._tcp179 return await asyncio.open_connection(host, port)180 181 async def _receive_loop(self, stream: FrameStream) -> None:182 """Read frames until the channel ends; hub gone → orphan.183 184 A protocol violation from the hub (oversized or non-wsx frame) is a185 clean death: logged, the loop breaks and the orphan path follows —186 the exception never leaves the task. The ``finally`` closes the187 stream so the writer never outlives the loop.188 """189 try:190 while True:191 try:192 frame = await stream.read()193 except ValueError:194 self._logger.exception("Protocol violation from the hub; closing the channel")195 break196 if frame is None:197 break198 await self._fire(self.on_message, frame)199 except asyncio.CancelledError:200 return201 finally:202 if self._stream is stream:203 self._connected = False204 self._stream = None205 await stream.close()206 self._closed_event.set()207 if not self._closing:208 self._logger.info("Hub connection lost: %s is orphan", self.name)209 await self._fire(self.on_orphan, self)210 211 async def _fire(self, callback: Callable[..., Any] | None, *args: Any) -> None:212 """Run a sync-or-async callback; a consumer bug must not sever the channel."""213 if callback is None:214 return215 try:216 result = callback(*args)217 if inspect.isawaitable(result):218 await result219 except Exception:220 self._logger.exception("Channel callback %r failed", callback)