diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 41 |
1 files changed, 36 insertions, 5 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index eab1cac..a892be2 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -297,6 +297,7 @@ class WebRTCPeerSession: # Flow control for video: how many segments the client says it can take. self._stream_credit = 0 self._stream_credit_evt = asyncio.Event() + self._stream_stopped = False self._gek_challenge: bytes | None = None # Same value as the GEK challenge, but kept for the life of the connection: # a join_request is signed over it, and it must stay verifiable after the @@ -384,6 +385,8 @@ class WebRTCPeerSession: asyncio.ensure_future(self._stream_video(msg)) elif mtype == MNP.STREAM_MORE: self._grant_stream_credit(msg) + elif mtype == MNP.STREAM_STOP: + self._stop_stream() else: log.warning("Unknown MNP message type on DataChannel: %s", mtype) except Exception as e: @@ -1203,6 +1206,7 @@ class WebRTCPeerSession: file_path, chunk_index, file_hash, + entry.id, ) self._send(chunk_data) if chunk_index == 0: @@ -1721,6 +1725,18 @@ class WebRTCPeerSession: self._stream_credit += max(0, min(n, STREAM_MAX_CREDIT)) self._stream_credit_evt.set() + def _stop_stream(self) -> None: + """ + The viewer was closed. Stop transcoding and let go of the slot. + + Without this the only thing that ended a stream was the credit timeout, + so ffmpeg kept running and held one of the node's two transcode slots + for two minutes after nobody was watching — which is how closing a video + made the next one answer "server busy". + """ + self._stream_stopped = True + self._stream_credit_evt.set() + async def _await_stream_credit(self) -> bool: """ Block until the client has room. False if it stopped asking. @@ -1730,6 +1746,8 @@ class WebRTCPeerSession: JavaScript array while MediaSource consumes it a segment at a time. """ while self._stream_credit <= 0: + if self._stream_stopped: + return False self._stream_credit_evt.clear() try: await asyncio.wait_for(self._stream_credit_evt.wait(), @@ -1738,6 +1756,8 @@ class WebRTCPeerSession: log.info("Stream stalled: no credit from peer=%s", (self._user_id or "?")[:8]) return False + if self._stream_stopped: + return False if self._channel is None or self._channel.readyState != "open": return False self._stream_credit -= 1 @@ -1809,12 +1829,17 @@ class WebRTCPeerSession: except (TypeError, ValueError): self._stream_credit = 0 paced = self._stream_credit > 0 + self._stream_stopped = False index = 0 try: while True: if paced and not await self._await_stream_credit(): break + if self._stream_stopped: + log.info("Stream stopped by peer=%s after %d segments", + (self._user_id or "?")[:8], index) + break data = await proc.stdout.read(STREAM_SEGMENT_SIZE) if not data: break @@ -1840,11 +1865,12 @@ class WebRTCPeerSession: pass await proc.wait() - self._send({ - "type": MNP.STREAM_END, - "v": MNP_VERSION, - "file_id": file_id, - }) + if not self._stream_stopped: + self._send({ + "type": MNP.STREAM_END, + "v": MNP_VERSION, + "file_id": file_id, + }) log.info("Streamed %s: %d segments", entry.name, index) self._audit("stream_video", entry.name) @@ -1868,6 +1894,7 @@ def _read_and_encrypt( file_path: Path, chunk_index: int, file_hash: bytes, + file_id: str = "", ) -> dict: with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) @@ -1879,6 +1906,10 @@ def _read_and_encrypt( return { "type": MNP.FILE_CHUNK, "v": MNP_VERSION, + # Named so a client running several downloads at once can tell whose + # reply this is. It used to carry only the index, which made matching a + # reply to its request a question of arrival order. + "file_id": file_id, "chunk_index": chunk_index, "plaintext_size": len(plaintext), "nonce": nonce, |