From fe5d6c3b3ff684b09e8070510f70c39854899cc8 Mon Sep 17 00:00:00 2001 From: foxyz Date: Fri, 9 Oct 2026 23:35:31 +0300 Subject: [PATCH] Relay HLS/DASH-only pages as pieces tracks (protocol 11) A track can be served as pieces: the server writes a VOD playlist whose entries are its own piece URLs (segments, init maps, AES-128 keys; byte ranges folded in; yt-dlp's DASH fragments become the same list). A piece buffer per track fills like the byte buffer. Picks a single file, then a split byte pair, then pieces. DRM, live, master playlists are refused. Co-Authored-By: Claude Opus 5.5 --- README.md | 10 +- .../lib/src/session/session_coordinator.dart | 31 +- docs/ARCHITECTURE.md | 80 +++- docs/PROTOCOL.md | 62 ++- docs/ROADMAP.md | 24 +- packages/core/lib/core.dart | 1 + packages/core/lib/src/link_probe.dart | 107 ++++- packages/core/lib/src/protocol.dart | 53 ++- packages/core/lib/src/relay.dart | 80 +++- packages/core/lib/src/relay_pieces.dart | 385 ++++++++++++++++++ packages/core/lib/src/yt_dlp.dart | 133 ++++-- packages/core/test/relay_pieces_test.dart | 130 ++++++ packages/core/test/yt_dlp_test.dart | 36 +- packages/server/lib/src/piece_buffer.dart | 368 +++++++++++++++++ packages/server/lib/src/relay_buffer.dart | 63 ++- packages/server/lib/src/relay_routes.dart | 166 +++++--- packages/server/lib/src/server_hub.dart | 11 +- packages/server/test/relay_test.dart | 60 ++- 18 files changed, 1593 insertions(+), 207 deletions(-) create mode 100644 packages/core/lib/src/relay_pieces.dart create mode 100644 packages/core/test/relay_pieces_test.dart create mode 100644 packages/server/lib/src/piece_buffer.dart diff --git a/README.md b/README.md index b2d4354..b764801 100644 --- a/README.md +++ b/README.md @@ -342,10 +342,12 @@ computer and are used by its yt-dlp only (your own playback, "through me", and a server you host). Cookies log yt-dlp in as you on those sites: keep the file private, and for YouTube consider a throwaway account (Google may flag accounts used with yt-dlp). -YouTube works in all three too. The last two never show the link to -anyone; they can't pass on live streams or pages that only offer streaming -playlists (use "Everyone fetches it" for those). Your upload has to beat the video's -bitrate: a 25-minute 2 GB episode is about 11 Mbit/s. +YouTube works in all three too, and so do pages and links that only +offer streaming playlists (HLS `.m3u8`, DASH `.mpd`, the HLS of X videos). +The last two never show the link to anyone; they can't pass on live +streams or DRM-protected video (use "Everyone fetches it" for live). +Your upload has to beat the video's bitrate: a 25-minute 2 GB episode is +about 11 Mbit/s. ## Video quality and shaders diff --git a/apps/desktop/lib/src/session/session_coordinator.dart b/apps/desktop/lib/src/session/session_coordinator.dart index 59b2fe3..fcd017a 100644 --- a/apps/desktop/lib/src/session/session_coordinator.dart +++ b/apps/desktop/lib/src/session/session_coordinator.dart @@ -544,9 +544,10 @@ class SessionCoordinator { } } - /// This viewer's `/r/` URL; an mpv EDL of every track when there are - /// several (a page's separate video and audio). - String _relayMediaUrlFor(String name, int mediaId, [int tracks = 1]) => + /// This viewer's `/r/` URL; with tracks, a pieces track's playlist, or + /// an mpv EDL of every track when there are several (a page's separate + /// video and audio). + String _relayMediaUrlFor(String name, int mediaId, [List tracks = const []]) => relayViewerUrl('http://$_wsHost:$_wsPort/r/default/${sync.token}/$mediaId', name, tracks: tracks); /// Serves [media] for media [mediaId] on this member's relay channel. @@ -687,7 +688,7 @@ class SessionCoordinator { _awaitLiveFor = r.mediaId; health.value = const HealthStatus(HealthSeverity.info, 'the server is getting the video…'); } else { - await _openMedia(_relayMediaUrlFor(name, r.mediaId, relay.tracks.length)); + await _openMedia(_relayMediaUrlFor(name, r.mediaId, relay.tracks)); } } else { var url = stateMediaUrl(r); @@ -707,7 +708,7 @@ class SessionCoordinator { } else if (_awaitLiveFor == r.mediaId && (stateMediaRelay(r)?.live ?? false)) { _awaitLiveFor = -1; await _openMedia( - _relayMediaUrlFor((r.media?['name'] as String?) ?? '', r.mediaId, stateMediaRelay(r)!.tracks.length)); + _relayMediaUrlFor((r.media?['name'] as String?) ?? '', r.mediaId, stateMediaRelay(r)!.tracks)); } _jellyfinTransport(r); _relayStatus(r); @@ -1262,21 +1263,27 @@ class SessionCoordinator { return null; } if (relay == LinkRelay.me) { - // This computer reads it: a file, or a page's formats (one file, or - // separate video and audio tracks). + // This computer reads it: a file, or a page's formats (one file, + // separate video and audio tracks, or segmented playlists). if (probe == null) return 'cannot pass this link on: $probeError'; final problem = probe.relayProblem; if (problem != null) return problem; final y = await ytDlp.ytDlp; final check = LinkRelayMedia(uri, ytDlp: y, probe: probe); var size = 0; - final List kinds; + final List specs; try { final tracks = await check.open().timeout(const Duration(seconds: 15)); for (final t in tracks) { - size += await t.bytes.size().timeout(const Duration(seconds: 15)); + // A playlist is read (DRM and live refused) but has no size. + final pieces = t.pieces; + if (pieces != null) { + await pieces.list().timeout(const Duration(seconds: 15)); + } else { + size += await t.bytes!.size().timeout(const Duration(seconds: 15)); + } } - kinds = [for (final t in tracks) t.kind]; + specs = relayTrackSpecs(tracks); } catch (e) { return 'cannot pass this link on: $e'; } finally { @@ -1284,11 +1291,11 @@ class SessionCoordinator { } _pendingOrigin = _LinkOrigin(url, probe, y); _swap(HelloMedia( - name: name, size: size, duration: 0, relay: MediaRelay(tracks: kinds.length > 1 ? kinds : const []))); + name: name, size: size, duration: 0, relay: MediaRelay(tracks: specs))); return null; } // The server reads it with its own yt-dlp; only a page this app knows - // it cannot relay (live, no video file) is refused here (the server + // it cannot relay (live, nothing a relay can read) is refused here (the server // may reach what this computer cannot). if (probe != null && probe.isPage && probe.info != null && probe.relayProblem != null) { return probe.relayProblem; diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 7561f76..88faa83 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -423,7 +423,8 @@ errors leave it out). A member's media through the server's memory, for viewers who cannot reach it themselves (`packages/core/lib/src/relay.dart` for the wire and -the source, `packages/server/lib/src/relay_buffer.dart`, +the source, `relay_pieces.dart` for segmented streams, +`packages/server/lib/src/relay_buffer.dart`, `piece_buffer.dart`, `relay_routes.dart`; wire in [PROTOCOL.md](PROTOCOL.md) "Relay"). What gets relayed (`SessionCoordinator.swapPicked`): @@ -450,6 +451,27 @@ streams in sync across seeks; one URL keeps the app's "one media URL" path (sync, swap, reopen) as is. The source member plays the page itself (yt-dlp hook) like any other link. +A page with only segmented streams (HLS `m3u8_native`, DASH +`http_dash_segments`, e.g. X's HLS, generic `.m3u8`/`.mpd`) is relayed as +**pieces** tracks (`"as":"pieces"`, one muxed playlist or video + audio): +the fetcher parses the upstream media playlist (`parseHlsPlaylist`), or +yt-dlp's DASH `fragments` (`dashPieces`: a first fragment without a +duration is the init section, so no MPD parser), into a +`RelayPieceList`: each segment, `EXT-X-MAP` init section and AES-128 key +is a piece, one upstream GET, with `EXT-X-BYTERANGE` folded into the +piece. The server serves a VOD media playlist it writes itself from that +list (`RelayPieceList.playlist`: durations, media sequence, +discontinuities, MAP, AES-128 keys with their IVs; every URI its own +`p/`), so a source can never point viewers anywhere else. One URL per +piece also avoids mpv's seek hang on byte-range fMP4 playlists (Apple's +`bipbop` hung on the first seek read directly). Refused: DRM (SAMPLE-AES, +`skd:`, any non-identity key format, `EXT-X-SESSION-KEY`, yt-dlp's +`has_drm`), a live playlist (no `EXT-X-ENDLIST`; yt-dlp's `is_live` +first), a master or I-frame playlist. A remux on the fetcher (ffmpeg +`-c copy` into one pipe) and yt-dlp's own download + merge were rejected: +the size is unknown until the end, a seek past the written part cannot be +served, and it needs ffmpeg or disk. + How bytes move: - The source dials `/r/.../source` (outbound, so NAT/CGNAT are fine) and answers the server's pulls from the file or the URL (`RelaySource` over @@ -467,8 +489,8 @@ How bytes move: 8.6 to 12.5 MB/s); chunked, a 1080p relay seeks in under 0.8 s (4 to 6.6 s unchunked) and even beats mpv reading YouTube directly, since the buffer reads ahead at full speed. -- The server keeps one `RelayBuffer` per track of a relayed media (one - for a file): 1 MiB slices in +- The server keeps one buffer per track of a relayed media (one for a + file), a `RelayBuffer` for bytes: 1 MiB slices in memory, filled by one sequential pull at a time. Kept: the first and last 2 MiB (container indexes; mpv reads both on open and on seeks), 64 MiB behind the slowest viewer and 256 MiB past the furthest, the @@ -480,9 +502,27 @@ How bytes move: each, which keeps both windows at the same playback distance (1080p video + aac: about 246 + 10 MiB ahead); each pins its own head and tail. A viewer is its token in every track's buffer. -- Filling: a viewer blocked on a slice first (oldest wait first, handing - the source over at slice boundaries), then the pinned slices, then - read-ahead from the slowest viewer. Nothing is fetched while nobody +- A pieces track gets a `PieceBuffer`: whole pieces by index, each one + pull (a piece names its index on the channel), served whole with Range + inside a piece. A viewer's position is its newest requested segment + (opening the playlist puts a new viewer at the first one, so its first + pieces are fetched before it asks). Filled: a blocked viewer's piece + (oldest wait first; a pull nobody waits on gives way at once), the init + and key pieces the viewers' segments use, then segments ahead of each + viewer up to the ahead budget, slowest first. Init and key pieces are + kept; segments are evicted by distance (behind costs 4x), sizes + estimated from the bytes per second of the pieces in so far (before + that 4 Mbit/s video, 160 kbit/s audio, which also splits the budget). + A failed pull is retried after 1 s. Measured (mpv 0.41 through the + dedicated server, loopback, ~30 MB/s upstream): Mux's TS HLS (1080p, + one muxed playlist) opened in 0.8 s, seeks 0.07 to 1.1 s; Akamai's DASH + (1080p + aac, two playlists, EDL) 1.3 s, seeks 0.03 to 1.4 s; an X + video's HLS video + audio 0.3 to 0.75 s, seeks 0.03 to 0.4 s; A/V sync + within 0.2 ms after every seek; a second viewer of the same pieces + opened in 0.02 s from the buffer. +- Filling a byte buffer: a viewer blocked on a slice first (oldest wait + first, handing the source over at slice boundaries), then the pinned + slices, then read-ahead from the slowest viewer. Nothing is fetched while nobody watches through the relay (a Jellyfin source with only account holders uploads nothing). - mpv leaves superseded requests open but unread after a seek: only a @@ -498,7 +538,8 @@ How bytes move: stop. A crash frees it with the process. - Source away: viewers keep playing from the buffer; a request waits up to 60 s for missing bytes. The source reconnects with backoff (same - token after a resume); a different size drops that track's buffer. + token after a resume); a different size (or piece count and roles) + drops that track's buffer. `state.media.relay.live` follows the channel, and viewers show "waiting for to pass the video on" while it is down. @@ -581,7 +622,12 @@ all three link modes. Nothing is bundled or installed by hand. and the server buffer, and mpv decodes them all), then bitrate; audio the original language, not a `-drc` variant unless alone, m4a/aac before opus, then bitrate. YouTube has only these (299 + 140 for a 1080p60 video). - - Neither (only HLS/DASH playlists) or a live stream: nothing to relay. + - Neither: segmented streams (`pickPieces`: `m3u8_native`/`m3u8` with a + media playlist URL, or `http_dash_segments` with http(s) `fragments`; + not `has_drm`): one muxed playlist (video and audio not known to be + absent) unless a video-only + audio-only pair is taller, with the + same height, codec and audio rules. Relayed as pieces (see "Relay"). + - A live stream, or nothing of the above: nothing to relay. - **Probe** (`probeLink`, `link_probe.dart`): a `Range: bytes=0-0` GET; a video/audio type, or a generic one on a media file name, is a file (`looksLikeMediaFile`); anything else goes to yt-dlp, and when yt-dlp @@ -595,14 +641,16 @@ all three link modes. Nothing is bundled or installed by hand. (`script-opts=ytdl_hook-ytdl_path=...`, `ytdl-format=bv*[height<=1080]+ba/b`, once per player), waiting for the download unless the URL names a media file. A hook error shows as "can't open this link: ...". -- **Through me**: the putter's probe must give tracks, else the swap is - refused ("Live streams can't be passed on", or no video file: only - streaming playlists) with a hint to use "Everyone fetches it". The swap - names the tracks; its relay source reads them with yt-dlp's headers - (`PageRelayMedia`). On 403/410 (expired format URLs, 6 h on YouTube) from - any track it resolves the page again once for all tracks, finds the - same format ids with the same sizes (else "the video changed on the - page") and goes on from where it stopped. Format URLs are bound to the +- **Through me**: the putter's probe must give tracks, and a pieces + track's playlist must be readable, else the swap is refused ("Live + streams can't be passed on", DRM, or nothing a relay can read) with a + hint to use "Everyone fetches it". The swap names the tracks; its relay + source reads them with yt-dlp's headers (`PageRelayMedia`). On 403/410 + (expired format URLs, 6 h on YouTube; signed segment URLs) from any + track it resolves the page again once for all tracks, finds the same + format ids with the same sizes and piece counts and roles (playlists + re-read and re-mapped by index; else "the video changed on the page") + and goes on from where it stopped. Format URLs are bound to the resolving machine's address, so they never leave it. - **Server fetches it**: the server probes the link itself (`LinkRelayMedia`) with its own yt-dlp (off with `--no-yt-dlp` / diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index cab8776..f7ca0ea 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -235,17 +235,22 @@ flow control, cases) is in [ARCHITECTURE.md](ARCHITECTURE.md), "Relay". swap media.relay = {} the swapper's machine serves the bytes media.relay = {"tracks":[{"kind":"video"},{"kind":"audio"}]} the same, as two tracks played together + media.relay = {"tracks":[{"kind":"video","as":"pieces"}]} + the same, as one segmented playlist media.relay = {"fetch":true} the server reads media.url itself state media.relay = {"source":"m2","live":true} or {"fetch":true,"live":true} media.relay = {"fetch":true,"live":true,"tracks":[{"kind":"video"},{"kind":"audio"}]} ``` -Tracks: a relayed media is one file, or several byte tracks (a page's -separate video and audio files) that viewers play as one. `tracks` lists -them in order, 2 to 4 entries of `kind` `video` or `audio`; absent (or -anything else) means one file. Every track is a Range-readable byte -stream; other kinds of track (e.g. segmented playlists) would add a field -to the entry. +Tracks: a relayed media is one file, or tracks that viewers play as one +(a page's separate video and audio). `tracks` lists them in order, up to +4 entries of `kind` `video` or `audio`, each a Range-readable byte stream, +or with `"as":"pieces"` a segmented stream (an HLS media playlist or DASH +fragments) relayed piece by piece and served as a VOD media playlist. A +piece is one segment, init section (`EXT-X-MAP`) or AES-128 key, fetched +upstream whole with one GET. At least 2 entries, or 1 pieces entry (a +muxed playlist, `kind` `video`); absent, an unknown `kind` or `as`, or +anything else means one file. - `swap`: `relay` with `jellyfin` keeps the ref (the swapper serves it to viewers without an account; `fetch` is ignored there). `relay` without @@ -269,19 +274,27 @@ to the entry. Clients build it after a swap like the `/m` one: `http:///r/default///` (name with `/` and `\` made `_`, then URL-encoded; the server ignores it). With `tracks`, - it is an mpv EDL of every track's URL - `http:///r/default////`, each framed by - its byte length: + each track's URL is `http:///r/default////` + (a pieces track: `.m3u8`); one track is opened as is, several as + an mpv EDL of them all, each framed by its byte length: `edl://!no_clip;%%;!new_stream;!no_clip;%%`. - `GET /r////`: the media (one file), Range table as for `/m`. `GET /r/////`: track `i` (from 0) of a media with `tracks` (track 0 of a one-file media - too). Errors: 401 bad token, 404 not the current relayed - media (a player still on an older `mediaId` gets 404, never the new - bytes) or no such track (the one-file URL of a media with `tracks`), 410 - unknown session, 503 + `retry-after` the source never connected or the - server could not read the link (60 s). A request whose bytes do not - arrive within 60 s ends early. Bytes count toward `bytesServed`. + too); for a pieces track (any name), its media playlist + (`application/vnd.apple.mpegurl`): `#EXT-X-PLAYLIST-TYPE:VOD`, + `EXT-X-MEDIA-SEQUENCE`, `EXTINF` durations, `EXT-X-DISCONTINUITY`, + `EXT-X-MAP` and `EXT-X-KEY:METHOD=AES-128` (with `IV` when given) as + upstream, `EXT-X-ENDLIST`; every URI is `p/`. `GET + /r/////p/`: piece `j` of pieces track `i`, + whole, once it is in (a Range inside the piece answers 206). Errors: 401 + bad token, 404 not the current relayed media (a player still on an + older `mediaId` gets 404, never the new bytes), no such track (the + one-file URL of a media with `tracks`) or piece (or `p/` of a byte + track), 410 unknown session, 503 + `retry-after` the source never + connected or the server could not read the link (60 s), or a piece did + not arrive within 60 s. A byte request whose bytes do not arrive within + 60 s ends early. Bytes count toward `bytesServed`. - `GET /r///source` (WebSocket): the source's relay channel, dialled by the source (works behind NAT). Only the current source's token is accepted; anyone else, or the source after the next swap, is @@ -293,18 +306,31 @@ source -> server {"t":"info","size":} first frame, one file {"t":"info","tracks":[{"kind":"video","size":}, {"kind":"audio","size":}]} first frame, tracks + {"t":"info","tracks":[{"kind":"video","as":"pieces","seq":5, + "pieces":[{"init":true},{"key":true}, + {"d":6.0,"map":0,"key":1,"iv":"0x..."}, + {"d":4.5,"map":0,"disc":true}]}]} + a pieces track binary: [u32 big-endian pull id][bytes] in order per pull {"t":"end","id":1} {"t":"fail","id":1,"reason":"..."} server -> source {"t":"pull","id":1,"track":0,"start":0,"end":1048575} inclusive; no track: 0 + {"t":"pull","id":2,"track":1,"piece":17} piece 17, whole {"t":"ack","id":1,"n":65536} bytes arrived {"t":"cancel","id":1} stop that pull ``` Pull ids are unique across the tracks of a connection; `track` is sent - only for a media with `tracks`. An `info` whose track count differs - from the swap's `tracks` closes the channel with 4001. A track `size` - different from the one before (a reconnect) drops that track's buffer. + only for a media with `tracks`. A pieces track's `pieces` are its + pieces in order: `{"init":true}` an init section, `{"key":true}` an + AES-128 key, else a segment of `d` seconds whose `map` / `key` name the + init and key pieces it uses, `iv` an explicit IV (`0x` + hex), `disc` a + discontinuity before it; `seq` (default 0) the media sequence of the + first segment. A piece pull's bytes end with `end` like a range pull. + An `info` whose tracks differ from the swap's `tracks` (count, bytes or + pieces), or a malformed piece list, closes the channel with 4001. A + track `size` or piece count/roles different from the one before (a + reconnect) drops that track's buffer. ## Join and reconnect diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index bf92d4a..a32ea03 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -19,22 +19,18 @@ off by deleting it in the same commit as the code. Wire changes also edit through media_kit; if not, resolve in the app and open the relay tracks' formats. YouTube: yt-dlp wants a JS runtime (deno) for full extraction. Formats that need yt-dlp's cookies. - - Relay "pieces" tracks for pages with only HLS/DASH (`m3u8_native`, - `http_dash_segments`): a track served as a synthesized VOD media - playlist whose entries are pieces (one segment, init `EXT-X-MAP` or - AES-128 key = one upstream GET, served whole, Range inside a piece - honoured), with a piece buffer per track (pieces by index, filled - like `RelayBuffer`: a blocked viewer, the init piece, then ahead of - the slowest viewer; evicted by distance); `EXT-X-BYTERANGE` folded - into pieces (it also avoids mpv's seek hang on byte-range fMP4 - playlists); yt-dlp's DASH `fragments` turned into the same playlist, - so no MPD parser. Wire: a track entry gains `"as":"pieces"`, `info` - lists piece durations, a pull names a piece. Refuse DRM - (`skd:`/SAMPLE-AES/Widevine). Measured in a spike: piece relay of - Mux/Apple/Akamai HLS/DASH opened in 1.5 to 2.2 s, seeks 0.02 to 2.4 s. - Live streams through the relay (today refused: the sync model needs a fixed duration and seeking). - - `tool/relay_e2e.dart`: a two-track case over a shaped source link. + - `tool/relay_e2e.dart`: a two-track case (bytes and pieces) over a + shaped source link. Pieces on a slow source: a cold piece can take + longer than media_kit's 5 s `network-timeout`; if it does, fetch the + first pieces before setting `relay.live`. + - Pieces, unverified so far: the 403/410 re-resolve against a real + expiring HLS (signed segment URLs), HLS discontinuities / ad + splicing, Windows' bundled ffmpeg with extensionless piece URLs + (`p/`; fine with ffmpeg 9 on Linux). + - Thumbnails and scrubbing previews of a relayed media with tracks + should open the video track's URL alone, not the EDL. 4. **AniList account**: sign in (OAuth via a localhost callback), "My list" filter on the Schedule; today's "Following" (AnimeFlow's monitored diff --git a/packages/core/lib/core.dart b/packages/core/lib/core.dart index 01bed06..572c04d 100644 --- a/packages/core/lib/core.dart +++ b/packages/core/lib/core.dart @@ -5,6 +5,7 @@ export 'src/clock_sync.dart'; export 'src/config.dart'; export 'src/protocol.dart'; export 'src/relay.dart'; +export 'src/relay_pieces.dart'; export 'src/link_probe.dart'; export 'src/yt_dlp.dart'; export 'src/cookies.dart'; diff --git a/packages/core/lib/src/link_probe.dart b/packages/core/lib/src/link_probe.dart index 62be871..c334bb5 100644 --- a/packages/core/lib/src/link_probe.dart +++ b/packages/core/lib/src/link_probe.dart @@ -5,6 +5,7 @@ import 'dart:typed_data'; import 'constants.dart'; import 'relay.dart'; +import 'relay_pieces.dart'; import 'yt_dlp.dart'; // What a pasted link is: a media file (range-readable as is) or a web page @@ -71,12 +72,8 @@ String _unescape(String s) => s String notAVideoText(String? ytDlpError) => ytDlpError == null ? 'Not a video' : "Not a video (and yt-dlp couldn't read it: $ytDlpError)"; -/// The text for a page with no video file a relay can read. -const noRelayFormatText = 'This page has no video file to pass on (only streaming playlists). ' - "Choose 'Everyone fetches it' instead."; - -/// The text for a live stream. -const liveStreamText = "Live streams can't be passed on. Choose 'Everyone fetches it' instead."; +/// The text for a page with no video a relay can read. +const noRelayFormatText = "This page has no video a relay can read. Choose 'Everyone fetches it' instead."; class LinkProbeException implements Exception { final String message; @@ -190,18 +187,22 @@ Future _readText(HttpClientResponse res) async { return utf8.decode(out.takeBytes(), allowMalformed: true); } -/// A page's relay tracks (see [YtDlpInfo.tracks]: one file, or video + -/// audio), range-read with yt-dlp's headers in GETs of the format's -/// `http_chunk_size`. Format URLs expire: on 403/410 from any track the -/// page is resolved again once for all tracks (same formats, same sizes) -/// and the read goes on from where it stopped. +/// A page's relay tracks (see [YtDlpInfo.tracks]: one file, video + +/// audio, or segmented playlists). A byte track is range-read with +/// yt-dlp's headers in GETs of the format's `http_chunk_size`; a pieces +/// track's playlist (or DASH fragments) is read on first use, then each +/// piece is one GET. Format URLs expire: on 403/410 from any track the page +/// is resolved again once for all tracks (same format ids, same sizes and +/// piece counts) and the read goes on from where it stopped. class PageRelayMedia implements RelayMedia { final Uri page; final YtDlp ytDlp; List? _picked; List _inner = const []; + List _pieces = const []; List? _tracks; final _sizes = {}; + final _http = HttpClient(); var _gen = 0; Future? _refreshing; PageRelayMedia(this.page, this.ytDlp, {List? tracks}) : _picked = tracks; @@ -218,7 +219,13 @@ class PageRelayMedia implements RelayMedia { final picked = _picked ??= (await _info()).tracks; if (picked.isEmpty) throw const LinkProbeException(noRelayFormatText); _inner = List.filled(picked.length, null, growable: false); - return _tracks = [for (var i = 0; i < picked.length; i++) RelayTrack(picked[i].kind, _PageTrack(this, i))]; + _pieces = List.filled(picked.length, null, growable: false); + return _tracks = [ + for (var i = 0; i < picked.length; i++) + picked[i].format.isPieces + ? RelayTrack.pieces(picked[i].kind, _PagePieces(this, i)) + : RelayTrack(picked[i].kind, _PageTrack(this, i)), + ]; } HttpRelayBytes _open(int i) => _inner[i] ??= () { @@ -227,26 +234,45 @@ class PageRelayMedia implements RelayMedia { headers: f.headers, chunkSize: f.chunkSize ?? RelayConstants.relayUpstreamChunkBytes); }(); + /// Reads track [i]'s playlist (or DASH fragments) into pieces. + Future _loadPieces(YtDlpFormat f) async { + final frags = f.fragments; + if (f.protocol == 'http_dash_segments') { + if (frags == null) throw const RelayPiecesException('no DASH fragments'); + return dashPieces(frags); + } + return fetchHlsPlaylist(_http, f.url, f.headers); + } + static bool _expired(Object e) => e is UpstreamHttpException && (e.status == 403 || e.status == 410); - /// Resolves the page again: the same format ids with the same sizes, - /// else the video changed. + /// Resolves the page again: the same format ids with the same sizes and + /// piece shapes, else the video changed. Future _refresh(int gen) { if (gen != _gen) return Future.value(); return _refreshing ??= () async { try { + const changed = LinkProbeException('the video changed on the page'); final info = await _info(); final next = []; for (final t in _picked!) { final f = info.formats[t.format.id]; - if (f == null) throw const LinkProbeException('the video changed on the page'); + if (f == null || f.isPieces != t.format.isPieces) throw changed; next.add((kind: t.kind, format: f)); } + final pieces = List.filled(next.length, null, growable: false); + for (var i = 0; i < next.length; i++) { + final had = _pieces[i]; + if (had == null) continue; + final p = pieces[i] = await _loadPieces(next[i].format); + if (!p.list.sameShape(had.list)) throw changed; + } final old = _inner; _picked = next; + _pieces = pieces; _inner = List.filled(next.length, null, growable: false); for (final e in _sizes.entries) { - if (await _open(e.key).size() != e.value) throw const LinkProbeException('the video changed on the page'); + if (await _open(e.key).size() != e.value) throw changed; } _gen++; for (final b in old) { @@ -271,6 +297,19 @@ class PageRelayMedia implements RelayMedia { } } + Future _piecesOf(int i) async { + final known = _pieces[i]; + if (known != null) return known; + final gen = _gen; + try { + return _pieces[i] = await _loadPieces(_picked![i].format); + } catch (e) { + if (!_expired(e)) rethrow; + await _refresh(gen); + return _pieces[i] ??= await _loadPieces(_picked![i].format); + } + } + Stream> _read(int i, int start, int end) async* { var pos = start; var retried = false; @@ -290,11 +329,31 @@ class PageRelayMedia implements RelayMedia { } } + /// Piece [index] of track [i]; on expiry once more after a re-resolve + /// (nothing was sent yet: an expired URL fails before its first byte). + Stream> _readPiece(int i, int index) async* { + var retried = false; + while (true) { + final gen = _gen; + final p = await _piecesOf(i); + if (index >= p.sources.length) throw RangeError.index(index, p.sources, 'piece'); + try { + yield* readPiece(_http, p.sources[index], _picked![i].format.headers); + return; + } catch (e) { + if (retried || !_expired(e)) rethrow; + retried = true; + await _refresh(gen); + } + } + } + @override Future close() async { for (final b in _inner) { await b?.close(); } + _http.close(force: true); } } @@ -314,6 +373,22 @@ class _PageTrack implements RelayBytes { Future close() async {} } +class _PagePieces implements RelayPieces { + final PageRelayMedia media; + final int index; + _PagePieces(this.media, this.index); + + @override + Future list() async => (await media._piecesOf(index)).list; + + @override + Stream> read(int piece) => media._readPiece(index, piece); + + /// The media closes its tracks. + @override + Future close() async {} +} + /// A link of either kind, probed on first [open] (or given [probe]): a /// file as one track ([HttpRelayBytes]), a page as its picked formats /// ([PageRelayMedia]). Throws [LinkProbeException] with the reason when it diff --git a/packages/core/lib/src/protocol.dart b/packages/core/lib/src/protocol.dart index 9d08e72..b4ec910 100644 --- a/packages/core/lib/src/protocol.dart +++ b/packages/core/lib/src/protocol.dart @@ -144,46 +144,75 @@ class HelloMedia { } } +/// One track of a relayed media: [kind] `video` or `audio`; a byte stream +/// (a Range-readable file) unless [pieces] (a segmented playlist relayed +/// piece by piece, served as a media playlist). JSON `{"kind":"video"}`, +/// `{"kind":"audio","as":"pieces"}`. +class RelayTrackSpec { + final String kind; + final bool pieces; + const RelayTrackSpec(this.kind, {this.pieces = false}); + + Map toJson() => {'kind': kind, if (pieces) 'as': 'pieces'}; + + /// Null when [j] is not a track entry this side knows. + static RelayTrackSpec? fromJson(Object? j) { + if (j is! Map || !MediaRelay.trackKinds.contains(j['kind'])) return null; + final as = j['as']; + if (as != null && as != 'bytes' && as != 'pieces') return null; + return RelayTrackSpec(j['kind'] as String, pieces: as == 'pieces'); + } + + @override + bool operator ==(Object other) => other is RelayTrackSpec && other.kind == kind && other.pieces == pieces; + @override + int get hashCode => Object.hash(kind, pieces); + @override + String toString() => pieces ? '$kind pieces' : kind; +} + /// Relayed media: the bytes go through the server's buffer (`/r/...`). /// Either from the [source] member's machine (it dials the relay channel), /// or, with [fetch], the server reads the swap's link itself. A swap sends /// `{}` or `{"fetch":true}`; the server fills in [source] (the swapper's /// member id, empty with [fetch]) and whether bytes can flow ([live]). -/// [tracks]: the kinds (`video`, `audio`) of a media relayed as several -/// tracks played together; empty for one file. A member swap names them, -/// the server fills them in for [fetch] once it has read the link. +/// [tracks]: the tracks of a media relayed as several played together, or +/// as one segmented playlist; empty for one file. A member swap names +/// them, the server fills them in for [fetch] once it has read the link. class MediaRelay { final String source; final bool fetch; final bool live; - final List tracks; + final List tracks; const MediaRelay({this.source = '', this.fetch = false, this.live = false, this.tracks = const []}); static const trackKinds = {'video', 'audio'}; static const maxTracks = 4; - MediaRelay copyWith({bool? live, List? tracks}) => + /// [tracks] as sent: several, or one pieces track; else one file. + static bool validTracks(List t) => + t.length <= maxTracks && (t.length > 1 || (t.length == 1 && t.single.pieces)); + + MediaRelay copyWith({bool? live, List? tracks}) => MediaRelay(source: source, fetch: fetch, live: live ?? this.live, tracks: tracks ?? this.tracks); Map toJson() => { if (source.isNotEmpty) 'source': source, if (fetch) 'fetch': true, if (live) 'live': true, - if (tracks.length > 1) 'tracks': [for (final k in tracks) {'kind': k}], + if (validTracks(tracks)) 'tracks': [for (final t in tracks) t.toJson()], }; - /// Unknown kinds, a single entry or too many make it one file. + /// Unknown kinds, a single byte track or too many make it one file. factory MediaRelay.fromJson(Map json) { final t = json['tracks']; - final kinds = t is List - ? [for (final e in t) e is Map && e['kind'] is String ? e['kind'] as String : ''] - : []; - final ok = kinds.length > 1 && kinds.length <= maxTracks && kinds.every(trackKinds.contains); + final specs = t is List ? [for (final e in t) RelayTrackSpec.fromJson(e)] : const []; + final ok = !specs.contains(null) && validTracks(specs.cast()); return MediaRelay( source: json['source'] is String ? json['source'] as String : '', fetch: json['fetch'] == true, live: json['live'] == true, - tracks: ok ? List.unmodifiable(kinds) : const []); + tracks: ok ? List.unmodifiable(specs.cast()) : const []); } } diff --git a/packages/core/lib/src/relay.dart b/packages/core/lib/src/relay.dart index a942e36..3f2f2da 100644 --- a/packages/core/lib/src/relay.dart +++ b/packages/core/lib/src/relay.dart @@ -7,20 +7,26 @@ import 'dart:typed_data'; import 'constants.dart'; import 'logging.dart'; +import 'protocol.dart'; +import 'relay_pieces.dart'; // The relay channel: a member's machine (the source) serves the current // media's bytes to the server, which buffers them for every viewer (`/r/...`). // The source dials out (works behind NAT/CGNAT) on its own WebSocket, so // video bytes never queue behind `/sync` messages. A media is one track (a -// file) or several (a page's separate video and audio). +// file) or several (a page's separate video and audio); a track is a +// byte stream or pieces (a segmented playlist, see relay_pieces.dart). // // source -> server {"t":"info","size":N} first frame, one track // {"t":"info","tracks":[{"kind":"video","size":N},...]} +// {"t":"info","tracks":[{"kind":"video","as":"pieces", +// "pieces":[{"init":true},{"d":6.0,"map":0},...]}]} // binary [u32 BE pull id][bytes] in order per pull // {"t":"end","id":1} pull complete // {"t":"fail","id":1,"reason":"..."} cannot read // server -> source {"t":"pull","id":1,"track":0,"start":0,"end":N} // bytes start..end +// {"t":"pull","id":2,"track":0,"piece":17} one piece, whole // {"t":"ack","id":1,"n":65536} n bytes arrived // {"t":"cancel","id":1} stop this pull // @@ -37,11 +43,14 @@ String relayUrlName(String name) => Uri.encodeComponent(name.replaceAll(RegExp(r /// What a viewer opens for relayed media at [base] /// (`http:///r///`): `/` for one -/// track, else an mpv EDL playing `//` of every track -/// together (see [mpvEdl]). -String relayViewerUrl(String base, String name, {int tracks = 1}) { +/// file; with [tracks], each track's URL (`//`, a pieces +/// track's playlist `//.m3u8`), several played together as +/// an mpv EDL (see [mpvEdl]). +String relayViewerUrl(String base, String name, {List tracks = const []}) { final n = relayUrlName(name); - return tracks <= 1 ? '$base/$n' : mpvEdl([for (var i = 0; i < tracks; i++) '$base/$i/$n']); + if (tracks.isEmpty) return '$base/$n'; + final urls = [for (var i = 0; i < tracks.length; i++) '$base/$i/$n${tracks[i].pieces ? '.m3u8' : ''}']; + return urls.length == 1 ? urls.single : mpvEdl(urls); } /// An mpv `edl://` URL playing [urls] as the streams of one file (a video @@ -244,14 +253,40 @@ class RelayWindow { } } -/// One byte stream of a relayed media: [kind] is `video` or `audio` when -/// a media has several, empty for a single file. +/// Where a source reads a pieces track from (see relay_pieces.dart). +abstract interface class RelayPieces { + /// The track's shape; a playlist is fetched on the first call. Throws + /// [RelayPiecesException] when it cannot be relayed (DRM, live). + Future list(); + + /// Piece [index], whole, in order. Cancelling stops reading. + Stream> read(int index); + + Future close(); +} + +/// One track of a relayed media: [kind] is `video` or `audio` when a +/// media has several (or is pieces), empty for a single file. Read as +/// [bytes] (a Range-readable stream) or as [pieces]. class RelayTrack { final String kind; - final RelayBytes bytes; - const RelayTrack(this.kind, this.bytes); + final RelayBytes? bytes; + final RelayPieces? pieces; + const RelayTrack(this.kind, RelayBytes this.bytes) : pieces = null; + const RelayTrack.pieces(this.kind, RelayPieces this.pieces) : bytes = null; + + RelayTrackSpec get spec => RelayTrackSpec(kind.isEmpty ? 'video' : kind, pieces: pieces != null); } +/// The wire's `relay.tracks` for [tracks]: empty for one file. +List relayTrackSpecs(List tracks) => + tracks.length == 1 && tracks.single.pieces == null ? const [] : [for (final t in tracks) t.spec]; + +/// A track's `info` entry: its kind and size, or its pieces. +Future> relayTrackInfo(RelayTrack t) async => t.pieces != null + ? {...t.spec.toJson(), ...(await t.pieces!.list()).toJson()} + : {'kind': t.kind, 'size': await t.bytes!.size()}; + /// What a relay serves: one track, or several played together (a page's /// separate video and audio). abstract interface class RelayMedia { @@ -317,16 +352,11 @@ class RelaySource { /// when it cannot be read. Future start() async { final tracks = await media.open(); - final sizes = [for (final t in tracks) await t.bytes.size()]; + final infos = [for (final t in tracks) await relayTrackInfo(t)]; _tracks = tracks; unawaited(_loop({ 't': 'info', - if (tracks.length == 1) - 'size': sizes.single - else - 'tracks': [ - for (var i = 0; i < tracks.length; i++) {'kind': tracks[i].kind, 'size': sizes[i]}, - ], + if (relayTrackSpecs(tracks).isEmpty) 'size': infos.single['size'] else 'tracks': infos, })); } @@ -373,12 +403,20 @@ class RelaySource { if (id is! int) return; switch (j['t']) { case 'pull': - final start = j['start'], end = j['end'], track = j['track'] ?? 0; - if (start is! int || end is! int || start < 0 || end < start) return; + final start = j['start'], end = j['end'], track = j['track'] ?? 0, piece = j['piece']; if (track is! int || track < 0 || track >= _tracks.length) return; + final t = _tracks[track]; + final Stream> Function() read; + if (piece is int) { + if (t.pieces == null || piece < 0) return; + read = () => t.pieces!.read(piece); + } else { + if (start is! int || end is! int || start < 0 || end < start || t.bytes == null) return; + read = () => t.bytes!.read(start, end); + } final p = _Pull(id); _pulls[id] = p; - unawaited(_serve(ws, p, _tracks[track].bytes, start, end)); + unawaited(_serve(ws, p, read)); case 'ack': final p = _pulls[id]; final n = j['n']; @@ -409,8 +447,8 @@ class RelaySource { _pulls.clear(); } - Future _serve(WebSocket ws, _Pull p, RelayBytes bytes, int start, int end) async { - final it = p.it = StreamIterator(bytes.read(start, end)); + Future _serve(WebSocket ws, _Pull p, Stream> Function() read) async { + final it = p.it = StreamIterator(read()); final pending = BytesBuilder(copy: false); const frame = RelayConstants.relayFrameBytes; try { diff --git a/packages/core/lib/src/relay_pieces.dart b/packages/core/lib/src/relay_pieces.dart new file mode 100644 index 0000000..1a128ab --- /dev/null +++ b/packages/core/lib/src/relay_pieces.dart @@ -0,0 +1,385 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; +import 'dart:math'; + +import 'relay.dart'; + +// Pieces tracks: a segmented stream (an HLS media playlist, or yt-dlp's +// DASH fragments) relayed piece by piece. A piece is one upstream GET +// served whole: a segment, an init section (`EXT-X-MAP`) or an AES-128 +// key. The fetcher (the source member, or the server for a link it reads +// itself) parses the upstream playlist into a [RelayPieceList] (no URLs); +// the server turns that into the VOD media playlist viewers open, whose +// entries are its own `p/` piece URLs (see [RelayPieceList.playlist]). + +/// Why a segmented stream cannot be relayed. +class RelayPiecesException implements Exception { + final String message; + const RelayPiecesException(this.message); + @override + String toString() => message; +} + +/// The text for DRM-protected media. +const drmText = "This video is DRM-protected; it can't be relayed."; + +/// The text for a live stream. +const liveStreamText = "Live streams can't be passed on. Choose 'Everyone fetches it' instead."; + +enum RelayPieceRole { segment, init, key } + +/// One piece of a [RelayPieceList]. A segment has its duration [d], the +/// init piece [map] and AES-128 key piece [key] (with an explicit [iv]) +/// that apply to it, and [disc] when an `EXT-X-DISCONTINUITY` precedes it. +class RelayPiece { + final RelayPieceRole role; + final double d; + final int? map, key; + final String? iv; + final bool disc; + const RelayPiece.segment(this.d, {this.map, this.key, this.iv, this.disc = false}) + : role = RelayPieceRole.segment; + const RelayPiece.init() + : role = RelayPieceRole.init, + d = 0, + map = null, + key = null, + iv = null, + disc = false; + const RelayPiece.key() + : role = RelayPieceRole.key, + d = 0, + map = null, + key = null, + iv = null, + disc = false; + + bool get isSegment => role == RelayPieceRole.segment; + + Map toJson() => switch (role) { + RelayPieceRole.init => {'init': true}, + RelayPieceRole.key => {'key': true}, + RelayPieceRole.segment => { + 'd': d, + if (map != null) 'map': map, + if (key != null) 'key': key, + if (iv != null) 'iv': iv, + if (disc) 'disc': true, + }, + }; +} + +final _ivRe = RegExp(r'^0[xX][0-9a-fA-F]{1,32}$'); + +/// The shape of a pieces track: what the server needs to serve it (no +/// upstream URLs). Sent in the relay channel's `info` (see PROTOCOL.md). +class RelayPieceList { + final List pieces; + /// `EXT-X-MEDIA-SEQUENCE` of the first segment (AES-128 without an + /// explicit IV uses the segment's sequence number). + final int seq; + const RelayPieceList(this.pieces, {this.seq = 0}); + + static const maxPieces = 200000; + + int get length => pieces.length; + double get duration => pieces.fold(0.0, (a, p) => a + p.d); + + /// Same piece count and roles (a re-resolve or a reconnect of the same + /// media). + bool sameShape(RelayPieceList o) { + if (o.length != length) return false; + for (var i = 0; i < length; i++) { + if (o.pieces[i].role != pieces[i].role) return false; + } + return true; + } + + Map toJson() => { + if (seq != 0) 'seq': seq, + 'pieces': [for (final p in pieces) p.toJson()], + }; + + /// Parses and checks a track entry's `seq` / `pieces`; null when + /// malformed (an index out of range or of the wrong role, a bad IV). + static RelayPieceList? fromJson(Map j) { + final raw = j['pieces']; + final seq = j['seq'] ?? 0; + if (raw is! List || raw.isEmpty || raw.length > maxPieces || seq is! int || seq < 0) return null; + final roles = []; + for (final e in raw) { + if (e is! Map) return null; + roles.add(e['init'] == true + ? RelayPieceRole.init + : e['key'] == true + ? RelayPieceRole.key + : RelayPieceRole.segment); + } + int? ref(Object? v, RelayPieceRole want) { + if (v == null) return null; + if (v is! int || v < 0 || v >= roles.length || roles[v] != want) throw const FormatException(); + return v; + } + + final out = []; + try { + for (var i = 0; i < raw.length; i++) { + final e = raw[i] as Map; + switch (roles[i]) { + case RelayPieceRole.init: + out.add(const RelayPiece.init()); + case RelayPieceRole.key: + out.add(const RelayPiece.key()); + case RelayPieceRole.segment: + final d = e['d'], iv = e['iv']; + if (d is! num || !d.isFinite || d < 0 || d > 3600) return null; + if (iv != null && (iv is! String || !_ivRe.hasMatch(iv))) return null; + out.add(RelayPiece.segment(d.toDouble(), + map: ref(e['map'], RelayPieceRole.init), + key: ref(e['key'], RelayPieceRole.key), + iv: iv as String?, + disc: e['disc'] == true)); + } + } + } on FormatException { + return null; + } + if (!out.any((p) => p.isSegment)) return null; + return RelayPieceList(List.unmodifiable(out), seq: seq); + } + + /// The VOD media playlist viewers open: every entry is `p/` + /// (relative to the playlist's URL), init and key pieces included. + String playlist() { + final segs = pieces.where((p) => p.isSegment); + final target = max(1, segs.map((p) => p.d).fold(0.0, max).ceil()); + final b = StringBuffer() + ..writeln('#EXTM3U') + ..writeln('#EXT-X-VERSION:6') + ..writeln('#EXT-X-TARGETDURATION:$target') + ..writeln('#EXT-X-MEDIA-SEQUENCE:$seq') + ..writeln('#EXT-X-PLAYLIST-TYPE:VOD'); + int? map, key; + String? iv; + for (var i = 0; i < pieces.length; i++) { + final p = pieces[i]; + if (!p.isSegment) continue; + if (p.disc) b.writeln('#EXT-X-DISCONTINUITY'); + if (p.map != null && p.map != map) b.writeln('#EXT-X-MAP:URI="p/${p.map}"'); + if (p.key != key || p.iv != iv) { + b.writeln(p.key == null + ? '#EXT-X-KEY:METHOD=NONE' + : '#EXT-X-KEY:METHOD=AES-128,URI="p/${p.key}"${p.iv == null ? '' : ',IV=${p.iv}'}'); + } + map = p.map; + key = p.key; + iv = p.iv; + b + ..writeln('#EXTINF:${p.d.toStringAsFixed(6)},') + ..writeln('p/$i'); + } + b.writeln('#EXT-X-ENDLIST'); + return b.toString(); + } +} + +/// Where the fetcher reads piece `i`: one GET of [url], of [length] bytes +/// at [offset] when it is a byte range of a bigger file. +typedef PieceSource = ({Uri url, int? offset, int? length}); + +/// A parsed upstream stream: the shape plus where each piece lives. +typedef UpstreamPieces = ({RelayPieceList list, List sources}); + +Map _attrs(String s) { + final out = {}; + for (final m in RegExp(r'([A-Z0-9-]+)=("[^"]*"|[^,]*)').allMatches(s)) { + var v = m.group(2)!; + if (v.length >= 2 && v.startsWith('"') && v.endsWith('"')) v = v.substring(1, v.length - 1); + out[m.group(1)!] = v; + } + return out; +} + +/// An HLS media playlist at [base] as pieces: every segment, `EXT-X-MAP` +/// and AES-128 `EXT-X-KEY` becomes a piece; `EXT-X-BYTERANGE` (and a MAP's +/// `BYTERANGE`) is folded into the piece (the offset defaults to the end +/// of the previous range of the same URI). Throws [RelayPiecesException] +/// for DRM (SAMPLE-AES, `skd:`, any non-identity key format), a live +/// playlist (no `EXT-X-ENDLIST`), a master playlist or no segments. +UpstreamPieces parseHlsPlaylist(String text, Uri base) { + final lines = const LineSplitter().convert(text).map((l) => l.trim()).toList(); + if (lines.isEmpty || !lines.first.startsWith('#EXTM3U')) { + throw const RelayPiecesException('not an HLS playlist'); + } + final pieces = []; + final sources = []; + final lastEnd = {}; + final shared = {}; // init/key piece by "url offset length" + var seq = 0; + var ended = false; + double? extinf; + int? brLen, brOff, map, key; + String? iv; + var disc = false; + + int add(RelayPiece p, Uri url, int? len, int? off) { + if (len != null) { + off ??= lastEnd['$url'] ?? 0; + lastEnd['$url'] = off + len; + } + pieces.add(p); + sources.add((url: url, offset: len == null ? null : off, length: len)); + return pieces.length - 1; + } + + int sharedPiece(RelayPiece p, String uri, (int, int?)? range) { + final url = base.resolve(uri); + if (!url.isScheme('http') && !url.isScheme('https')) throw const RelayPiecesException(drmText); + final k = '${p.role} $url ${range?.$2} ${range?.$1}'; + return shared[k] ??= add(p, url, range?.$1, range?.$2); + } + + (int, int?) range(String v) { + final parts = v.split('@'); + final len = int.tryParse(parts[0]); + final off = parts.length > 1 ? int.tryParse(parts[1]) : null; + if (len == null || len <= 0 || (parts.length > 1 && (off == null || off < 0))) { + throw const RelayPiecesException('bad byte range in the playlist'); + } + return (len, off); + } + + for (final line in lines) { + if (line.isEmpty) continue; + if (!line.startsWith('#')) { + final url = base.resolve(line); + if (!url.isScheme('http') && !url.isScheme('https')) { + throw const RelayPiecesException('the playlist has a segment that is not http(s)'); + } + add(RelayPiece.segment(extinf ?? 0, map: map, key: key, iv: iv, disc: disc), url, brLen, brOff); + extinf = null; + brLen = brOff = null; + disc = false; + continue; + } + final colon = line.indexOf(':'); + final tag = colon < 0 ? line : line.substring(0, colon); + final value = colon < 0 ? '' : line.substring(colon + 1); + switch (tag) { + case '#EXT-X-STREAM-INF': + throw const RelayPiecesException('a master playlist, not a media playlist'); + case '#EXT-X-I-FRAMES-ONLY': + throw const RelayPiecesException('an I-frame playlist'); + case '#EXT-X-SESSION-KEY': + throw const RelayPiecesException(drmText); + case '#EXT-X-MEDIA-SEQUENCE': + seq = int.tryParse(value) ?? 0; + case '#EXT-X-ENDLIST': + ended = true; + case '#EXT-X-DISCONTINUITY': + disc = true; + case '#EXTINF': + extinf = double.tryParse(value.split(',').first.trim()); + case '#EXT-X-BYTERANGE': + (brLen, brOff) = range(value); + case '#EXT-X-MAP': + final a = _attrs(value); + final uri = a['URI']; + if (uri == null) throw const RelayPiecesException('EXT-X-MAP without a URI'); + map = sharedPiece(const RelayPiece.init(), uri, a['BYTERANGE'] == null ? null : range(a['BYTERANGE']!)); + case '#EXT-X-KEY': + final a = _attrs(value); + final method = (a['METHOD'] ?? '').toUpperCase(); + if (method == 'NONE') { + key = null; + iv = null; + continue; + } + final format = a['KEYFORMAT']; + final uri = a['URI'] ?? ''; + if (method != 'AES-128' || (format != null && format != 'identity') || uri.isEmpty) { + throw const RelayPiecesException(drmText); + } + final v = a['IV']; + if (v != null && !_ivRe.hasMatch(v)) throw const RelayPiecesException('bad IV in the playlist'); + key = sharedPiece(const RelayPiece.key(), uri, null); + iv = v; + } + } + if (!ended) throw const RelayPiecesException(liveStreamText); + if (!pieces.any((p) => p.isSegment)) throw const RelayPiecesException('the playlist has no segments'); + if (pieces.length > RelayPieceList.maxPieces) throw const RelayPiecesException('the playlist is too long'); + return (list: RelayPieceList(List.unmodifiable(pieces), seq: seq), sources: List.unmodifiable(sources)); +} + +/// yt-dlp's DASH `fragments` (`http_dash_segments`) as pieces: a first +/// fragment without a duration is the init section, every other one a +/// segment (so no MPD parser of our own). +UpstreamPieces dashPieces(List<({Uri url, double? duration})> fragments) { + if (fragments.isEmpty) throw const RelayPiecesException('no fragments'); + final pieces = []; + final sources = []; + int? map; + for (var i = 0; i < fragments.length; i++) { + final f = fragments[i]; + if (i == 0 && f.duration == null) { + map = 0; + pieces.add(const RelayPiece.init()); + } else { + pieces.add(RelayPiece.segment(f.duration ?? 0, map: map)); + } + sources.add((url: f.url, offset: null, length: null)); + } + if (!pieces.any((p) => p.isSegment)) throw const RelayPiecesException('no fragments'); + return (list: RelayPieceList(List.unmodifiable(pieces)), sources: List.unmodifiable(sources)); +} + +/// Fetches an HLS media playlist with [headers] and parses it +/// ([parseHlsPlaylist]). +Future fetchHlsPlaylist(HttpClient client, Uri url, Map headers) async { + final req = await client.getUrl(url).timeout(const Duration(seconds: 20)); + headers.forEach(req.headers.set); + final res = await req.close().timeout(const Duration(seconds: 20)); + if (res.statusCode != 200) { + await res.drain().catchError((_) {}); + throw UpstreamHttpException(res.statusCode, url); + } + final text = await res.transform(utf8.decoder).join().timeout(const Duration(seconds: 30)); + // Relative entries resolve against where the playlist ended up. + final at = res.redirects.isEmpty ? url : url.resolveUri(res.redirects.last.location); + return parseHlsPlaylist(text, at); +} + +/// Reads one piece: a GET of its URL (with its byte range), whole. +Stream> readPiece(HttpClient client, PieceSource s, Map headers) async* { + final req = await client.getUrl(s.url).timeout(const Duration(seconds: 20)); + headers.forEach(req.headers.set); + req.headers.set(HttpHeaders.acceptEncodingHeader, 'identity'); + final off = s.offset, len = s.length; + if (len != null) req.headers.set(HttpHeaders.rangeHeader, 'bytes=${off ?? 0}-${(off ?? 0) + len - 1}'); + final res = await req.close().timeout(const Duration(seconds: 20)); + if (res.statusCode != 200 && res.statusCode != 206) { + await res.drain().catchError((_) {}); + throw UpstreamHttpException(res.statusCode, s.url); + } + if (len == null) { + yield* res; + return; + } + // A server ignoring Range sends the whole file: cut the piece out. + var skip = res.statusCode == 200 ? (off ?? 0) : 0; + var left = len; + await for (final c in res) { + var from = 0; + if (skip > 0) { + from = min(skip, c.length); + skip -= from; + } + final take = min(c.length - from, left); + if (take > 0) yield from == 0 && take == c.length ? c : c.sublist(from, from + take); + left -= take; + if (left <= 0) return; + } + if (left > 0) throw HttpException('upstream ended early', uri: s.url); +} diff --git a/packages/core/lib/src/yt_dlp.dart b/packages/core/lib/src/yt_dlp.dart index cc461a2..a7f3291 100644 --- a/packages/core/lib/src/yt_dlp.dart +++ b/packages/core/lib/src/yt_dlp.dart @@ -25,7 +25,8 @@ class YtDlpException implements Exception { } /// One playable format of a resolved page: a URL and the headers yt-dlp -/// says to send with it. +/// says to send with it. A segmented format ([isPieces]) is an HLS media +/// playlist at [url], or DASH [fragments]. class YtDlpFormat { final String id; final Uri url; @@ -35,8 +36,23 @@ class YtDlpFormat { /// yt-dlp's `downloader_options.http_chunk_size`: the longest GET the /// site serves at full speed. final int? chunkSize; + /// yt-dlp's `protocol`: `https`/`http` (a file), `m3u8_native`/`m3u8` + /// (an HLS media playlist), `http_dash_segments` ([fragments]). + final String protocol; + /// DASH fragments (a first one without a duration is the init section). + final List<({Uri url, double? duration})>? fragments; const YtDlpFormat( - {required this.id, required this.url, this.headers = const {}, this.height, this.ext = '', this.chunkSize}); + {required this.id, + required this.url, + this.headers = const {}, + this.height, + this.ext = '', + this.chunkSize, + this.protocol = 'https', + this.fragments}); + + /// Relayed as pieces (see relay_pieces.dart), not as a byte stream. + bool get isPieces => _piecesProtocols.contains(protocol); static YtDlpFormat _of(Map f) { final headers = f['http_headers']; @@ -50,10 +66,34 @@ class YtDlpFormat { height: h == 0 ? null : h, ext: '${f['ext']}', chunkSize: chunk is num && chunk > 0 ? chunk.toInt() : null, + protocol: '${f['protocol']}', + fragments: f['protocol'] == 'http_dash_segments' ? _fragments(f) : null, ); } + + /// yt-dlp's `fragments` (`url`, or `path` under `fragment_base_url`), + /// or null when one is not plain http(s). + static List<({Uri url, double? duration})>? _fragments(Map f) { + final raw = f['fragments']; + final base = Uri.tryParse('${f['fragment_base_url'] ?? f['url'] ?? ''}'); + if (raw is! List || raw.isEmpty) return null; + final out = <({Uri url, double? duration})>[]; + for (final fr in raw) { + if (fr is! Map) return null; + final u = fr['url'] is String + ? Uri.tryParse(fr['url'] as String) + : fr['path'] is String && base != null + ? base.resolve(fr['path'] as String) + : null; + if (u == null || (!u.isScheme('http') && !u.isScheme('https'))) return null; + out.add((url: u, duration: fr['duration'] is num ? (fr['duration'] as num).toDouble() : null)); + } + return out; + } } +const _piecesProtocols = {'m3u8_native', 'm3u8', 'http_dash_segments'}; + /// A format a relay plays as one of a media's tracks ([kind] `video` or /// `audio`; empty for a single file holding both). typedef YtDlpTrack = ({String kind, YtDlpFormat format}); @@ -69,11 +109,11 @@ class YtDlpInfo { final String? thumbnail; /// A live stream (never relayed). final bool isLive; - /// What a relay range-reads (see [pickTracks]): one file, or a video and - /// an audio file; empty when the page has neither (only streaming - /// playlists, or live). + /// What a relay reads (see [pickTracks]): one file, a video and an + /// audio file, or segmented playlists; empty when there is nothing (or + /// live). final List tracks; - /// Every plain http(s) format by id, to find [tracks] again after a + /// Every relayable format by id, to find [tracks] again after a /// re-resolve. final Map formats; const YtDlpInfo({ @@ -110,7 +150,7 @@ class YtDlpInfo { tracks: isLive ? const [] : pickTracks(formats), formats: { for (final f in formats) - if (_httpUrl(f) && f is Map) '${f['format_id'] ?? ''}': YtDlpFormat._of(f), + if ((_httpUrl(f) || _piecesUrl(f)) && f is Map) '${f['format_id'] ?? ''}': YtDlpFormat._of(f), }, ); } @@ -129,6 +169,16 @@ bool _httpUrl(Object? f) { return uri != null && (uri.isScheme('http') || uri.isScheme('https')); } +/// A segmented format (HLS playlist or DASH fragments) over http(s), not +/// DRM-protected (the playlist's keys are checked when it is read). +bool _piecesUrl(Object? f) { + if (f is! Map || f['has_drm'] == true || !_piecesProtocols.contains(f['protocol'])) return false; + final url = f['url']; + final uri = url is String ? Uri.tryParse(url) : null; + if (uri == null || (!uri.isScheme('http') && !uri.isScheme('https'))) return false; + return f['protocol'] != 'http_dash_segments' || YtDlpFormat._fragments(f) != null; +} + int _height(Map f) => f['height'] is num ? (f['height'] as num).toInt() : 0; double _tbr(Map f) => f['tbr'] is num ? (f['tbr'] as num).toDouble() : (f['abr'] is num ? (f['abr'] as num).toDouble() : 0); @@ -180,19 +230,22 @@ int _videoCodecRank(Map f) { /// shortest above), HEVC before AV1 before VP9 before h264 (smallest /// first, see [_videoCodecRank]), then bitrate. Audio: the original language, not a `-drc` variant, m4a/aac /// before others (opus), then bitrate. -(YtDlpFormat, YtDlpFormat)? pickSplit(List formats, {int maxHeight = 1080}) { - final video = [ - for (final f in formats) - if (_httpUrl(f) && (f as Map)['acodec'] == 'none' && f['vcodec'] is String && f['vcodec'] != 'none' && - _singleFileExts.contains(f['ext'])) - f, - ]; - final audio = [ - for (final f in formats) - if (_httpUrl(f) && (f as Map)['vcodec'] == 'none' && f['acodec'] is String && f['acodec'] != 'none' && - _audioExts.contains(f['ext'])) - f, - ]; +(YtDlpFormat, YtDlpFormat)? pickSplit(List formats, {int maxHeight = 1080}) => _pickPair( + [ + for (final f in formats) + if (_httpUrl(f) && (f as Map)['acodec'] == 'none' && f['vcodec'] is String && f['vcodec'] != 'none' && + _singleFileExts.contains(f['ext'])) + f, + ], + [ + for (final f in formats) + if (_httpUrl(f) && (f as Map)['vcodec'] == 'none' && f['acodec'] is String && f['acodec'] != 'none' && + _audioExts.contains(f['ext'])) + f, + ], + maxHeight); + +(YtDlpFormat, YtDlpFormat)? _pickPair(List video, List audio, int maxHeight) { if (video.isEmpty || audio.isEmpty) return null; video.sort((a, b) { final h = _byHeight(_height(a), _height(b), maxHeight); @@ -213,20 +266,44 @@ int _videoCodecRank(Map f) { return (YtDlpFormat._of(video.first), YtDlpFormat._of(audio.first)); } +/// The pair if its video is taller than [single] (by the [maxHeight] rule). +bool _pairWins((YtDlpFormat, YtDlpFormat)? pair, YtDlpFormat? single, int maxHeight) => + pair != null && + (single == null || + (single.height != null && + pair.$1.height != null && + _byHeight(pair.$1.height!, single.height!, maxHeight) < 0)); + +/// Segmented formats (HLS / DASH): one muxed playlist (video and audio +/// not known to be absent) unless a video-only + audio-only pair is +/// taller; the same height, codec and audio rules as files. +List pickPieces(List formats, {int maxHeight = 1080}) { + final ok = [for (final f in formats) if (_piecesUrl(f)) f as Map]; + final muxed = [for (final f in ok) if (f['vcodec'] != 'none' && f['acodec'] != 'none') f]..sort((a, b) { + final h = _byHeight(_height(a), _height(b), maxHeight); + return h != 0 ? h : _tbr(b).compareTo(_tbr(a)); + }); + final pair = _pickPair([for (final f in ok) if (f['acodec'] == 'none' && f['vcodec'] != 'none') f], + [for (final f in ok) if (f['vcodec'] == 'none' && f['acodec'] != 'none') f], maxHeight); + final single = muxed.isEmpty ? null : YtDlpFormat._of(muxed.first); + if (_pairWins(pair, single, maxHeight)) { + return [(kind: 'video', format: pair!.$1), (kind: 'audio', format: pair.$2)]; + } + return single == null ? const [] : [(kind: 'video', format: single)]; +} + /// What a relay reads of [formats]: the single file ([pickSingleFile]) /// unless a video + audio pair ([pickSplit]) is taller (both by the -/// [maxHeight] rule); empty when there is neither. +/// [maxHeight] rule); with neither, segmented playlists ([pickPieces]); +/// empty when there is nothing. List pickTracks(List formats, {int maxHeight = 1080}) { final single = pickSingleFile(formats, maxHeight: maxHeight); final split = pickSplit(formats, maxHeight: maxHeight); - if (split != null && - (single == null || - (single.height != null && - split.$1.height != null && - _byHeight(split.$1.height!, single.height!, maxHeight) < 0))) { - return [(kind: 'video', format: split.$1), (kind: 'audio', format: split.$2)]; + if (_pairWins(split, single, maxHeight)) { + return [(kind: 'video', format: split!.$1), (kind: 'audio', format: split.$2)]; } - return single == null ? const [] : [(kind: '', format: single)]; + if (single != null) return [(kind: '', format: single)]; + return pickPieces(formats, maxHeight: maxHeight); } /// The official standalone build for this platform, or null if there is diff --git a/packages/core/test/relay_pieces_test.dart b/packages/core/test/relay_pieces_test.dart new file mode 100644 index 0000000..f1b2c68 --- /dev/null +++ b/packages/core/test/relay_pieces_test.dart @@ -0,0 +1,130 @@ +// Pieces tracks: HLS playlist parsing (byte ranges folded, init and key +// pieces, refusals), the synthesized playlist, and the track wire. +import 'package:sync_relay_core/core.dart'; +import 'package:test/test.dart'; + +void main() { + final base = Uri.parse('https://cdn.example/v/index.m3u8'); + + test('byte ranges fold into pieces; MAP, AES-128 key and sequence carry over to the served playlist', () { + final p = parseHlsPlaylist(''' +#EXTM3U +#EXT-X-VERSION:7 +#EXT-X-TARGETDURATION:6 +#EXT-X-MEDIA-SEQUENCE:5 +#EXT-X-MAP:URI="main.mp4",BYTERANGE="700@0" +#EXT-X-KEY:METHOD=AES-128,URI="https://keys.example/k1",IV=0x00000000000000000000000000000001 +#EXTINF:6.0, +#EXT-X-BYTERANGE:1000@700 +main.mp4 +#EXTINF:5.5, +#EXT-X-BYTERANGE:2000 +main.mp4 +#EXT-X-KEY:METHOD=NONE +#EXT-X-DISCONTINUITY +#EXTINF:4, +seg/3.ts?tok=a +#EXT-X-ENDLIST +''', base); + expect([for (final s in p.sources) (s.url.toString(), s.offset, s.length)], [ + ('https://cdn.example/v/main.mp4', 0, 700), + ('https://keys.example/k1', null, null), + ('https://cdn.example/v/main.mp4', 700, 1000), + ('https://cdn.example/v/main.mp4', 1700, 2000), // offset: end of the previous range + ('https://cdn.example/v/seg/3.ts?tok=a', null, null), + ]); + // Through the wire and back: same shape, same playlist. + final list = RelayPieceList.fromJson(p.list.toJson())!; + expect(list.sameShape(p.list), isTrue); + expect(list.playlist(), ''' +#EXTM3U +#EXT-X-VERSION:6 +#EXT-X-TARGETDURATION:6 +#EXT-X-MEDIA-SEQUENCE:5 +#EXT-X-PLAYLIST-TYPE:VOD +#EXT-X-MAP:URI="p/0" +#EXT-X-KEY:METHOD=AES-128,URI="p/1",IV=0x00000000000000000000000000000001 +#EXTINF:6.000000, +p/2 +#EXTINF:5.500000, +p/3 +#EXT-X-DISCONTINUITY +#EXT-X-KEY:METHOD=NONE +#EXTINF:4.000000, +p/4 +#EXT-X-ENDLIST +'''); + }); + + test('refused: DRM, live, master; a malformed piece list', () { + String pl(String body, {bool end = true}) => '#EXTM3U\n$body\n#EXTINF:4,\na.ts\n${end ? '#EXT-X-ENDLIST' : ''}'; + Matcher refused(String text) => + throwsA(isA().having((e) => e.message, 'message', text)); + expect(() => parseHlsPlaylist(pl('#EXT-X-KEY:METHOD=SAMPLE-AES,URI="skd://x"'), base), refused(drmText)); + expect(() => parseHlsPlaylist(pl('#EXT-X-KEY:METHOD=AES-128,URI="skd://x"'), base), refused(drmText)); + expect( + () => parseHlsPlaylist( + pl('#EXT-X-KEY:METHOD=AES-128,URI="k",KEYFORMAT="urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed"'), base), + refused(drmText)); + expect(() => parseHlsPlaylist(pl(''), base).list, returnsNormally); + expect(() => parseHlsPlaylist(pl('', end: false), base), refused(liveStreamText)); + expect(() => parseHlsPlaylist('#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1\nv.m3u8\n', base), + throwsA(isA())); + expect( + RelayPieceList.fromJson({ + 'pieces': [ + {'d': 4.0, 'map': 0}, + ], + }), + isNull, + reason: 'map must name an init piece'); + expect( + RelayPieceList.fromJson({ + 'pieces': [ + {'d': 4.0, 'iv': '0x1"\n#EXT'}, + ], + }), + isNull); + }); + + test('track wire: bytes entries as before, pieces with "as"; one pieces track is enough', () { + final two = MediaRelay.fromJson({ + 'source': 'm2', + 'tracks': [ + {'kind': 'video', 'as': 'pieces'}, + {'kind': 'audio', 'as': 'pieces'}, + ], + }); + expect(two.tracks, const [RelayTrackSpec('video', pieces: true), RelayTrackSpec('audio', pieces: true)]); + expect(two.toJson()['tracks'], [ + {'kind': 'video', 'as': 'pieces'}, + {'kind': 'audio', 'as': 'pieces'}, + ]); + final one = MediaRelay.fromJson({ + 'tracks': [ + {'kind': 'video', 'as': 'pieces'}, + ], + }); + expect(one.tracks.single.pieces, isTrue); + // One byte track is one file; an unknown "as" drops the list. + expect( + MediaRelay.fromJson({ + 'tracks': [ + {'kind': 'video'}, + ], + }).tracks, + isEmpty); + expect( + MediaRelay.fromJson({ + 'tracks': [ + {'kind': 'video', 'as': 'frames'}, + {'kind': 'audio'}, + ], + }).tracks, + isEmpty); + expect(const MediaRelay(tracks: [RelayTrackSpec('video'), RelayTrackSpec('audio')]).toJson()['tracks'], [ + {'kind': 'video'}, + {'kind': 'audio'}, + ]); + }); +} diff --git a/packages/core/test/yt_dlp_test.dart b/packages/core/test/yt_dlp_test.dart index 2a88d31..1c198fb 100644 --- a/packages/core/test/yt_dlp_test.dart +++ b/packages/core/test/yt_dlp_test.dart @@ -101,7 +101,8 @@ void main() { final single = _f('22', 'https', 'avc1', 'mp4a', 720); expect(pickTracks([single, ...youtube]).map((x) => x.kind), ['video', 'audio']); expect(pickTracks([_f('18', 'https', 'avc1', 'mp4a', 1080), ...youtube]).single.format.id, '18'); - expect(pickTracks([_f('96', 'm3u8_native', 'avc1', 'mp4a', 1080)]), isEmpty); + expect(pickTracks([_f('96', 'm3u8_native', 'avc1', 'mp4a', 1080)]).single.format.isPieces, isTrue); + expect(pickTracks([_f('96', 'rtmp', 'avc1', 'mp4a', 1080)]), isEmpty); final live = YtDlpInfo.parse({'title': 'x', 'is_live': true, 'formats': youtube}); expect(live.isLive, isTrue); expect(live.tracks, isEmpty); @@ -117,8 +118,10 @@ void main() { test('viewer URL: one track plain, several an mpv EDL with %length% framing', () { expect(relayViewerUrl('http://h:1/r/default/t/3', 'a/b c'), 'http://h:1/r/default/t/3/a_b%20c'); const v = 'http://h:1/r/default/t/3/0/x', a = 'http://h:1/r/default/t/3/1/x'; - expect(relayViewerUrl('http://h:1/r/default/t/3', 'x', tracks: 2), + expect(relayViewerUrl('http://h:1/r/default/t/3', 'x', tracks: const [RelayTrackSpec('video'), RelayTrackSpec('audio')]), 'edl://!no_clip;%${v.length}%$v;!new_stream;!no_clip;%${a.length}%$a'); + expect(relayViewerUrl('http://h:1/r/default/t/3', 'x', tracks: const [RelayTrackSpec('video', pieces: true)]), + 'http://h:1/r/default/t/3/0/x.m3u8'); expect(mpvEdl(['http://x/a;b,c']), 'edl://!no_clip;%14%http://x/a;b,c'); }); @@ -191,7 +194,7 @@ void main() { expect(probe.title, startsWith('a girl.')); expect(probe.relayProblem, isNull); final media = LinkRelayMedia(url, ytDlp: y, probe: probe); - final bytes = (await media.open()).single.bytes; + final bytes = (await media.open()).single.bytes!; final size = await bytes.size(); var n = 0; await for (final c in bytes.read(0, size - 1)) { @@ -201,6 +204,33 @@ void main() { expect(n, size); }, timeout: const Timeout(Duration(minutes: 3)), skip: Platform.environment['YTDLP_LIVE'] == '1' ? false : 'YTDLP_LIVE not set'); + test('pieces only without a file: the tweet\'s HLS video + audio; DASH fragments; muxed HLS', () { + final tweet = [ + _f('hls-audio-128000-Audio', 'm3u8_native', 'none', null, null, tbr: 128), + _f('hls-430', 'm3u8_native', 'avc1.64001F', 'none', 598, tbr: 430), + ]; + expect([for (final t in pickTracks(tweet)) (t.kind, t.format.id, t.format.isPieces)], + [('video', 'hls-430', true), ('audio', 'hls-audio-128000-Audio', true)]); + // A byte file (even a shorter one) comes first. + expect(pickTracks([...tweet, _f('http-832', 'https', null, null, 360)]).single.format.id, 'http-832'); + final dash = _f('v', 'http_dash_segments', 'avc1', 'none', 1080, extra: { + 'fragment_base_url': 'https://cdn.example/b/', + 'fragments': [ + {'path': 'init.mp4'}, + {'path': 's1.m4s', 'duration': 4.0}, + {'url': 'https://other.example/s2.m4s', 'duration': 2.5}, + ], + }); + final t = pickTracks([dash, _f('a', 'http_dash_segments', 'none', 'mp4a', null, extra: {'fragments': [{'url': 'https://cdn.example/a', 'duration': 6.5}]})]); + final p = dashPieces(t.first.format.fragments!); + expect(p.sources.map((s) => '${s.url}'), ['https://cdn.example/b/init.mp4', 'https://cdn.example/b/s1.m4s', 'https://other.example/s2.m4s']); + expect((p.list.pieces.first.role, p.list.pieces[2].map, p.list.duration), (RelayPieceRole.init, 0, 6.5)); + // A muxed playlist wins unless the pair is taller. + final muxed = _f('m', 'm3u8_native', 'avc1', 'mp4a', 720); + expect(pickTracks([muxed, ...tweet]).single.format.id, 'm'); + expect(pickTracks([muxed, ...tweet]).single.kind, 'video'); + }); + test('codec order: HEVC, then AV1, then VP9, then h264 at the same height', () { final audio = _f('140', 'https', 'none', 'mp4a.40.2', null); String pick(List codecs) => pickSplit([ diff --git a/packages/server/lib/src/piece_buffer.dart b/packages/server/lib/src/piece_buffer.dart new file mode 100644 index 0000000..6761ce6 --- /dev/null +++ b/packages/server/lib/src/piece_buffer.dart @@ -0,0 +1,368 @@ +import 'dart:async'; +import 'dart:math'; +import 'dart:typed_data'; + +import 'package:sync_relay_core/core.dart'; + +import 'relay_buffer.dart'; + +final _log = createLogger('relay'); + +class _PiecePull { + final int id, piece; + final buf = BytesBuilder(copy: false); + _PiecePull(this.id, this.piece); +} + +class _Viewer { + int pos = -1; // its newest requested segment + int? waiting; // that segment, while it is missing + int waitSince = 0; + int open = 0; + DateTime lastActive = DateTime.now(); +} + +/// The server side of a pieces track (see relay_pieces.dart): pieces by +/// index in memory, each filled by one whole pull, served whole (a Range +/// inside a piece is the route's job). +/// +/// - Fill order: a viewer blocked on a piece (oldest wait first), the init +/// and key pieces the viewers' segments use, then segments ahead of each +/// viewer, slowest first, up to [ahead] bytes. +/// - A viewer's position is its newest requested segment; a superseded +/// request still gets its piece but no longer drives the feed, and a pull +/// nobody waits on gives way to a blocked viewer at once. +/// - Kept: every init and key piece, plus segments by distance from the +/// viewers ([ahead] + [back] bytes; behind a viewer costs 4x), never LRU. +/// Sizes before a piece arrives are estimated from the bytes per second +/// of the ones that did, else [rate]. +class PieceBuffer implements TrackBuffer { + @override + int ahead, back; + final Duration waitLimit; + /// Bytes per second assumed before any segment has arrived. + double rate; + final int Function() _newId; + PieceBuffer({ + this.ahead = RelayConstants.relayAheadBytes, + this.back = RelayConstants.relayBackBytes, + this.waitLimit = const Duration(seconds: RelayConstants.relayWaitSecs), + this.rate = 500000, + int Function()? nextId, + }) : _newId = nextId ?? _counter(); + + static int Function() _counter() { + var n = 0; + return () => ++n; + } + + RelayFeed? _feed; + /// The track's shape, once the feed has connected. + RelayPieceList? list; + String? _playlist; + final _pieces = {}; + final _sizes = {}; // known sizes, kept after eviction + var _knownSecs = 0.0; + var _knownBytes = 0; + final _waiters = >>{}; + final _aux = {}; // init/key piece a request waits on -> since (us) + final _viewers = {}; + _PiecePull? _pull; + DateTime? _retryAt; + var _disposed = false; + Completer _kick = Completer(); + final Completer _ready = Completer(); + + @override + int bytesFromFeed = 0, bytesToViewers = 0; + + @override + bool get live => _feed != null && list != null; + + @override + int get cachedBytes => _pieces.values.fold(0, (a, p) => a + p.length); + + /// The media playlist viewers open (see [RelayPieceList.playlist]). + String? get playlist => _playlist ??= list?.playlist(); + + /// Attaches a (re)connected feed for a track of [list]'s shape. A + /// different shape means different content: the buffer starts over. + void attach(RelayFeed feed, RelayPieceList list) { + if (_disposed) return; + final old = this.list; + if (old != null && !old.sameShape(list)) { + _log.info('relay source changed pieces ${old.length} -> ${list.length}, dropping buffer'); + _pieces.clear(); + _sizes.clear(); + _knownSecs = 0; + _knownBytes = 0; + _viewers.clear(); + _aux.clear(); + } + this.list = list; + _playlist = null; + _feed = feed; + _pull = null; + if (!_ready.isCompleted) _ready.complete(); + if (_fillerRunning) { + _wake(); + } else { + unawaited(_filler()); + } + } + + @override + void detach(RelayFeed feed) { + if (!identical(_feed, feed)) return; + _feed = null; + _pull = null; + _wake(); + } + + @override + void dispose() { + _disposed = true; + _feed = null; + _pull = null; + _pieces.clear(); + for (final list in _waiters.values) { + for (final c in list) { + if (!c.isCompleted) c.complete(); + } + } + _waiters.clear(); + if (!_ready.isCompleted) _ready.complete(); + _wake(); + } + + void _wake() { + if (!_kick.isCompleted) _kick.complete(); + } + + // ------------------------------------------------------------ feed input + + @override + void feedBytes(int id, Uint8List data) { + final pl = _pull; + if (pl == null || pl.id != id || _disposed) return; + _feed?.arrived(id, data.length); + bytesFromFeed += data.length; + pl.buf.add(data); + } + + @override + void feedEnd(int id) { + final pl = _pull; + if (pl == null || pl.id != id || _disposed) return; + _pull = null; + final data = pl.buf.takeBytes(); + final i = pl.piece; + _pieces[i] = data; + final p = list!.pieces[i]; + if (p.isSegment && !_sizes.containsKey(i)) { + _knownSecs += p.d; + _knownBytes += data.length; + } + _sizes[i] = data.length; + for (final c in _waiters.remove(i) ?? const >[]) { + c.complete(); + } + _wake(); + } + + @override + void feedFail(int id, String reason) { + if (_pull?.id != id) return; + _log.warning('relay piece pull $id failed: $reason'); + _pull = null; + _retryAt = DateTime.now().add(const Duration(seconds: 1)); + _wake(); + } + + // ------------------------------------------------------------- scheduling + + double _est(int i) { + final known = _sizes[i]; + if (known != null) return known.toDouble(); + final p = list!.pieces[i]; + if (!p.isSegment) return 0; + return (_knownSecs > 0 ? _knownBytes / _knownSecs : rate) * p.d; + } + + Iterable<_Viewer> get _active { + final now = DateTime.now(); + return _viewers.values.where( + (v) => v.pos >= 0 && (v.open > 0 || now.difference(v.lastActive) < const Duration(seconds: 60))); + } + + /// Estimated byte offset of every piece (prefix sums), for distances. + List _offsets() { + final l = list!; + final out = List.filled(l.length + 1, 0); + for (var i = 0; i < l.length; i++) { + out[i + 1] = out[i] + _est(i); + } + return out; + } + + double _distance(int i, List off, List<_Viewer> active) { + var d = double.infinity; + for (final v in active) { + d = min(d, i >= v.pos ? off[i] - off[v.pos] : (off[v.pos] - off[i]) * 4); + } + return d; + } + + bool _waited(int i) => _aux.containsKey(i) || _viewers.values.any((v) => v.waiting == i); + + /// Cached segments past the budget, farthest first; returns the + /// distance of the farthest segment kept (a fill farther than it would + /// only be evicted again). + double _evict(List off, List<_Viewer> active) { + final l = list!; + final cached = [for (final i in _pieces.keys) if (l.pieces[i].isSegment) i]; + var total = cached.fold(0, (a, i) => a + _pieces[i]!.length); + cached.sort((a, b) => _distance(b, off, active).compareTo(_distance(a, off, active))); + final limit = ahead + back; + var k = 0; + for (; k < cached.length && total > limit; k++) { + final i = cached[k]; + if (_waited(i)) continue; + total -= _pieces.remove(i)!.length; + } + _cachedSegBytes = total; + return k < cached.length ? _distance(cached[k], off, active) : double.infinity; + } + + var _cachedSegBytes = 0; + + int? _nextMissing() { + final l = list!; + final active = _active.toList()..sort((a, b) => a.pos.compareTo(b.pos)); + final off = _offsets(); + final keep = _evict(off, active); + final blocked = <(int, int)>[ + for (final v in _viewers.values) + if (v.waiting != null && !_pieces.containsKey(v.waiting)) (v.waiting!, v.waitSince), + for (final e in _aux.entries) + if (!_pieces.containsKey(e.key)) (e.key, e.value), + ]..sort((a, b) => a.$2.compareTo(b.$2)); + if (blocked.isNotEmpty) return blocked.first.$1; + if (active.isEmpty) return null; // nobody watches: fetch nothing + for (final v in active) { + var budget = 0.0; + for (var i = v.pos; i < l.length; i++) { + final p = l.pieces[i]; + if (!p.isSegment) continue; + for (final dep in [p.map, p.key]) { + if (dep != null && !_pieces.containsKey(dep)) return dep; + } + if (!_pieces.containsKey(i)) { + final room = _cachedSegBytes + _est(i) <= ahead + back; + if (i == v.pos || room || _distance(i, off, active) < keep) return i; + break; + } + budget += _est(i); + if (budget > ahead) break; + } + } + return null; + } + + var _fillerRunning = false; + + Future _filler() async { + _fillerRunning = true; + try { + while (!_disposed) { + final feed = _feed; + final retry = _retryAt; + if (feed == null || _pull != null || list == null) { + await _sleep(); + continue; + } + if (retry != null && DateTime.now().isBefore(retry)) { + await _sleep(retry.difference(DateTime.now())); + continue; + } + _retryAt = null; + final i = _nextMissing(); + if (i == null) { + await _sleep(); + continue; + } + final pl = _pull = _PiecePull(_newId(), i); + feed.pullPiece(pl.id, i); + } + } finally { + _fillerRunning = false; + } + } + + Future _sleep([Duration limit = const Duration(seconds: 5)]) async { + _kick = Completer(); + // Also wake periodically: viewers age out of [_active]. + await _kick.future.timeout(limit, onTimeout: () {}); + } + + // ---------------------------------------------------------------- viewers + + @override + Future ready() async { + if (list != null) return true; + await _ready.future.timeout(waitLimit, onTimeout: () {}); + return list != null && !_disposed; + } + + /// [viewer] opened the playlist: a new viewer starts at the first + /// segment, so its first pieces are fetched before it asks. + void opened(String viewer) { + final l = list; + if (l == null) return; + final v = _viewers.putIfAbsent(viewer, _Viewer.new); + if (v.pos < 0) v.pos = l.pieces.indexWhere((p) => p.isSegment); + v.lastActive = DateTime.now(); + _wake(); + } + + /// Piece [index] for [viewer], once it is in; null when it does not + /// arrive within [waitLimit] (or the buffer is gone). + Future piece(String viewer, int index) async { + final l = list; + if (l == null || index < 0 || index >= l.length) return null; + final v = _viewers.putIfAbsent(viewer, _Viewer.new); + final segment = l.pieces[index].isSegment; + final now = DateTime.now().microsecondsSinceEpoch; + v.open++; + v.lastActive = DateTime.now(); + if (segment) { + v.pos = index; + v.waiting = _pieces.containsKey(index) ? null : index; + v.waitSince = now; + } else if (!_pieces.containsKey(index)) { + _aux.putIfAbsent(index, () => now); + } + // A pull nobody waits on gives way to this one at once. + final pl = _pull; + if (!_pieces.containsKey(index) && pl != null && pl.piece != index && !_waited(pl.piece)) { + _pull = null; + _feed?.cancel(pl.id); + } + _wake(); + try { + final deadline = DateTime.now().add(waitLimit); + while (!_pieces.containsKey(index)) { + if (_disposed || DateTime.now().isAfter(deadline)) return null; + final c = Completer(); + _waiters.putIfAbsent(index, () => []).add(c); + await c.future.timeout(const Duration(seconds: 2), onTimeout: () {}); + } + return _pieces[index]; + } finally { + v.open--; + v.lastActive = DateTime.now(); + if (segment && v.waiting == index) v.waiting = null; + if (!segment) _aux.remove(index); + } + } +} diff --git a/packages/server/lib/src/relay_buffer.dart b/packages/server/lib/src/relay_buffer.dart index 2785d92..77b4827 100644 --- a/packages/server/lib/src/relay_buffer.dart +++ b/packages/server/lib/src/relay_buffer.dart @@ -7,28 +7,54 @@ import 'package:sync_relay_core/core.dart'; final _log = createLogger('relay'); -/// What feeds a [RelayBuffer]: a member's relay channel, or the server +/// One track's buffer on the server: a byte stream ([RelayBuffer]) or +/// pieces ([PieceBuffer]). +abstract interface class TrackBuffer { + /// Budgets, set by the relay when a media's tracks share them. + abstract int ahead, back; + bool get live; + int get cachedBytes; + int get bytesFromFeed; + int get bytesToViewers; + /// Waits for the feed's first connect, up to the wait limit. + Future ready(); + void detach(RelayFeed feed); + void dispose(); + void feedBytes(int id, Uint8List data); + void feedEnd(int id); + void feedFail(int id, String reason); +} + +/// What feeds a [TrackBuffer]: a member's relay channel, or the server /// itself fetching a link ([LocalRelayFeed]). abstract class RelayFeed { - /// Ask for bytes [start]..[end]; they arrive via [RelayBuffer.feedBytes] - /// in order, then [RelayBuffer.feedEnd]. + /// Ask for bytes [start]..[end]; they arrive via [TrackBuffer.feedBytes] + /// in order, then [TrackBuffer.feedEnd]. void pull(int id, int start, int end); + + /// Ask for piece [index] of a pieces track, whole (arrives like [pull]). + void pullPiece(int id, int index); void cancel(int id); /// Bytes of [id] reached the buffer (the remote feed acks them). void arrived(int id, int n) {} } -/// Fetches from [bytes] in this process (the "server fetches it" mode). +/// Fetches [track] in this process (the "server fetches it" mode). class LocalRelayFeed extends RelayFeed { - final RelayBytes bytes; - final RelayBuffer buffer; + final RelayTrack track; + final TrackBuffer buffer; final _subs = >>{}; - LocalRelayFeed(this.bytes, this.buffer); + LocalRelayFeed(this.track, this.buffer); @override - void pull(int id, int start, int end) { - _subs[id] = bytes.read(start, end).listen( + void pull(int id, int start, int end) => _listen(id, track.bytes!.read(start, end)); + + @override + void pullPiece(int id, int index) => _listen(id, track.pieces!.read(index)); + + void _listen(int id, Stream> data) { + _subs[id] = data.listen( (c) => buffer.feedBytes(id, c is Uint8List ? c : Uint8List.fromList(c)), onError: (Object e) { unawaited(_subs.remove(id)?.cancel().catchError((_) {})); @@ -51,7 +77,8 @@ class LocalRelayFeed extends RelayFeed { await s.cancel().catchError((_) {}); } _subs.clear(); - await bytes.close(); + await track.bytes?.close(); + await track.pieces?.close(); } } @@ -66,6 +93,10 @@ class ChannelRelayFeed extends RelayFeed { void pull(int id, int start, int end) => send(jsonEncode({'t': 'pull', 'id': id, if (track != null) 'track': track, 'start': start, 'end': end})); + @override + void pullPiece(int id, int index) => + send(jsonEncode({'t': 'pull', 'id': id, if (track != null) 'track': track, 'piece': index})); + @override void cancel(int id) => send(jsonEncode({'t': 'cancel', 'id': id})); @@ -102,8 +133,9 @@ class _Viewer { /// - Only a viewer's newest request moves its position or asks the feed /// for anything; older ones still get bytes once they are buffered. /// Everything lives in memory: dropping the buffer is the cleanup. -class RelayBuffer { +class RelayBuffer implements TrackBuffer { /// Set by the relay when a media's tracks share the budget. + @override int ahead, back; final int pin; final Duration waitLimit; @@ -137,9 +169,12 @@ class RelayBuffer { final Completer _ready = Completer(); /// Totals, for logs and tests. + @override int bytesFromFeed = 0, bytesToViewers = 0; + @override bool get live => _feed != null && size > 0; + @override int get cachedBytes => _slices.values.fold(0, (a, s) => a + s.length); int get _nSlices => (size + slice - 1) ~/ slice; int _sliceEnd(int s) => min((s + 1) * slice, size) - 1; @@ -163,6 +198,7 @@ class RelayBuffer { } /// The feed went away; buffered slices keep serving. + @override void detach(RelayFeed feed) { if (!identical(_feed, feed)) return; _feed = null; @@ -170,6 +206,7 @@ class RelayBuffer { _wake(); } + @override void dispose() { _disposed = true; _feed = null; @@ -191,6 +228,7 @@ class RelayBuffer { // ------------------------------------------------------------ feed input + @override void feedBytes(int id, Uint8List data) { final pl = _pull; if (pl == null || pl.id != id || _disposed) return; @@ -221,6 +259,7 @@ class RelayBuffer { } } + @override void feedEnd(int id) { final pl = _pull; if (pl != null && pl.id == id) { @@ -229,6 +268,7 @@ class RelayBuffer { } } + @override void feedFail(int id, String reason) { if (_pull?.id != id) return; _log.warning('relay pull $id failed: $reason'); @@ -376,6 +416,7 @@ class RelayBuffer { // ---------------------------------------------------------------- viewers /// Waits for the size (the feed's first connect), up to [waitLimit]. + @override Future ready() async { if (size > 0) return true; await _ready.future.timeout(waitLimit, onTimeout: () {}); diff --git a/packages/server/lib/src/relay_routes.dart b/packages/server/lib/src/relay_routes.dart index 31c36c4..941729d 100644 --- a/packages/server/lib/src/relay_routes.dart +++ b/packages/server/lib/src/relay_routes.dart @@ -8,6 +8,7 @@ import 'package:sync_relay_core/core.dart'; import 'package:web_socket/web_socket.dart'; import 'media_endpoint.dart'; +import 'piece_buffer.dart'; import 'relay_buffer.dart'; import 'session.dart'; @@ -25,6 +26,15 @@ List<(int, int)> relayBudgets(List sizes, return [for (final n in sizes) (share(ahead, n), share(back, n))]; } +/// Bytes per second a pieces track is assumed to need before its first +/// pieces arrive (its share of the budget, its distances): 4 Mbit/s for +/// video, 160 kbit/s for audio. +double _piecesRate(String kind) => kind == 'audio' ? 20000 : 500000; + +/// One track as the source (or the fetcher) described it: a byte size, +/// or a piece list. +typedef _TrackInfo = ({int? size, RelayPieceList? pieces}); + /// The relay of the current media: one buffer per track and where the /// bytes come from. One per relayed swap; [dispose] on the next swap or /// server stop (the buffers are memory only, so that is the whole cleanup). @@ -35,11 +45,11 @@ class ActiveRelay { /// server fetches the link itself. final String? sourceToken; final void Function() onLiveChanged; - /// One per track, in track order; empty until the server has read a - /// fetched link. - List buffers = const []; - /// `media.relay.tracks`: the kinds when there are several tracks. - List kinds = const []; + /// One per track, in track order ([RelayBuffer] for bytes, [PieceBuffer] + /// for pieces); empty until the server has read a fetched link. + List buffers = const []; + /// `media.relay.tracks`: empty for one file. + List tracks = const []; final _known = Completer(); var _ids = 0; RelayMedia? _media; @@ -51,12 +61,43 @@ class ActiveRelay { int _nextId() => ++_ids; - void _setTracks(List kinds) { - this.kinds = kinds.length > 1 ? List.unmodifiable(kinds) : const []; - buffers = [for (var i = 0; i < max(1, kinds.length); i++) RelayBuffer(nextId: _nextId)]; + void _setTracks(List tracks) { + this.tracks = MediaRelay.validTracks(tracks) ? List.unmodifiable(tracks) : const []; + buffers = this.tracks.isEmpty + ? [RelayBuffer(nextId: _nextId)] + : [ + for (final t in this.tracks) + t.pieces ? PieceBuffer(nextId: _nextId, rate: _piecesRate(t.kind)) : RelayBuffer(nextId: _nextId), + ]; if (!_known.isCompleted) _known.complete(); } + /// The info of every track matches [tracks] (bytes for a byte track, + /// pieces for a pieces one). + bool _matches(List<_TrackInfo> infos) { + if (infos.length != buffers.length) return false; + for (var i = 0; i < infos.length; i++) { + if ((infos[i].pieces != null) != (buffers[i] is PieceBuffer)) return false; + } + return true; + } + + /// Attaches every buffer to its feed. + void _attachAll(List feeds, List<_TrackInfo> infos) { + _budget([ + for (var i = 0; i < infos.length; i++) + infos[i].size ?? (_piecesRate(tracks[i].kind) * infos[i].pieces!.duration).round(), + ]); + for (var i = 0; i < buffers.length; i++) { + final b = buffers[i]; + if (b is PieceBuffer) { + b.attach(feeds[i], infos[i].pieces!); + } else if (b is RelayBuffer) { + b.attach(feeds[i], infos[i].size!); + } + } + } + void _budget(List sizes) { final b = relayBudgets(sizes); for (var i = 0; i < buffers.length; i++) { @@ -69,12 +110,12 @@ class ActiveRelay { /// Bytes from member [sourceToken] over its relay channel, as the /// [tracks] its swap named (empty: one file). factory ActiveRelay.fromMember(int mediaId, String name, String sourceToken, - {List tracks = const [], required void Function() onLiveChanged}) => + {List tracks = const [], required void Function() onLiveChanged}) => ActiveRelay._(mediaId, name, sourceToken, onLiveChanged).._setTracks(tracks); /// The server reads [url] itself: a media file as is, a web page - /// through [ytDlp] (one file, or video + audio tracks). [onFail] gets - /// the reason when it cannot. + /// through [ytDlp] (one file, video + audio tracks, or segmented + /// playlists). [onFail] gets the reason when it cannot. static ActiveRelay fetch(int mediaId, String name, Uri url, {YtDlp? ytDlp, required void Function() onLiveChanged, void Function(String reason)? onFail}) { final r = ActiveRelay._(mediaId, name, null, onLiveChanged); @@ -82,14 +123,14 @@ class ActiveRelay { unawaited(() async { try { final tracks = await media.open(); - final sizes = [for (final t in tracks) await t.bytes.size()]; + final infos = <_TrackInfo>[ + for (final t in tracks) + t.pieces != null ? (size: null, pieces: await t.pieces!.list()) : (size: await t.bytes!.size(), pieces: null), + ]; if (r._disposed) return; - r._setTracks([for (final t in tracks) t.kind]); - r._budget(sizes); - r._local = [for (var i = 0; i < tracks.length; i++) LocalRelayFeed(tracks[i].bytes, r.buffers[i])]; - for (var i = 0; i < tracks.length; i++) { - r.buffers[i].attach(r._local[i], sizes[i]); - } + r._setTracks(relayTrackSpecs(tracks)); + r._local = [for (var i = 0; i < tracks.length; i++) LocalRelayFeed(tracks[i], r.buffers[i])]; + r._attachAll(r._local, infos); onLiveChanged(); } catch (e) { _log.warning('relay fetch: cannot read the link: $e'); @@ -113,10 +154,10 @@ class ActiveRelay { return known; } - /// The buffer of [track]; null (the single-file URL) only for a media of - /// one track. Null when there is no such track. - RelayBuffer? bufferAt(int? track) { - if (track == null) return buffers.length == 1 ? buffers.single : null; + /// The buffer of [track]; null (the single-file URL) only for one file. + /// Null when there is no such track. + TrackBuffer? bufferAt(int? track) { + if (track == null) return tracks.isEmpty && buffers.length == 1 ? buffers.single : null; return track >= 0 && track < buffers.length ? buffers[track] : null; } @@ -139,18 +180,15 @@ class ActiveRelay { if (ch != null) unawaited(_closeNotSource(ch)); } - void _attachChannel(RelicWebSocket ws, List sizes) { + void _attachChannel(RelicWebSocket ws, List<_TrackInfo> infos) { final old = _channel; if (old != null && !identical(old, ws)) unawaited(old.tryClose()); _channel = ws; - final multi = buffers.length > 1; + final named = tracks.isNotEmpty; final feeds = _channelFeeds = [ - for (var i = 0; i < buffers.length; i++) ChannelRelayFeed((t) => ws.trySendText(t), track: multi ? i : null), + for (var i = 0; i < buffers.length; i++) ChannelRelayFeed((t) => ws.trySendText(t), track: named ? i : null), ]; - _budget(sizes); - for (var i = 0; i < buffers.length; i++) { - buffers[i].attach(feeds[i], sizes[i]); - } + _attachAll(feeds, infos); onLiveChanged(); } @@ -166,17 +204,25 @@ class ActiveRelay { } } -/// The track sizes of a channel's `info` frame (`size`, or `tracks[].size`), -/// or null when malformed. -List? _infoSizes(Map j) { +/// The tracks of a channel's `info` frame (`size`, or `tracks[]` each +/// with a `size` or, `"as":"pieces"`, a piece list), or null when +/// malformed. +List<_TrackInfo>? _infoTracks(Map j) { final size = j['size'], tracks = j['tracks']; - if (size is int) return size > 0 ? [size] : null; - if (tracks is! List || tracks.isEmpty) return null; - final out = []; + if (size is int) return size > 0 ? [(size: size, pieces: null)] : null; + if (tracks is! List || tracks.isEmpty || tracks.length > MediaRelay.maxTracks) return null; + final out = <_TrackInfo>[]; for (final t in tracks) { - final n = t is Map ? t['size'] : null; - if (n is! int || n <= 0) return null; - out.add(n); + if (t is! Map) return null; + if (t['as'] == 'pieces') { + final list = RelayPieceList.fromJson(t); + if (list == null) return null; + out.add((size: null, pieces: list)); + } else { + final n = t['size']; + if (n is! int || n <= 0) return null; + out.add((size: n, pieces: null)); + } } return out; } @@ -192,7 +238,9 @@ Future _closeNotSource(RelicWebSocket ws) async { /// `GET /r///source` (WebSocket): the source member's relay /// channel. `GET /r////`: a viewer reads the /// relayed file, Range as for `/m`; `...///` one -/// track of a media relayed as several. [relay] gives the current relay. +/// track of a media with tracks (a pieces track's media playlist), and +/// `...///p/` piece `i` of a pieces track. [relay] +/// gives the current relay. void mountRelayRoutes( RelicApp app, SessionMap sessions, @@ -230,16 +278,17 @@ void mountRelayRoutes( switch (j['t']) { case 'info': if (attached) continue; - final sizes = _infoSizes(j); - if (sizes == null || sizes.length != r.buffers.length) { + final infos = _infoTracks(j); + if (infos == null || !r._matches(infos)) { // Not the media this swap named: the source stops. _log.warning('relay channel: info does not match the media (${r.name})'); await _closeNotSource(ws); return; } attached = true; - _log.info('relay channel up (${r.name}, ${sizes.join(' + ')} bytes)'); - r._attachChannel(ws, sizes); + _log.info('relay channel up (${r.name}, ' + '${infos.map((t) => t.size != null ? '${t.size} bytes' : '${t.pieces!.length} pieces').join(' + ')})'); + r._attachChannel(ws, infos); case 'end': if (id is int) { for (final b in r.buffers) { @@ -272,7 +321,7 @@ void mountRelayRoutes( }); }); - Future viewer(Request req, {required bool tracked}) async { + Future viewer(Request req, {required bool tracked, bool piece = false}) async { final stopwatch = Stopwatch()..start(); final sid = req.pathParameters.raw[#sid] ?? ''; final token = req.pathParameters.raw[#token] ?? ''; @@ -294,15 +343,39 @@ void mountRelayRoutes( headers: Headers.build((h) => h['retry-after'] = ['2'])); if (!await r.whenKnown()) return notConnected; final buffer = r.bufferAt(track); - if (buffer == null) return _plain(404); + if (buffer == null || (piece && buffer is! PieceBuffer)) return _plain(404); if (!await buffer.ready()) return notConnected; - final size = buffer.size; final typed = req.headers.range; String? rangeHeader; if (typed != null && typed.ranges.isNotEmpty) { final x = typed.ranges.first; rangeHeader = '${typed.unit}=${x.start?.toString() ?? ''}-${x.end?.toString() ?? ''}'; } + if (buffer is PieceBuffer) { + if (!piece) { + buffer.opened(token); + return Response(200, + body: Body.fromString(buffer.playlist ?? '', mimeType: MimeType('application', 'vnd.apple.mpegurl')), + headers: Headers.build((h) => h['cache-control'] = ['no-store'])); + } + final index = int.tryParse(req.pathParameters.raw[#piece] ?? ''); + if (index == null || index < 0 || index >= buffer.list!.length) return _plain(404); + final data = await buffer.piece(token, index); + if (data == null) return notConnected; + final res = resolveMediaRequest(rangeHeader: rangeHeader, size: data.length, filename: 'piece'); + final headers = Headers.build((h) => res.headers.forEach((k, v) => h[k] = [v])); + if (res.status != 200 && res.status != 206) { + return Response(res.status, body: Body.fromString('status ${res.status}'), headers: headers); + } + final end = res.end ?? data.length - 1; + final n = end - res.start + 1; + buffer.bytesToViewers += n; + session.addBytes(token, n); + return Response(res.status, + body: Body.fromData(Uint8List.sublistView(data, res.start, end + 1), mimeType: MimeType.octetStream), + headers: headers); + } + final size = (buffer as RelayBuffer).size; final res = resolveMediaRequest(rangeHeader: rangeHeader, size: size, filename: r.name); final headers = Headers.build((h) => res.headers.forEach((k, v) => h[k] = [v])); if (res.status != 200 && res.status != 206) { @@ -337,6 +410,7 @@ void mountRelayRoutes( // wants one parameter name per level). app.get('/r/:sid/:token/:mediaId/:seg', (final req) => viewer(req, tracked: false)); app.get('/r/:sid/:token/:mediaId/:seg/:name', (final req) => viewer(req, tracked: true)); + app.get('/r/:sid/:token/:mediaId/:seg/p/:piece', (final req) => viewer(req, tracked: true, piece: true)); } Response _plain(int status) => Response(status, body: Body.fromString('status $status')); diff --git a/packages/server/lib/src/server_hub.dart b/packages/server/lib/src/server_hub.dart index 0083e75..dffe568 100644 --- a/packages/server/lib/src/server_hub.dart +++ b/packages/server/lib/src/server_hub.dart @@ -58,8 +58,8 @@ class ServerHub { final m = engine.media; if (!identical(relay, r) || m?.relay == null) return; final was = m!.relay!; - if (was.live == r.live && was.tracks.join(',') == r.kinds.join(',')) return; - engine.media = m.copyWith(relay: was.copyWith(live: r.live, tracks: r.kinds)); + if (was.live == r.live && was.tracks.join(',') == r.tracks.join(',')) return; + engine.media = m.copyWith(relay: was.copyWith(live: r.live, tracks: r.tracks)); changed(); } @@ -244,7 +244,8 @@ void mountSyncRoute( String Function(String sid, String token, SessionMedia? media, String host) urlBuilder, { String Function(String sid, String token, String itemId, String host)? relayUrlBuilder, - String Function(String sid, String token, int mediaId, String name, int tracks, String host)? relayMediaUrlBuilder, + String Function(String sid, String token, int mediaId, String name, List tracks, String host)? + relayMediaUrlBuilder, }) { app.get('/sync', (final req) { // Host the client dialled (Host header), so file URLs work for remote joiners. @@ -551,7 +552,7 @@ void mountSyncRoute( relayMediaUrlBuilder != null ? (relayed.known ? relayMediaUrlBuilder( - 'default', activeToken, relayed.mediaId, relayed.name, relayed.buffers.length, dialledHost) + 'default', activeToken, relayed.mediaId, relayed.name, relayed.tracks, dialledHost) : null) : media?.file != null ? urlBuilder('default', activeToken, media, dialledHost) @@ -577,7 +578,7 @@ void mountSyncRoute( }); } -ActiveRelay _memberRelay(ServerHub hub, String token, String name, List tracks) { +ActiveRelay _memberRelay(ServerHub hub, String token, String name, List tracks) { late final ActiveRelay r; r = ActiveRelay.fromMember(hub.engine.mediaId, name, token, tracks: tracks, onLiveChanged: () => hub.relayLiveChanged(r)); diff --git a/packages/server/test/relay_test.dart b/packages/server/test/relay_test.dart index b9ce220..63a174b 100644 --- a/packages/server/test/relay_test.dart +++ b/packages/server/test/relay_test.dart @@ -283,6 +283,46 @@ void main() { expect(carol.welcome.mediaUrl, mpvEdl(['$carolBase/0/$name', '$carolBase/1/$name'])); }); + test('a pieces track and a bytes track: playlist, whole pieces, Range inside a piece', () async { + final alice = await join('alice'); + final bob = await join('bob'); + await swapRelay(alice, relay: { + 'tracks': [ + {'kind': 'video', 'as': 'pieces'}, + {'kind': 'audio'}, + ], + }); + final pieces = [for (var i = 0; i < 4; i++) Uint8List.fromList(List.generate(1000 + i, (j) => (i * 31 + j) % 256))]; + final shape = RelayPieceList([ + const RelayPiece.init(), + for (var i = 1; i < 4; i++) const RelayPiece.segment(6, map: 0), + ]); + final reads = []; + await startSource(alice, + from: _Tracks([ + RelayTrack.pieces('video', _MemPieces(shape, pieces, reads)), + RelayTrack('audio', FileRelayBytes(file)), + ])); + final id = mediaIdOf(alice); + final base = 'http://127.0.0.1:${server.boundPort}/r/default/${bob.welcome.token}/$id'; + final name = Uri.encodeComponent('ep 01.mkv'); + final (ps, ph, playlist) = await get('$base/0/$name.m3u8'); + expect(ps, 200); + expect(ph['content-type'], startsWith('application/vnd.apple.mpegurl')); + expect(utf8.decode(playlist), shape.playlist()); + expect((await get('$base/0/p/2')).$3, pieces[2]); + final (rs, rh, part) = await get('$base/0/p/3', range: 'bytes=10-19'); + expect((rs, rh['content-range']), (206, 'bytes 10-19/1003')); + expect(part, pieces[3].sublist(10, 20)); + expect((await get('$base/1/$name', range: 'bytes=5-')).$3, bytes.sublist(5)); + expect((await get('$base/0/p/4')).$1, 404); + expect((await get('$base/1/p/0')).$1, 404); + expect(reads, isNot(contains(-1))); + final carol = await join('carol'); + final carolBase = base.replaceFirst(bob.welcome.token, carol.welcome.token); + expect(carol.welcome.mediaUrl, mpvEdl(['$carolBase/0/$name.m3u8', '$carolBase/1/$name'])); + }); + test('Jellyfin swap can carry a relay', () async { final alice = await join('alice'); final before = alice.raw.length; @@ -412,7 +452,7 @@ void main() { setUp(() { buf = RelayBuffer(ahead: 2 * RelayBuffer.slice, back: RelayBuffer.slice, pin: RelayBuffer.slice); counted = _CountingBytes(FileRelayBytes(file)); - feed = LocalRelayFeed(counted, buf); + feed = LocalRelayFeed(RelayTrack('', counted), buf); buf.attach(feed, bytes.length); }); tearDown(() async { @@ -470,6 +510,24 @@ class _Tracks implements RelayMedia { Future close() async {} } +/// A pieces track from memory; [reads] logs the pieces pulled. +class _MemPieces implements RelayPieces { + final RelayPieceList shape; + final List data; + final List reads; + _MemPieces(this.shape, this.data, this.reads); + @override + Future list() async => shape; + @override + Stream> read(int index) async* { + reads.add(index); + yield data[index]; + } + + @override + Future close() async {} +} + /// yt-dlp that "resolves" any page to [tracks]. class _FakeYtDlp extends YtDlp { final List tracks;