src/genro_asgi/channel/local.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 """Local channel — the in-process wire, byte-identical to the socket one.16 17 The single role (design §3.5a) runs commander and worker in ONE process, and18 it must speak the very same protocol as a spawned child: not "the same API",19 the same *bytes*. ``LocalChannel`` is therefore two ``asyncio.Queue``s of20 encoded frames — every envelope crosses through ``Frame.encode()`` and is21 re-parsed on the other side with the same versioned info/bytes22 rules ``FrameStream.read`` applies. A payload dict mutated after ``send()``23 cannot reach the peer, exactly as over a socket.24 25 ``LocalFrameStream`` is the codec twin of ``FrameStream`` (the only module26 above the frame protocol allowed to touch bytes): ``read()`` returns ``None``27 at EOF, an oversized or malformed frame raises ``ValueError``. A queue sentinel28 models EOF in both directions, so closing either end has the socket meaning —29 the peer's read ends and the link-loss callback runs.30 31 ``LocalChannel`` itself IS the member face, with the ``ChannelClient`` API32 (``connect``/``send``/``close``/``wait_closed``, ``on_message``/``on_orphan``,33 ``connected``), plus ``send_frame(frame)`` for the frames whose id is not the34 sender's to mint — a REPLY reuses the CALL's id. The hub side is consumed by35 ``ChannelHub.attach_local()``, which registers it through the same REGISTER36 path as any socket member: one rubric, no parallel bookkeeping.37 """38 39 from __future__ import annotations40 41 import asyncio42 import inspect43 import logging44 import os45 from typing import Any, Callable46 47 from .frame import (48 REGISTER_METHOD,49 REGISTER_PATH,50 Frame,51 FrameCodec,52 )53 from .control import ControlPayload54 55 __all__ = ["LocalChannel", "LocalFrameStream"]56 57 58 class LocalFrameStream:59 """Frame codec over a pair of byte queues — the in-process ``FrameStream``.60 61 Reads from ``inbound``, writes to ``outbound``; a ``None`` in a queue is62 the EOF sentinel. ``close()`` sends it to the peer and unparks its own63 reader, so both sides observe the end of the channel.64 """65 66 def __init__(67 self,68 inbound: asyncio.Queue[bytes | None],69 outbound: asyncio.Queue[bytes | None],70 *,71 max_size: int | None = None,72 max_queue_size: int = 16,73 ) -> None:74 if max_queue_size < 1:75 raise ValueError("max_queue_size must be positive")76 self.inbound = inbound77 self.outbound = outbound78 self.max_queue_size = max_queue_size79 self.codec = FrameCodec(max_size=max_size)80 self.max_size = self.codec.max_size81 self._closed = False82 83 @property84 def closed(self) -> bool:85 """Whether this end has been closed."""86 return self._closed87 88 async def read(self) -> Frame | None:89 """The next frame, or ``None`` when the channel ended."""90 wire = await self.inbound.get()91 if wire is None:92 return None93 return self.codec.get_frame(wire)94 95 async def write(self, frame: Frame) -> None:96 """Encode and enqueue one frame; a full or closed end raises."""97 if self.outbound.qsize() >= self.max_queue_size:98 raise ConnectionError("local channel queue full; frame not sent")99 wire = self.codec.encode(frame)100 if self._closed:101 raise BrokenPipeError("local channel end is closed")102 await self.outbound.put(wire)103 104 async def close(self) -> None:105 """Close this end: EOF to the peer, EOF to our own parked reader."""106 if self._closed:107 return108 self._closed = True109 await self.outbound.put(None)110 await self.inbound.put(None)111 112 113 class LocalChannel:114 """In-process channel endpoint: the member face of a queue-backed wire.115 116 Built by whoever owns the in-process worker, then handed to117 ``ChannelHub.attach_local()``; ``connect()`` presents the REGISTER frame118 just like ``ChannelClient`` does, and the queues buffer it whichever side119 goes first.120 """121 122 def __init__(123 self,124 name: str,125 *,126 on_message: Callable[..., Any] | None = None,127 on_orphan: Callable[..., Any] | None = None,128 max_size: int | None = None,129 max_queue_size: int = 16,130 ) -> None:131 if max_queue_size < 1:132 raise ValueError("max_queue_size must be positive")133 self.name = name134 self.on_message = on_message135 self.on_orphan = on_orphan136 self.max_size = max_size137 self.address = "local:"138 to_hub: asyncio.Queue[bytes | None] = asyncio.Queue()139 to_member: asyncio.Queue[bytes | None] = asyncio.Queue()140 self._member_stream = LocalFrameStream(141 to_member,142 to_hub,143 max_size=max_size,144 max_queue_size=max_queue_size,145 )146 self._hub_stream = LocalFrameStream(147 to_hub,148 to_member,149 max_size=max_size,150 max_queue_size=max_queue_size,151 )152 self._logger = logging.getLogger(__name__)153 self._receive_task: asyncio.Task[None] | None = None154 self._connected = False155 self._closing = False156 self._closed_event = asyncio.Event()157 158 @property159 def hub_stream(self) -> LocalFrameStream:160 """The hub-side end, consumed by ``ChannelHub.attach_local()``."""161 return self._hub_stream162 163 @property164 def connected(self) -> bool:165 """Whether the channel is up (REGISTER sent, receive loop running)."""166 return self._connected167 168 @property169 def closed(self) -> bool:170 """Whether the channel ended (either side; ``False`` before connect)."""171 return self._closed_event.is_set()172 173 async def connect(self) -> None:174 """Present the REGISTER frame and start the receive loop."""175 if self.connected:176 raise RuntimeError("local channel is already connected")177 register = Frame(178 method=REGISTER_METHOD,179 path=REGISTER_PATH,180 payload=ControlPayload().encode({"name": self.name, "pid": os.getpid()}),181 )182 await self._member_stream.write(register)183 self._connected = True184 self._closed_event.clear()185 self._receive_task = asyncio.create_task(self._receive_loop())186 self._logger.info("Local channel connected as %s", self.name)187 188 async def close(self) -> None:189 """Deliberate local close: no orphan signal."""190 self._closing = True191 if self._receive_task is not None and not self._receive_task.done():192 self._receive_task.cancel()193 try:194 await self._receive_task195 except asyncio.CancelledError:196 pass197 await self._member_stream.close()198 self._connected = False199 self._closed_event.set()200 201 async def wait_closed(self) -> None:202 """Block until the channel ends (either side); the member's main wait."""203 await self._closed_event.wait()204 205 async def send(self, *, method: str = "POST", path: str = "/", data: Any = None) -> str:206 """Send one frame to the hub (fire-and-forget); returns the frame id."""207 return await self.send_frame(208 Frame(method=method, path=path, payload=ControlPayload().encode(data))209 )210 211 async def send_frame(self, frame: Frame) -> str:212 """Send an already-built frame — a REPLY reuses the CALL's id."""213 if not self.connected:214 raise ConnectionError("not connected")215 await self._member_stream.write(frame)216 return frame.id217 218 async def _receive_loop(self) -> None:219 """Read frames until the channel ends; hub gone → orphan."""220 try:221 while True:222 try:223 frame = await self._member_stream.read()224 except ValueError:225 self._logger.exception("Protocol violation from the hub; closing the channel")226 break227 if frame is None:228 break229 await self._fire(self.on_message, frame)230 except asyncio.CancelledError:231 return232 finally:233 self._connected = False234 await self._member_stream.close()235 self._closed_event.set()236 if not self._closing:237 self._logger.info("Hub side gone: %s is orphan", self.name)238 await self._fire(self.on_orphan, self)239 240 async def _fire(self, callback: Callable[..., Any] | None, *args: Any) -> None:241 """Run a sync-or-async callback; a consumer bug must not sever the channel."""242 if callback is None:243 return244 try:245 result = callback(*args)246 if inspect.isawaitable(result):247 await result248 except Exception:249 self._logger.exception("Channel callback %r failed", callback)