diff --git a/src/phx_channel/channel.py b/src/phx_channel/channel.py index 8fa5bac..e00f36c 100644 --- a/src/phx_channel/channel.py +++ b/src/phx_channel/channel.py @@ -37,6 +37,14 @@ def __init__(self, socket: Socket, topic: str, params: dict[str, Any]): self._event_handlers: dict[str, list[Callable[..., Any]]] = {} self._pending_replies: dict[str, asyncio.Future[dict[str, Any]]] = {} self._push_buffer: list[tuple[str, Any, asyncio.Future[dict[str, Any]]]] = [] + # Inbound pushes that arrived before any handler was registered. + # Server scenarios can `autoPush` in response to a join frame, and + # that push can land on the asyncio queue before the caller has + # had a chance to register a handler post-join. We replay these + # to the first matching handler registered within the window. + # Bounded so a forgotten handler doesn't leak unbounded memory. + self._pending_pushes: dict[str, list[Any]] = {} + self._pending_pushes_cap: int = 32 @property def topic(self) -> str: @@ -195,11 +203,22 @@ def on(self, event: str, callback: Callable[..., Any]) -> Callable[[], None]: """ Register a callback for a channel event. - Returns an unsubscribe function. + Returns an unsubscribe function. If pushes for this event already + arrived before any handler was registered (e.g. server-side + autoPush fired on join), they're replayed to this callback in + arrival order so the caller can't miss them due to scheduling. """ handlers = self._event_handlers.setdefault(event, []) handlers.append(callback) + pending = self._pending_pushes.pop(event, None) + if pending: + for payload in pending: + try: + callback(payload) + except Exception: + logger.exception("Error in handler for %s:%s (replayed)", self._topic, event) + def unsubscribe() -> None: handlers.remove(callback) @@ -225,13 +244,22 @@ def _on_message( elif event == "phx_error": self._handle_error(payload) else: - # User event — dispatch to handlers - handlers = self._event_handlers.get(event, []) - for handler in handlers: - try: - handler(payload) - except Exception: - logger.exception("Error in handler for %s:%s", self._topic, event) + # User event — dispatch to handlers, or buffer if none yet. + handlers = self._event_handlers.get(event) + if handlers: + for handler in handlers: + try: + handler(payload) + except Exception: + logger.exception("Error in handler for %s:%s", self._topic, event) + else: + buf = self._pending_pushes.setdefault(event, []) + if len(buf) < self._pending_pushes_cap: + buf.append(payload) + else: + # Drop the oldest to bound memory for orphan events. + buf.pop(0) + buf.append(payload) def _handle_reply(self, ref: str | None, payload: Any) -> None: if ref and ref in self._pending_replies: diff --git a/src/phx_channel/tests/test_unit.py b/src/phx_channel/tests/test_unit.py index ee80a08..a1d7da8 100644 --- a/src/phx_channel/tests/test_unit.py +++ b/src/phx_channel/tests/test_unit.py @@ -287,6 +287,35 @@ async def test_null_join_ref_accepted(): assert received == [{"data": "broadcast"}] +async def test_pushes_before_handler_are_replayed_on_registration(): + """ + Server `autoPush` can fire in response to a join frame and land on the + asyncio queue before the caller has registered a handler. Verify the + Channel buffers those pushes and replays them when on() is called. + """ + _, ch = await _joined_channel() + # Two pushes arrive with no handlers yet — both should be buffered. + ch._on_message(None, None, "new_entry", {"id": "a"}) + ch._on_message(None, None, "new_entry", {"id": "b"}) + received: list[dict] = [] + ch.on("new_entry", lambda p: received.append(p)) + assert received == [{"id": "a"}, {"id": "b"}] + # Subsequent pushes go straight through. + ch._on_message(None, None, "new_entry", {"id": "c"}) + assert received == [{"id": "a"}, {"id": "b"}, {"id": "c"}] + + +async def test_pending_push_buffer_is_bounded(): + _, ch = await _joined_channel() + ch._pending_pushes_cap = 3 + for i in range(5): + ch._on_message(None, None, "evt", {"i": i}) + received: list[dict] = [] + ch.on("evt", lambda p: received.append(p)) + # Oldest dropped once cap hit; newest kept in order. + assert received == [{"i": 2}, {"i": 3}, {"i": 4}] + + async def test_handler_error_does_not_crash_dispatch(): _, ch = await _joined_channel() received = []