From d60768b3812f30cf4109e2a5dab498d916262572 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Thu, 24 Sep 2026 12:43:47 +0200 Subject: refactor(node): dispatch MNP messages through a table The pre-authentication guards stay explicit code, in the same order and text. After the handshake, a table maps each type to its handler and to whether it runs as a task, the choice each branch made; the three inline blocks become StreamingMixin methods, unchanged. The dispatch golden is identical, including types that are not strings. Co-Authored-By: Claude Opus 5.5 --- .../transport/webrtc/apps/streaming.py | 58 ++++++++++++++++++++++ 1 file changed, 58 insertions(+) (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc/apps') diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py index 05ab9e7..a425c65 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py @@ -641,3 +641,61 @@ class StreamingMixin: reason, index, time.monotonic() - self._stream_started_at) log.info("Streamed %s: %d segments", entry.name, index) self._audit("stream_video", entry.name) + + def _do_stream_request(self, msg: dict) -> None: + sem = self._ctx.get("_transcode_sem") + log.info("stream: req file=%s credits=%s slots_free=%s prev=%s", + str(msg.get("file_id"))[:12], msg.get("credits"), + getattr(sem, "_value", "?"), + "alive" if (self._stream_task and + not self._stream_task.done()) else "none") + self._spawn(self._replace_stream(msg)) + + def _do_client_diag(self, msg: dict) -> None: + # Diagnostics only. The node acts on none of it — it writes it + # next to its own view of the same stream, which is the only + # place the two halves can be compared when the client is a + # phone with no console. + # Every field is peer-controlled, so each is stringified and + # cut short: this is a log line, not a channel for writing + # whatever one likes into the operator's file. + def _f(key: str, n: int = 24) -> str: + return str(msg.get(key))[:n].replace("\n", " ") + if msg.get("event"): + # Once per stream or per seek, not once per five seconds — + # and a seek nobody asked for looks exactly like a viewer + # dragging the scrubber from this side, so it has to be + # visible without turning DEBUG on. + log.info( + "stream: client %s target=%s t=%ss offset=%s ready=%s " + "duration=%s ranges=[%s]", + _f("event", 16), _f("target"), _f("t"), _f("offset"), + _f("ready"), _f("duration"), _f("ranges", 120)) + # Debug: one line every five seconds per viewer. Run the daemon + # with --log-level debug to see inside a player that is + # misbehaving — it is the only view of the browser there is + # when the browser is a phone. + else: + # `ahead` on its own cannot say whether a short buffer is + # the player's own gate holding or the network failing to + # keep up, and those two want opposite answers. `limit` is + # what the gate is set to for this film and `budget` the + # byte budget it was derived from, so the three read as one + # sentence. + log.debug( + "stream: client t=%ss ahead=%ss/%ss budget=%sMB " + "ready=%s paused=%s " + "stalled=%s q=%s inflight=%s appending=%s updating=%s " + "quota=%s ms=%s err=%s ranges=[%s] (sent=%d)", + _f("t"), _f("ahead"), _f("limit"), _f("budgetMB"), + _f("ready"), _f("paused"), + _f("stalled"), _f("q"), _f("inflight"), _f("appending"), + _f("updating"), _f("quota"), _f("ms"), _f("err", 80), + _f("ranges", 120), self._stream_segments) + + def _do_stream_stop(self) -> None: + age = (time.monotonic() - self._stream_started_at + if self._stream_started_at else -1) + log.info("stream: stop received %.1fs after start, %d segments sent", + age, self._stream_segments) + self._stop_stream() -- cgit v1.2.3