diff --git a/README.md b/README.md index ad2db37..5f3b1c9 100644 --- a/README.md +++ b/README.md @@ -169,13 +169,16 @@ Watching: your watched state, Continue watching and resume point follow what you saw (reported on start, every 10 s, on pause, resume, seek and track change, and when it stops). -- Friends without an account watch through the host: the server fetches - the file with the host's sign-in and passes it on (only the current item, - only to people in the session). They show "via host" in the member list. - A dedicated server relays when started with `--jellyfin ` (see - [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md), Jellyfin relay). -- Without either, the app says "This needs a Jellyfin account on ..." with - a button to link one; it plays as soon as you are signed in. +- Friends without an account watch through the server: whoever put the + item up passes it on automatically, with their own sign-in, which never + leaves their computer (only the current item, only to people in the + session). They show "via host" in the member list. When the app hosts, + or a dedicated server has its own login (`--jellyfin `, see + [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md), Jellyfin relay), the server + fetches it itself instead. +- If nobody can pass it on (the person who put it up left), the app says + "No Jellyfin account here for ..." with a button to link one; it plays as + soon as you are signed in. Episodes: the ⏮ / ⏭ buttons next to the controls (or `<` / `>`, Shift+P / Shift+N) put up the previous or next episode for everyone. From @@ -194,7 +197,24 @@ next episode. This is per person; nobody else is affected. "Change video" in the session panel (Tab) opens the same picker as the Host tab: the library starts on the series you are watching with the current -season, else where you left it. File is offered to the host only. +season, else where you left it. + +Anyone can put up a file from their own computer, also on a dedicated +server: it streams from your computer through the server to everyone (the +server keeps a few minutes in memory, so seeks and skipping the opening +are quick). Keep the app open while it plays. For a link you choose how +the others get it: +- Everyone fetches it (default): each app plays the link itself. +- Server fetches it: the server downloads it once for everyone, for links + that are slow or blocked for some. The server must be able to reach it, + and it must have a password (an open server refuses: anyone could make + it fetch addresses only it can reach). +- Through me: your computer downloads it and passes it on, for links only + you can open. + +The last two need a link to a video file (not a page like YouTube) and +never show the link to anyone. 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/jellyfin/session_media.dart b/apps/desktop/lib/src/jellyfin/session_media.dart index a7f0d05..4f98d4a 100644 --- a/apps/desktop/lib/src/jellyfin/session_media.dart +++ b/apps/desktop/lib/src/jellyfin/session_media.dart @@ -201,7 +201,8 @@ final class OwnPlay extends JellyfinPlan { bool get jfAccess => true; } -/// Through the sync server's relay (`/j/...`). +/// Through the server: its own Jellyfin login (`/j/...`) or the member +/// who put it up (`/r/...`). final class RelayPlay extends JellyfinPlan { final String url; const RelayPlay(this.url); @@ -210,7 +211,7 @@ final class RelayPlay extends JellyfinPlan { bool get jfAccess => false; } -/// Neither: no (usable) account here and the server cannot relay. +/// Neither: no (usable) account here and nothing relays it. final class BlockedPlay extends JellyfinPlan { final String serverId; /// Known only when this install has (had) an account there. @@ -224,20 +225,22 @@ final class BlockedPlay extends JellyfinPlan { String get message { final where = serverName == null || serverName!.isEmpty ? 'that Jellyfin server' : serverName!; - final base = 'This needs a Jellyfin account on $where (or the host can relay)'; - return problem == null ? base : '$base — $problem'; + final base = 'No Jellyfin account here for $where, and nobody is passing it on'; + return problem == null ? base : '$base ($problem)'; } } /// Picks how to play [ref]: this install's account first (PlaybackInfo -/// with its own user; a 401 marks the account expired), else the relay -/// when the server offers it ([relay] = `state.relay`, [relayUrl] this -/// member's `/j/...` URL), else [BlockedPlay]. +/// with its own user; a 401 marks the account expired), else the server's +/// own Jellyfin login ([relay] = `state.relay`, [relayUrl] this member's +/// `/j/...` URL), else the member who put it up ([memberRelayUrl], the +/// `/r/...` URL when `media.relay` is set), else [BlockedPlay]. Future resolveJellyfinRef( JellyfinAccounts accounts, JellyfinRef ref, { required bool relay, required String? relayUrl, + String? memberRelayUrl, Duration timeout = const Duration(seconds: 15), }) async { await accounts.ensureLoaded(); @@ -282,6 +285,7 @@ Future resolveJellyfinRef( } } if (relay && relayUrl != null && relayUrl.isNotEmpty) return RelayPlay(relayUrl); + if (memberRelayUrl != null && memberRelayUrl.isNotEmpty) return RelayPlay(memberRelayUrl); return BlockedPlay(serverId: ref.serverId, serverName: account?.displayName, problem: problem); } diff --git a/apps/desktop/lib/src/screens/home/media_picker.dart b/apps/desktop/lib/src/screens/home/media_picker.dart index 9a47fba..ad0fa99 100644 --- a/apps/desktop/lib/src/screens/home/media_picker.dart +++ b/apps/desktop/lib/src/screens/home/media_picker.dart @@ -37,6 +37,14 @@ class MediaPickerController extends ChangeNotifier { final TextEditingController file; MediaKind _source; String? _notice; + LinkRelay _linkRelay = LinkRelay.direct; + + LinkRelay get linkRelay => _linkRelay; + set linkRelay(LinkRelay v) { + if (v == _linkRelay) return; + _linkRelay = v; + notifyListeners(); + } JellyfinMedia? _jellyfin; /// The Jellyfin item picked in the browser (kept while another source @@ -71,7 +79,7 @@ class MediaPickerController extends ChangeNotifier { switch (_source) { case MediaKind.link: final url = link.text.trim(); - return url.isEmpty ? null : LinkMedia(url); + return url.isEmpty ? null : LinkMedia(url, relay: _linkRelay); case MediaKind.file: final path = file.text.trim(); return path.isEmpty ? null : FileMedia(path); @@ -117,15 +125,21 @@ class MediaPicker extends StatefulWidget { final bool expanded; /// Where the Jellyfin browser opens instead of its remembered location. final BrowseStart? jellyfinStart; - /// False hides the File source (only the host can serve a file). + /// False hides the File source. final bool allowFile; /// Shown under the Link field instead of the default help. final String? linkHelp; + /// Shown under the File field instead of the default help. + final String? fileHelp; + /// Offer the [LinkRelay] choice under the Link field (in a session). + final bool linkRelayChoice; const MediaPicker({ super.key, this.jellyfinStart, this.allowFile = true, this.linkHelp, + this.fileHelp, + this.linkRelayChoice = false, required this.controller, this.onChanged, this.onSubmit, @@ -247,10 +261,25 @@ class _MediaPickerState extends State { icon: const Icon(Icons.content_paste, size: 18), label: const Text('Paste'), ), - help: widget.linkHelp ?? - 'A direct http(s) link to a video file; everyone streams it from there. ' - 'Jellyfin links are shared as the item, without your sign-in.', + help: widget.linkRelayChoice + ? _linkRelayHelp(_c.linkRelay) + : widget.linkHelp ?? + 'A direct http(s) link to a video file; everyone streams it from there. ' + 'Jellyfin links are shared as the item, without your sign-in.', notice: _c.notice, + extra: widget.linkRelayChoice + ? SegmentedButton( + key: const ValueKey('link-relay'), + showSelectedIcon: false, + segments: const [ + ButtonSegment(value: LinkRelay.direct, label: Text('Everyone fetches it')), + ButtonSegment(value: LinkRelay.server, label: Text('Server fetches it')), + ButtonSegment(value: LinkRelay.me, label: Text('Through me')), + ], + selected: {_c.linkRelay}, + onSelectionChanged: (s) => _c.linkRelay = s.first, + ) + : null, ), MediaKind.file => _FieldRow( key: const ValueKey('media-file'), @@ -264,7 +293,8 @@ class _MediaPickerState extends State { icon: const Icon(Icons.folder_open, size: 18), label: const Text('Browse'), ), - help: 'Friends stream it from you while you watch. Optional — you can host without media.', + help: widget.fileHelp ?? + 'Friends stream it from you while you watch. Optional — you can host without media.', notice: _c.notice, ), }; @@ -325,8 +355,10 @@ class _FieldRow extends StatelessWidget { final Widget button; final String help; final String? notice; + /// Under the field, above the help (e.g. the link relay choice). + final Widget? extra; const _FieldRow( - {super.key, required this.field, required this.button, required this.help, this.notice}); + {super.key, required this.field, required this.button, required this.help, this.notice, this.extra}); @override Widget build(BuildContext context) { @@ -346,6 +378,7 @@ class _FieldRow extends StatelessWidget { button, ], ), + if (extra != null) ...[const SizedBox(height: 10), extra!], const SizedBox(height: 8), Text(notice ?? help, style: small), ], @@ -354,3 +387,12 @@ class _FieldRow extends StatelessWidget { ); } } + +String _linkRelayHelp(LinkRelay r) => switch (r) { + LinkRelay.direct => 'Everyone streams the link themselves. Jellyfin links are shared as the item, ' + 'without your sign-in.', + LinkRelay.server => 'The server downloads it once and streams it to everyone: for links that are ' + 'slow or blocked for some. The server must be able to reach the link; it is not shown to anyone.', + LinkRelay.me => 'This computer downloads it and passes it on through the server: for links only you ' + 'can open. Keep the app open while it plays; the link is not shown to anyone.', + }; diff --git a/apps/desktop/lib/src/screens/home/picked_media.dart b/apps/desktop/lib/src/screens/home/picked_media.dart index a217e2b..39e4a40 100644 --- a/apps/desktop/lib/src/screens/home/picked_media.dart +++ b/apps/desktop/lib/src/screens/home/picked_media.dart @@ -7,10 +7,16 @@ sealed class PickedMedia { String get label; } -/// A direct http(s) link everyone streams from. +/// How the others get a link: each fetches it ([direct]), the server +/// fetches it once for everyone ([server]), or this computer does and +/// passes it on through the server ([me]; for links only you can reach). +enum LinkRelay { direct, server, me } + +/// An http(s) link. final class LinkMedia extends PickedMedia { final String url; - const LinkMedia(this.url); + final LinkRelay relay; + const LinkMedia(this.url, {this.relay = LinkRelay.direct}); @override String get label { @@ -25,14 +31,15 @@ final class LinkMedia extends PickedMedia { } @override - bool operator ==(Object other) => other is LinkMedia && other.url == url; + bool operator ==(Object other) => other is LinkMedia && other.url == url && other.relay == relay; @override - int get hashCode => url.hashCode; + int get hashCode => Object.hash(url, relay); @override String toString() => 'LinkMedia($url)'; } -/// A file on this computer; friends stream it from the host. +/// A file on this computer; the others stream it from here (the host's +/// app serves it itself, anyone else's goes through the server's relay). final class FileMedia extends PickedMedia { final String path; const FileMedia(this.path); diff --git a/apps/desktop/lib/src/screens/session/jellyfin_widgets.dart b/apps/desktop/lib/src/screens/session/jellyfin_widgets.dart index 04717d3..d93839d 100644 --- a/apps/desktop/lib/src/screens/session/jellyfin_widgets.dart +++ b/apps/desktop/lib/src/screens/session/jellyfin_widgets.dart @@ -144,7 +144,9 @@ class JellyfinBlockedCard extends StatelessWidget { const SizedBox(height: 10), Text(blocked.message, textAlign: TextAlign.center, style: t.titleSmall), const SizedBox(height: 6), - Text('Link your own account in Integrations, or ask the host to relay it.', + Text( + 'Whoever put it up passes it on automatically while their app is open; ' + 'if they left, ask them to put it up again. Or link your own account in Integrations.', textAlign: TextAlign.center, style: t.bodySmall?.copyWith(color: Colors.grey)), const SizedBox(height: 14), Wrap( diff --git a/apps/desktop/lib/src/screens/session/swap_panel.dart b/apps/desktop/lib/src/screens/session/swap_panel.dart index 38e695e..1ebb3ba 100644 --- a/apps/desktop/lib/src/screens/session/swap_panel.dart +++ b/apps/desktop/lib/src/screens/session/swap_panel.dart @@ -6,8 +6,9 @@ import '../home/media_picker.dart'; /// "Change video" in a session: the same [MediaPicker] as Home's Host tab /// (Jellyfin browser, Link, File) and a button that puts the pick up for -/// everyone. File is offered to the host only (the file streams from the -/// host's disk). [onSwap] returns an error text, or null when sent. +/// everyone. A file streams from this computer: the host's app serves it, +/// anyone else's goes through the server's relay. [onSwap] returns an +/// error text, or null when sent. class SwapPanel extends StatefulWidget { final MediaPickerController controller; final JellyfinAccounts accounts; @@ -35,13 +36,6 @@ class _SwapPanelState extends State { bool _busy = false; String? _error; - @override - void initState() { - super.initState(); - final c = widget.controller; - if (!widget.isHost && c.source == MediaKind.file) c.source = MediaKind.link; - } - Future _swap() async { final media = widget.controller.selection; if (media == null || _busy) return; @@ -91,7 +85,10 @@ class _SwapPanelState extends State { controller: widget.controller, jellyfin: widget.accounts, enabled: !_busy, - allowFile: widget.isHost, + linkRelayChoice: true, + fileHelp: widget.isHost + ? 'Friends stream it from your computer.' + : 'Streams from your computer through the server. Keep the app open while it plays.', jellyfinStart: widget.jellyfinStart, onOpenIntegrations: widget.onOpenIntegrations, onSubmit: _swap, diff --git a/apps/desktop/lib/src/session/session_coordinator.dart b/apps/desktop/lib/src/session/session_coordinator.dart index e5d0219..d51f9b2 100644 --- a/apps/desktop/lib/src/session/session_coordinator.dart +++ b/apps/desktop/lib/src/session/session_coordinator.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:io'; import 'package:flutter/widgets.dart'; import 'package:signals/signals_flutter.dart'; @@ -92,6 +93,35 @@ class ToastItem { const _toastActions = {'join', 'leave', 'kick', 'swap', 'chat'}; +/// `media.relay` of a state, or null when the media is not relayed. +MediaRelay? stateMediaRelay(StateBroadcast state) { + final r = state.media?['relay']; + return r is Map ? MediaRelay.fromJson(Map.from(r)) : null; +} + +/// What this member serves when its own swap is relayed: set at swap time, +/// claimed by the state that names this member as the source. +sealed class _RelayOrigin { + const _RelayOrigin(); +} + +final class _FileOrigin extends _RelayOrigin { + final String path; + const _FileOrigin(this.path); +} + +final class _LinkOrigin extends _RelayOrigin { + final String url; + const _LinkOrigin(this.url); +} + +/// A Jellyfin item: served with this install's account, else with the +/// pasted [link] (its api_key stays on this machine). +final class _JellyfinOrigin extends _RelayOrigin { + final String? link; + const _JellyfinOrigin(this.link); +} + String? stateMediaUrl(StateBroadcast state) { final media = state.media; if (media == null) return null; @@ -184,6 +214,14 @@ class SessionCoordinator { String? _openUrl; String _wsHost = ''; int _wsPort = 0; + _RelayOrigin? _pendingOrigin; + // What this member serves for media [$1]; survives a reconnect (the + // state then looks new again). + (int, _RelayOrigin)? _activeOrigin; + RelaySource? _relaySource; + int _relaySourceMediaId = -1; + /// This member serves the current media's bytes through the server. + final relaying = signal(false); // Server seek applied to the player (event id and when), checked once it // should have landed. int? _pendingSeekEvent; @@ -392,6 +430,78 @@ class SessionCoordinator { } } + bool _relayWaitShown = false; + + /// Relayed through a member whose channel is down: say who everyone is + /// waiting for (the buffer still plays what it has). + void _relayStatus(StateBroadcast r) { + final relay = stateMediaRelay(r); + final waiting = relay != null && + relay.source.isNotEmpty && + relay.source != sync.memberId && + !relay.live && + _openUrl != null && + jellyfinPlan.peek() is! OwnPlay; + if (waiting) { + final who = r.members.where((m) => m.id == relay.source).firstOrNull?.name; + health.value = HealthStatus(HealthSeverity.warn, + who == null ? 'the member who put this up left' : 'waiting for $who to pass the video on'); + _relayWaitShown = true; + } else if (_relayWaitShown) { + _relayWaitShown = false; + health.value = const HealthStatus(HealthSeverity.ok, 'connected'); + } + } + + String _relayMediaUrlFor(String name, int mediaId) => + 'http://$_wsHost:$_wsPort/r/default/${sync.token}/$mediaId/${Uri.encodeComponent(name)}'; + + /// Serves [bytes] for media [mediaId] on this member's relay channel. + /// The server only pulls while someone watches through it. + void _startRelaySource(int mediaId, RelayBytes bytes, {bool announce = false}) { + // Back after a reconnect on the same media: it is still serving (and + // redials by itself). + if (_relaySource != null && _relaySourceMediaId == mediaId) { + unawaited(bytes.close()); + return; + } + _stopRelaySource(); + final src = RelaySource( + Uri.parse('ws://$_wsHost:$_wsPort/r/default/${sync.token}/source'), bytes); + _relaySource = src; + _relaySourceMediaId = mediaId; + relaying.value = true; + if (announce) { + _pushToast('Passing the video on through the server. Keep the app open while it plays.'); + } + unawaited(() async { + try { + await src.start(); + } catch (e) { + debugPrint('relay source failed: $e'); + if (identical(_relaySource, src)) { + _relaySource = null; + relaying.value = false; + _pushToast('Cannot pass the video on: $e'); + } + return; + } + await src.done; + if (identical(_relaySource, src)) { + _relaySource = null; + relaying.value = false; + } + }()); + } + + void _stopRelaySource() { + final src = _relaySource; + _relaySource = null; + _relaySourceMediaId = -1; + relaying.value = false; + if (src != null) unawaited(src.close()); + } + String _fileUrlFor(String name) { return 'http://$_wsHost:$_wsPort/m/default/${sync.token}/${Uri.encodeComponent(name)}'; } @@ -440,12 +550,35 @@ class SessionCoordinator { r.mediaId == _jfMediaId && sameJellyfinItem(ref, _jfRef) && jellyfinPlan.peek() is OwnPlay; + final relay = stateMediaRelay(r); + final mine = relay != null && relay.source.isNotEmpty && relay.source == sync.memberId; + _RelayOrigin? origin; + if (mine) { + final pending = _pendingOrigin; + if (pending != null) { + _pendingOrigin = null; + _activeOrigin = (r.mediaId, pending); + } + origin = _activeOrigin?.$1 == r.mediaId ? _activeOrigin!.$2 : null; + } + if (_relaySourceMediaId != r.mediaId) _stopRelaySource(); if (!same) { _seekGate.reset(); _clearPendingSeek(); _leaveJellyfinItem(); + final name = (r.media?['name'] as String?) ?? ''; if (ref != null) { - await _playJellyfin(ref, r); + await _playJellyfin(ref, r, + origin: origin is _JellyfinOrigin ? origin : null, + memberRelayUrl: relay != null && !mine ? _relayMediaUrlFor(name, r.mediaId) : null); + } else if (origin is _FileOrigin) { + await _openMedia(origin.path); + _startRelaySource(r.mediaId, FileRelayBytes(File(origin.path)), announce: true); + } else if (origin is _LinkOrigin) { + await _openMedia(origin.url); + _startRelaySource(r.mediaId, HttpRelayBytes(Uri.parse(origin.url)), announce: true); + } else if (relay != null && !mine) { + await _openMedia(_relayMediaUrlFor(name, r.mediaId)); } else { var url = stateMediaUrl(r); if ((url == null || url.isEmpty) && sync.token.isNotEmpty) { @@ -463,6 +596,7 @@ class SessionCoordinator { _clearPendingSeek(); } _jellyfinTransport(r); + _relayStatus(r); if (!_playerReady || _openUrl == null) { _wasPlayAllowed = false; _updateStarving(r, 0); @@ -825,35 +959,77 @@ class SessionCoordinator { } /// A pasted link: a recognised Jellyfin item link goes up as a ref - /// (see [mediaForLink]), anything else as the link. - Future swapLink(String url) async { + /// (see [mediaForLink]) that this member relays for viewers without an + /// account (with the link's key, which stays here); anything else as the + /// link, or relayed per [relay]. Returns an error text, or null when sent. + Future swapLink(String url, {LinkRelay relay = LinkRelay.direct}) async { markInput(); - sync.sendSwap(await mediaForLink(jellyfin, url)); + final media = await mediaForLink(jellyfin, url); + if (media.jellyfin != null) { + _pendingOrigin = _JellyfinOrigin(url); + sync.sendSwap(media.copyWith(relay: const MediaRelay())); + return null; + } + if (relay == LinkRelay.direct) { + sync.sendSwap(media); + return null; + } + // Relayed: it must be a single file the relay can range-read (not a + // page like YouTube, which only the player can resolve). + final probe = HttpRelayBytes(Uri.parse(url)); + final int size; + try { + size = await probe.size().timeout(const Duration(seconds: 15)); + } catch (e) { + return relay == LinkRelay.me + ? 'cannot pass this link on: $e' + : 'cannot check this link: $e'; + } finally { + await probe.close(); + } + if (relay == LinkRelay.me) { + _pendingOrigin = _LinkOrigin(url); + sync.sendSwap(HelloMedia(name: media.name, size: size, duration: 0, relay: const MediaRelay())); + } else { + sync.sendSwap(HelloMedia( + name: media.name, size: size, duration: 0, url: url, relay: const MediaRelay(fetch: true))); + } + return null; } - /// A Jellyfin pick (browser, next / previous episode) goes up as a ref. + /// A Jellyfin pick (browser, next / previous episode) goes up as a ref, + /// relayed by this member for viewers without an account. void swapJellyfin(JellyfinMedia media) { markInput(); - sync.sendSwap(mediaForPick(media)); + _pendingOrigin = const _JellyfinOrigin(null); + sync.sendSwap(mediaForPick(media).copyWith(relay: const MediaRelay())); } - /// The host's File swap: the server takes the file, then the swap puts - /// it up. Returns an error text, or null when sent. + /// A file on this computer. The host's app serves it itself; anyone + /// else's goes through the server's relay. Returns an error text, or null + /// when sent. Future swapFile(String path) async { markInput(); final host = hostFile; - if (host == null) return 'only the host can share a file'; - final file = await host(path); - if (file == null) return 'file not found — check the path'; - sync.sendSwap(HelloMedia(name: file.name, size: file.size, duration: 0)); + if (host != null) { + final file = await host(path); + if (file == null) return 'file not found — check the path'; + sync.sendSwap(HelloMedia(name: file.name, size: file.size, duration: 0)); + return null; + } + final file = File(path); + if (!await file.exists()) return 'file not found — check the path'; + _pendingOrigin = _FileOrigin(path); + sync.sendSwap(HelloMedia( + name: file.uri.pathSegments.last, size: await file.length(), duration: 0, relay: const MediaRelay())); return null; } /// Any picker selection. Future swapPicked(PickedMedia media) async { switch (media) { - case LinkMedia(:final url): - await swapLink(url); + case LinkMedia(:final url, :final relay): + return swapLink(url, relay: relay); case JellyfinMedia(): swapJellyfin(media); case FileMedia(:final path): @@ -978,7 +1154,8 @@ class SessionCoordinator { /// Resolves [ref] for this viewer and opens it: own account (direct /// play, progress reports, neighbouring episodes), the relay, or the /// blocked state. Then carries the audio/subtitle choice over. - Future _playJellyfin(JellyfinRef ref, StateBroadcast r) async { + Future _playJellyfin(JellyfinRef ref, StateBroadcast r, + {_JellyfinOrigin? origin, String? memberRelayUrl}) async { final gen = ++_jfGen; _jfRef = ref; _jfMediaId = r.mediaId; @@ -992,13 +1169,24 @@ class SessionCoordinator { await playback.player.pause(); } catch (_) {} } - final JellyfinPlan plan; + JellyfinPlan plan; try { - plan = await resolveJellyfinRef(jellyfin, ref, relay: r.relay, relayUrl: _relayUrlFor(ref)); + plan = await resolveJellyfinRef(jellyfin, ref, + relay: r.relay, relayUrl: _relayUrlFor(ref), memberRelayUrl: memberRelayUrl); } finally { if (gen == _jfGen) _opening = false; } if (gen != _jfGen) return; + // This member put it up: serve it to viewers without an account, with + // its own sign-in, else the pasted link (played directly here too). + if (origin != null) { + if (plan is OwnPlay) { + _startRelaySource(r.mediaId, HttpRelayBytes(Uri.parse(plan.url), headers: plan.headers)); + } else if (origin.link != null) { + plan = RelayPlay(origin.link!); + _startRelaySource(r.mediaId, HttpRelayBytes(Uri.parse(origin.link!))); + } + } jellyfinPlan.value = plan; switch (plan) { case BlockedPlay(): @@ -1321,6 +1509,9 @@ class SessionCoordinator { _seekGate.reset(); reconnecting.value = false; await _stopJellyfin(); + _stopRelaySource(); + _pendingOrigin = null; + _activeOrigin = null; _playerReady = false; try { await playback.dispose(); diff --git a/apps/desktop/pubspec.lock b/apps/desktop/pubspec.lock index d85d3e3..407c71d 100644 --- a/apps/desktop/pubspec.lock +++ b/apps/desktop/pubspec.lock @@ -723,14 +723,14 @@ packages: path: "../../packages/core" relative: true source: path - version: "0.2.1" + version: "0.2.2" sync_relay_server: dependency: "direct main" description: path: "../../packages/server" relative: true source: path - version: "0.2.1" + version: "0.2.2" synchronized: dependency: transitive description: diff --git a/apps/desktop/test/jellyfin_e2e_test.dart b/apps/desktop/test/jellyfin_e2e_test.dart index 034bc1d..c144a3c 100644 --- a/apps/desktop/test/jellyfin_e2e_test.dart +++ b/apps/desktop/test/jellyfin_e2e_test.dart @@ -10,6 +10,7 @@ import 'package:watch_together/src/jellyfin/accounts.dart'; import 'package:watch_together/src/jellyfin/session_media.dart'; import 'package:watch_together/src/jellyfin/track_prefs.dart'; import 'package:watch_together/src/playback/media_marks.dart'; +import 'package:watch_together/src/screens/home/picked_media.dart'; import 'package:watch_together/src/session/session_coordinator.dart'; // Two account holders and a guest watch Jellyfin episodes together over a @@ -444,7 +445,7 @@ void main() { await b.connect(ConnectParams(host: '127.0.0.1', port: server.boundPort, secret: '', name: 'bob')); await _poll(() => b.jellyfinBlocked.peek() != null, 'blocked'); expect(b.serverState.peek()!.relay, isFalse); - expect(b.jellyfinBlocked.peek()!.message, contains('needs a Jellyfin account')); + expect(b.jellyfinBlocked.peek()!.message, contains('nobody is passing it on')); // Linking an account (Integrations) retries by itself. await accB.link( baseUrl: jelly.base, @@ -462,4 +463,74 @@ void main() { } }, ); + + // The VPS case: a dedicated server with no Jellyfin login. Whoever puts + // the item up passes it on to guests without an account. + test( + 'dedicated server without a login: the guest plays through the member who put it up', + timeout: const Timeout(Duration(minutes: 3)), + () async { + MediaKit.ensureInitialized(); + final jelly = _FakeJellyfin(await _media()); + await jelly.start(); + final server = SyncServer(const AppConfig(port: 0, secret: '', name: 'vps')); + await server.start(); + final accA = await _account(jelly.base, 'ATOKEN', 'ua', 'A-PC'); + Future none() async { + final x = JellyfinAccounts(store: MemoryJellyfinStore()); + await x.ensureLoaded(validate: false); + return x; + } + + final a = SessionCoordinator(jellyfin: accA, trackPrefs: TrackPrefs(store: MemoryJellyfinStore())); + final b = SessionCoordinator(jellyfin: await none(), trackPrefs: TrackPrefs(store: MemoryJellyfinStore())); + final d = SessionCoordinator(jellyfin: await none(), trackPrefs: TrackPrefs(store: MemoryJellyfinStore())); + ConnectParams params(String name) => + ConnectParams(host: '127.0.0.1', port: server.boundPort, secret: '', name: name); + try { + await a.connect(params('alice')); + await b.connect(params('guest')); + await d.connect(params('dave')); + for (final x in [a, b, d]) { + await x.playback.player.setVolume(0); + } + // Alice picks from her browser: a ref plus a relay she sources. + await a.swapPicked(JellyfinMedia(serverId: _sid, baseUrl: jelly.base, itemId: _ep1, title: 'Pilot')); + await _poll(() => b.jellyfinPlan.peek() is RelayPlay && b.playback.durationSecs > 19, 'guest plays'); + expect((b.jellyfinPlan.peek() as RelayPlay).url, contains('/r/default/')); + expect(a.jellyfinPlan.peek(), isA()); + expect(a.relaying.peek(), isTrue); + expect(b.jellyfinBlocked.peek(), isNull); + for (final x in [a, b]) { + x.toggleReady(); + } + await _poll(() => a.playback.player.state.playing && b.playback.player.state.playing, 'playing'); + await Future.delayed(const Duration(seconds: 3)); + expect((a.playback.positionSecs - b.playback.positionSecs).abs(), lessThan(1.0)); + // Every stream request carried alice's token, from her machine. + expect(jelly.streams('ATOKEN'), isNotEmpty); + expect(jelly.calls.where((c) => c.path.startsWith('/Videos/')).every((c) => + _FakeJellyfin.authFields(c.auth)['Token'] == 'ATOKEN'), isTrue); + for (final x in [a, b]) { + x.toggleReady(); + } + + // Dave has no account but a link with a key: he passes it on, the + // key never leaves his machine. + final link = '${jelly.base}/Videos/$_ep2/stream?static=true&api_key=ATOKEN'; + expect(await d.swapPicked(LinkMedia(link)), isNull); + await _poll(() => (b.serverState.peek()?.media?['jellyfin'] as Map?)?['itemId'] == _ep2, 'ep2 up'); + expect(b.serverState.peek()!.toJson().toString(), isNot(contains('api_key'))); + await _poll(() => d.relaying.peek() && d.playback.durationSecs > 19, 'dave plays his link'); + await _poll(() => b.jellyfinPlan.peek() is RelayPlay && b.playback.durationSecs > 19, 'guest plays ep2'); + expect((b.jellyfinPlan.peek() as RelayPlay).url, contains('/r/default/')); + } finally { + for (final x in [a, b, d]) { + await x.dispose(); + } + await server.stop(); + await jelly.stop(); + } + }, + ); } diff --git a/apps/desktop/test/jellyfin_session_test.dart b/apps/desktop/test/jellyfin_session_test.dart index a54f740..964307b 100644 --- a/apps/desktop/test/jellyfin_session_test.dart +++ b/apps/desktop/test/jellyfin_session_test.dart @@ -93,7 +93,7 @@ void main() { final plan = await resolveJellyfinRef(accounts, episodeRef, relay: false, relayUrl: null); expect(plan, isA()); expect((plan as BlockedPlay).message, - 'This needs a Jellyfin account on that Jellyfin server (or the host can relay)'); + 'No Jellyfin account here for that Jellyfin server, and nobody is passing it on'); expect(plan.jfAccess, isNull); }); @@ -105,7 +105,7 @@ void main() { expect(plan, isA()); expect(accounts.statusOf(fixtureServerId), JellyfinAccountStatus.expired); final blocked = await resolveJellyfinRef(accounts, episodeRef, relay: false, relayUrl: null); - expect((blocked as BlockedPlay).message, contains('on Example Server')); + expect((blocked as BlockedPlay).message, contains('for Example Server')); expect(blocked.message, contains('sign-in expired')); }); @@ -511,13 +511,13 @@ void main() { blocked: const BlockedPlay(serverId: fixtureServerId, serverName: 'Example Server'), onLinkAccount: () => link++, onRetry: () {}))); - expect(find.text('This needs a Jellyfin account on Example Server (or the host can relay)'), + expect(find.text('No Jellyfin account here for Example Server, and nobody is passing it on'), findsOneWidget); await tester.tap(find.byKey(const ValueKey('link-jellyfin-account'))); expect(link, 1); }); - testWidgets('swap panel: the shared picker, File only for the host', (tester) async { + testWidgets('swap panel: the shared picker, File for everyone, link relay choice', (tester) async { await setSize(tester, const Size(1400, 900)); final (accounts, _) = await linkedAccounts(); final c = MediaPickerController(source: MediaKind.link); @@ -535,14 +535,16 @@ void main() { onClose: () => closed++, ))); await settle(tester); - expect(find.widgetWithText(ButtonSegment, 'File'), findsNothing); - expect(find.text('File'), findsNothing); + // A guest's file goes through the server's relay. + expect(find.text('File'), findsOneWidget); expect(find.text('Jellyfin'), findsOneWidget); await tester.enterText(find.byType(TextField), 'https://cdn.example/movie.mkv'); await tester.pump(); + await tester.tap(find.text('Through me')); + await tester.pump(); await tester.tap(find.byKey(const ValueKey('swap-go'))); await settle(tester); - expect(swapped, [const LinkMedia('https://cdn.example/movie.mkv')]); + expect(swapped, [const LinkMedia('https://cdn.example/movie.mkv', relay: LinkRelay.me)]); expect(closed, 1); // The Jellyfin source is the library browser; a pick goes up as is. @@ -554,6 +556,9 @@ void main() { controller: c, accounts: accounts, isHost: true, onSwap: (_) async => 'x', onClose: () {}))); await settle(tester); expect(find.text('File'), findsOneWidget); + await tester.tap(find.text('File')); + await settle(tester); + expect(find.text('Friends stream it from your computer.'), findsOneWidget); }); testWidgets('swap panel opens on the series of the episode playing, episode highlighted', diff --git a/apps/desktop/test/relay_e2e_test.dart b/apps/desktop/test/relay_e2e_test.dart new file mode 100644 index 0000000..e0096ef --- /dev/null +++ b/apps/desktop/test/relay_e2e_test.dart @@ -0,0 +1,242 @@ +import 'dart:async'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:media_kit/media_kit.dart'; +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:watch_together/src/jellyfin/accounts.dart'; +import 'package:watch_together/src/jellyfin/track_prefs.dart'; +import 'package:watch_together/src/screens/home/picked_media.dart'; +import 'package:watch_together/src/session/session_coordinator.dart'; + +// The relay with real coordinators, real mpv and a dedicated server that +// has no file and no Jellyfin login of its own (the VPS setup): a guest's +// file, and links passed on by a member or fetched by the server. + +final _tmp = Platform.environment['WT_TEST_TMP'] ?? Directory.systemTemp.path; + +Future _media() async { + final path = '$_tmp/relay_e2e_20s.mkv'; + if (await File(path).exists()) return path; + final r = await Process.run('ffmpeg', [ + '-y', '-v', 'error', + '-f', 'lavfi', '-i', 'testsrc2=size=320x180:rate=24', + '-f', 'lavfi', '-i', 'sine=frequency=440:sample_rate=48000', + '-t', '20', '-c:v', 'libx264', '-preset', 'ultrafast', '-g', '24', '-pix_fmt', 'yuv420p', + '-c:a', 'aac', path, + ]); + if (r.exitCode != 0) fail('ffmpeg failed: ${r.stderr}'); + return path; +} + +Future _poll(FutureOr Function() cond, String what, + {Duration timeout = const Duration(seconds: 30)}) async { + final deadline = DateTime.now().add(timeout); + while (DateTime.now().isBefore(deadline)) { + if (await cond()) return; + await Future.delayed(const Duration(milliseconds: 200)); + } + fail('timed out waiting for $what'); +} + +/// Serves one file with ranges; [requests] counts hits. +Future<(HttpServer, List)> _origin(String file) async { + final s = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + final requests = []; + s.listen((req) async { + requests.add(req.uri.toString()); + final size = await File(file).length(); + final r = resolveMediaRequest(rangeHeader: req.headers.value('range'), size: size, filename: 'v.mkv'); + req.response.statusCode = r.status; + r.headers.forEach(req.response.headers.set); + try { + if (r.status == 200 || r.status == 206) { + await req.response.addStream(File(file).openRead(r.start, (r.end ?? size - 1) + 1)); + } + await req.response.close(); + } catch (_) {} + }); + return (s, requests); +} + +void main() { + late SyncServer server; + late List all; + + Future member(String name) async { + final acc = JellyfinAccounts(store: MemoryJellyfinStore()); + await acc.ensureLoaded(validate: false); + final c = SessionCoordinator(jellyfin: acc, trackPrefs: TrackPrefs(store: MemoryJellyfinStore())); + all.add(c); + await c.connect(ConnectParams(host: '127.0.0.1', port: server.boundPort, secret: 'pw', name: name)); + await c.playback.player.setVolume(0); + return c; + } + + Future playTogether(List xs) async { + await _poll(() => xs.every((x) => x.playback.durationSecs > 19), 'all opened it'); + for (final x in xs) { + x.toggleReady(); + } + await _poll(() => xs.every((x) => x.playback.player.state.playing), 'all playing'); + await Future.delayed(const Duration(seconds: 3)); + final pos = [for (final x in xs) x.playback.positionSecs]; + // ignore: avoid_print + print('positions after 3 s: $pos'); + for (final p in pos) { + expect((p - pos.first).abs(), lessThan(1.0), reason: 'in sync: $pos'); + } + for (final x in xs) { + x.toggleReady(); + } + await _poll(() => xs.every((x) => !x.playback.player.state.playing), 'held'); + } + + setUp(() async { + MediaKit.ensureInitialized(); + all = []; + server = SyncServer(const AppConfig(port: 0, secret: 'pw', name: 'vps')); + await server.start(); + }); + + tearDown(() async { + for (final x in all) { + await x.dispose(); + } + await server.stop(); + }); + + test('a guest puts up a file: it plays for everyone through the relay', timeout: const Timeout(Duration(minutes: 2)), + () async { + final file = await _media(); + final alice = await member('alice'); + final bob = await member('bob'); + expect(await bob.swapPicked(FileMedia(file)), isNull); + await _poll(() => stateMediaRelay(alice.serverState.peek()!)?.source == bob.sync.memberId, 'relay state'); + final state = alice.serverState.peek()!; + expect(state.media!.containsKey('url'), isFalse); + expect(bob.relaying.peek(), isTrue); + await playTogether([alice, bob]); + final relay = server.hub.relay!; + expect(relay.buffer.bytesToViewers, greaterThan(0), reason: 'alice played through the relay'); + expect(relay.sourceToken, bob.sync.token); + + // The next swap retires bob's source and the buffer. + alice.swapPicked(const LinkMedia('http://127.0.0.1:1/none.mkv')); + await _poll(() => !bob.relaying.peek(), 'source stopped'); + expect(server.hub.relay, isNull); + }); + + test('links: through me, and fetched by the server; never broadcast', timeout: const Timeout(Duration(minutes: 3)), + () async { + final (origin, hits) = await _origin(await _media()); + try { + final alice = await member('alice'); + final bob = await member('bob'); + final link = 'http://127.0.0.1:${origin.port}/private/ep.mkv?api_key=SECRET'; + + expect(await bob.swapPicked(LinkMedia(link, relay: LinkRelay.me)), isNull); + await _poll(() => stateMediaRelay(alice.serverState.peek()!)?.source == bob.sync.memberId, 'me state'); + expect(alice.serverState.peek()!.toJson().toString(), isNot(contains('SECRET'))); + await playTogether([alice, bob]); + expect(server.hub.relay!.buffer.bytesToViewers, greaterThan(0)); + + final before = hits.length; + expect(await alice.swapPicked(LinkMedia(link, relay: LinkRelay.server)), isNull); + await _poll(() => stateMediaRelay(bob.serverState.peek()!)?.fetch == true, 'fetch state'); + expect(bob.serverState.peek()!.toJson().toString(), isNot(contains('SECRET'))); + expect(alice.relaying.peek(), isFalse, reason: 'the server fetches, nobody sources'); + await playTogether([alice, bob]); + expect(hits.length, greaterThan(before)); + + // A page instead of a file is refused up front. + final page = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + page.listen((r) { + r.response.headers.contentType = ContentType.html; + r.response.write(''); + r.response.close(); + }); + expect(await alice.swapPicked(LinkMedia('http://127.0.0.1:${page.port}/watch?v=x', relay: LinkRelay.me)), + contains('cannot pass this link on')); + await page.close(force: true); + } finally { + await origin.close(force: true); + } + }); + + test('the source drops uncleanly and resumes: it keeps its copy and keeps serving', + timeout: const Timeout(Duration(minutes: 2)), () async { + final file = await _media(); + final proxy = await _Blackhole.start(server.boundPort); + try { + final alice = await member('alice'); + final acc = JellyfinAccounts(store: MemoryJellyfinStore()); + await acc.ensureLoaded(validate: false); + final bob = SessionCoordinator(jellyfin: acc, trackPrefs: TrackPrefs(store: MemoryJellyfinStore())); + all.add(bob); + await bob.connect(ConnectParams(host: '127.0.0.1', port: proxy.port, secret: 'pw', name: 'bob')); + await bob.playback.player.setVolume(0); + expect(await bob.swapPicked(FileMedia(file)), isNull); + await _poll(() => server.hub.relay?.live == true, 'source live'); + await _poll(() => alice.playback.durationSecs > 19 && bob.playback.durationSecs > 19, 'both opened'); + final token = bob.sync.token; + + // Bob's network goes silent (no FIN): states stop, he reconnects + // with `resume`, the server still has him. + proxy.cut(); + await _poll(() => bob.reconnecting.peek(), 'reconnect engaged', timeout: const Duration(seconds: 15)); + await _poll(() => !bob.reconnecting.peek() && bob.serverState.peek() != null, 'back', + timeout: const Duration(seconds: 20)); + expect(bob.sync.token, token, reason: 'resumed as the same member'); + await _poll(() => server.hub.relay?.live == true, 'source redialled', timeout: const Duration(seconds: 15)); + expect(bob.relaying.peek(), isTrue); + expect(bob.playback.durationSecs, greaterThan(19), reason: 'still his own copy'); + + // Alice can still get bytes the buffer never had. + alice.seekTo(17); + await _poll(() => (alice.playback.positionSecs - 17).abs() < 1.5, 'alice seeked'); + } finally { + await proxy.close(); + } + }); +} + +/// TCP proxy whose [cut] silently stops forwarding on every connection made +/// so far (no FIN, like a dead Wi-Fi); later connections work. +class _Blackhole { + final ServerSocket _server; + final _conns = <(Socket, Socket, List)>[]; + _Blackhole(this._server); + int get port => _server.port; + + static Future<_Blackhole> start(int target) async { + final b = _Blackhole(await ServerSocket.bind(InternetAddress.loopbackIPv4, 0)); + b._server.listen((c) async { + final up = await Socket.connect('127.0.0.1', target); + final dead = [false]; + b._conns.add((c, up, dead)); + c.listen((d) { + if (!dead[0]) up.add(d); + }, onDone: () => dead[0] ? null : up.destroy(), onError: (_) {}); + up.listen((d) { + if (!dead[0]) c.add(d); + }, onDone: () => dead[0] ? null : c.destroy(), onError: (_) {}); + }); + return b; + } + + void cut() { + for (final (_, _, dead) in _conns) { + dead[0] = true; + } + } + + Future close() async { + await _server.close(); + for (final (a, b, _) in _conns) { + a.destroy(); + b.destroy(); + } + } +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 888a0b9..e3e9eb4 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -8,7 +8,7 @@ Plain overview first: [HOW-IT-WORKS.md](HOW-IT-WORKS.md). Wire format: | Path | What | Rule | |---|---|---| | `packages/core` | protocol messages, sync engine (`SyncEngine`), constants, config, Range parsing, logging, Jellyfin client | pure Dart, no Flutter | -| `packages/server` | Relic HTTP + WebSocket server (`SyncServer`, `ServerHub`, `Session`, media endpoint, Jellyfin relay), the `serve()` entrypoint, `bin/server.dart` (dedicated server) | pure Dart, no Flutter | +| `packages/server` | Relic HTTP + WebSocket server (`SyncServer`, `ServerHub`, `Session`, media endpoint, relay buffer, Jellyfin relay), the `serve()` entrypoint, `bin/server.dart` (dedicated server) | pure Dart, no Flutter | | `apps/desktop` | Flutter app: UI, `SyncClient`, `SessionCoordinator`, `PlaybackController` (media_kit/mpv), `ServerIsolate`, updater | ships with `watch-together-server` | | `tool` | harnesses: `range_logger`, `shaper`, `fake_viewer`, `e2e_check` | dev only | | `deploy` | systemd unit for the dedicated server | | @@ -144,7 +144,8 @@ Additive wire changes only; bump `proto` and PROTOCOL.md in the same commit. ## Media -Three sources, same sync. +Three sources, same sync. Any of them can also go through the relay +(below). - **Link**: the media field holds an http(s) URL. It travels as `media.url` in `swap` and `state`, and as `mediaUrl` in `welcome`. Each app plays it @@ -166,8 +167,8 @@ Three sources, same sync. - **Jellyfin item**: the media is a ref (`{serverId, itemId, ...}`, wire in [PROTOCOL.md](PROTOCOL.md) "Jellyfin media"); the server never stores or sends a URL or token for it. Each app resolves it with its own account and - direct-plays the static stream; guests use the relay (see "Jellyfin - relay"). + direct-plays the static stream; guests use the server's own login when + it has one ("Jellyfin relay"), else the member who put it up ("Relay"). Range table (`resolveMediaRequest`, tested byte-exact): @@ -362,6 +363,86 @@ errors leave it out). subscription does). - No progress reporting in the server: watched state is per client. +## Relay + +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`, +`relay_routes.dart`; wire in [PROTOCOL.md](PROTOCOL.md) "Relay"). + +What gets relayed (`SessionCoordinator.swapPicked`): + +| Put up | Others get it | +|---|---| +| File, by the host | `/m` from the hosting app (no relay) | +| File, by anyone else | relayed from that member's computer | +| Link, "Everyone fetches it" (default) | each app fetches it | +| Link, "Server fetches it" | the server reads it once into the buffer; the link is never broadcast; refused on an open server (it would fetch any address it can reach for anyone) | +| Link, "Through me" | relayed from that member's computer; for links only they can open | +| Jellyfin item (browser pick or pasted link) | own account first, then the server's login (`/j`), then relayed from the member who put it up (with their sign-in, or the pasted link's key, which stays on their machine) | + +Relayed links must be one file the relay can range-read: a `text/*` +response (YouTube and other pages, which only the player resolves) is +refused before the swap. + +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`, + `FileRelayBytes`, `HttpRelayBytes`; a Jellyfin source sends its own + `Authorization` header from this machine). Viewers only ever talk HTTP + to the server, so the source never sees mpv's request pattern. +- The server keeps one `RelayBuffer` per relayed media: 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 + same as each viewer's mpv buffer (`PlaybackController`), so a skip + inside the viewers' buffer window is served from memory. Evicted by + distance from the viewers, never LRU. With 2+ viewers the source sends + each byte about once. +- 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 + 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 + viewer's newest request moves its position or asks the source for + anything; older ones still get bytes that are already buffered. +- Flow control on the source (`RelayWindow`): one in-flight budget per + connection, 2 x delivery rate x min RTT (rate max-filtered over 2 s), + clamped 256 KiB to 8 MiB; the server acks bytes as they arrive. A + cancelled pull's unacked bytes are written off, and a cancel that lands + while a frame is being read stops it before it is sent. +- Cleanup: the buffer is memory only. Any swap disposes it (its URLs then + 404, the source is closed with 4001 and stops), and so does server + 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 file size drops the buffer. + `state.media.relay.live` follows the channel, and viewers show "waiting + for to pass the video on" while it is down. + +Measured (mpv 0.41 viewers, real server and source, a shaped link +between source and server; `tool/relay_e2e.dart`, 2-minute 1080p test +file averaging 10 Mbit/s with a 15 Mbit/s busy section): + +| Source uplink | Open | Seeks (incl. +90 s skip) | Stalls in 50 s | +|---|---|---|---| +| 400 Mbit/s, 20 ms | 0.8 s | 0.01 to 0.03 s | 0 | +| 40 Mbit/s, 40 ms | 2.1 s | 1.1 s first, then 0.03 s | 0 | +| 20 Mbit/s, 150 ms | 5.4 s | 1.1 to 5.3 s | 3 | +| 12 Mbit/s, 40 ms | 5.4 s | 0.5 to 2.9 s | 3 | + +3 viewers skipping together on 40 Mbit/s: no stalls, the source sent +101 MiB for 287 MiB served. A late joiner opened in 0.04 s from the +buffer. A 5 s source outage: no stall; 30 s: one stall once the buffer ran +out, then it recovered. Freezes of 0.3 to 0.8 s in 30% of seconds: no +stall. So the source's upload must beat the media's average bitrate +(peaks are covered by read-ahead once it has some lead), and comfortably +so for quick seeks. Rejected along the way: a fixed per-request window +(seeks of 14 s on slow links, data for abandoned positions clogging the +pipe), one shared window across requests (deadlocked on mpv's unread +requests), Dart `http2` (connection window stuck at 64 KiB). + ## Jellyfin in a session Client side (`lib/src/jellyfin/session_media.dart`, `progress.dart`, @@ -374,8 +455,9 @@ Client side (`lib/src/jellyfin/session_media.dart`, `progress.dart`, MediaBrowser` with this install's fields), not in the URL: `OwnPlay`, `report.jf {serverId: true}`. Else, when `state.relay`, `welcome.relayUrl` or `/j/default//` on the dialled host: `RelayPlay`, - `jf: false`. Else `BlockedPlay`: the session shows "This needs a - Jellyfin account on (or the host can relay)" with a button that + `jf: false`. Else, when `media.relay` names a source, the `/r/` URL: + `RelayPlay`. Else `BlockedPlay`: the session shows "No Jellyfin account + here for , and nobody is passing it on" with a button that opens Integrations over the session; linking (or reconnecting) the account retries by itself. A 401 marks the account expired. The member list shows "via host" for members whose `jf` is false for the current @@ -577,6 +659,13 @@ this repo, written by `tool/release.sh`. - Plain `ws://`, no TLS: friends use LAN, VPN or a forwarded port; TLS (e.g. a reverse proxy) is a documentation item. - One session per server, id `default`: the map is keyed for later but unused. -- No server storage: config and logs only; media never touches the server disk. +- No server storage: config and logs only; media never touches the server + disk. The relay buffer is memory only and sized like a viewer's buffer + (about 320 MiB at most), so a small VPS can carry it. +- The relay is the source's own connection through the server, not a + shared token: a member's Jellyfin sign-in or a link's key never leaves + their machine. Prior art (Plex Relay, Tailscale DERP) relays as a + bandwidth-capped fallback on someone else's servers; here the server is + the owner's, so it carries full quality. - Avatars ride in the 1 Hz `state` broadcast (cap 256 KiB base64 per avatar): simple, costs host uplink; hash-based serving is on the roadmap. diff --git a/docs/HOW-IT-WORKS.md b/docs/HOW-IT-WORKS.md index 46d64c2..26a53f8 100644 --- a/docs/HOW-IT-WORKS.md +++ b/docs/HOW-IT-WORKS.md @@ -19,8 +19,11 @@ A five-minute tour. Details: [ARCHITECTURE.md](ARCHITECTURE.md), wire format: video bytes come from somewhere else: - a link (any http/https URL): each app fetches it directly - a Jellyfin item: each app plays it with its own Jellyfin account, - friends without one through the server (/j/... endpoint) + friends without one through the server - a file on the host's disk: the server streams it (/m/... endpoint) + - the relay (/r/...): a member's computer passes a file, a link or a + Jellyfin item on to the server, which keeps a few minutes in memory + and streams it to everyone who needs it ``` - **App**: the desktop window. It draws the video with mpv, shows who is @@ -31,7 +34,8 @@ A five-minute tour. Details: [ARCHITECTURE.md](ARCHITECTURE.md), wire format: - **Jellyfin** (or any web host): where the video lives. For a Jellyfin item the server only knows which item it is; each app signs in with its own account and reports what it watched. For friends without an account - the server fetches the item with the host's sign-in and passes it on. + the member who put it up passes it on through the server (their sign-in + stays on their computer). ## Ways to run diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index cea4ef0..64f5b98 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -1,4 +1,4 @@ -# Protocol (proto = 7) +# Protocol (proto = 8) Wire format only. Behavior (gate, clock, timeouts) is in [ARCHITECTURE.md](ARCHITECTURE.md). Source of truth: @@ -22,7 +22,8 @@ with different `proto` refuse each other at `hello` (`proto mismatch`). `uptimeSecs`; unauthenticated). `GET /m///` serves the host's file (Range table in ARCHITECTURE.md). `GET /j///` relays the current Jellyfin media (see - "Jellyfin media" below). + "Jellyfin media" below). `/r///...` is the relay of a + member's media (see "Relay" below). ## Client to server @@ -35,6 +36,7 @@ with different `proto` refuse each other at `hello` (`proto mismatch`). {"t":"seek","position":123.4,"seq":3} {"t":"swap","media":{"name":"","size":0,"duration":0,"url":"..."}} {"t":"swap","media":{"name":"","size":0,"duration":0,"jellyfin":{}}} +{"t":"swap","media":{"name":"","size":123,"duration":0,"url":"...","relay":{}}} {"t":"report","position":123.1,"buffered":6.2,"eventId":41, "duration":1440.5,"mediaId":2,"jf":{"":true}} {"t":"ping","c":1767225600000} @@ -71,7 +73,7 @@ Field notes: - `chat.text` is cut to 500 characters. - `swap.media.url` is the link; `name`, `size`, `duration` are informational. `swap.media.jellyfin` makes it Jellyfin media (see below); then `url` is - dropped. + dropped. `swap.media.relay` makes it relayed media (see "Relay"). - Frames before `welcome` (other than `hello`) are ignored; unparsable or unknown frames are ignored silently. @@ -104,7 +106,8 @@ Field notes: `playing`, on the server's clock (see `pong`). `finished`: the media reached its end (`position` = duration); it stays until a seek or swap. `eventId` increases on seek, swap and transport changes. - `mediaId` increases on swap. `media.duration` is 0 until a client reports + `mediaId` increases on swap. `media.relay` is present for relayed media + (see "Relay"). `media.duration` is 0 until a client reports it. `seekBy` (member id) and `seekSeq` are present only while the current `eventId` came from a seek. `relay` (omitted when false): the server can relay the current Jellyfin media. `members[].jf` is the member's last @@ -152,6 +155,55 @@ Field notes: credential. Bytes count toward `bytesServed`. - Pasted Jellyfin links sent as plain `url` stay plain links. +## Relay + +Media whose bytes go through the server's buffer: a file on a member's +computer, a link passed on by a member, a link the server fetches itself, +or a Jellyfin item for viewers without an account. Behavior (buffer, +flow control, cases) is in [ARCHITECTURE.md](ARCHITECTURE.md), "Relay". + +``` +swap media.relay = {} the swapper's machine serves the bytes + media.relay = {"fetch":true} the server reads media.url itself +state media.relay = {"source":"m2","live":true} or {"fetch":true,"live":true} +``` + +- `swap`: `relay` with `jellyfin` keeps the ref (the swapper serves it to + viewers without an account; `fetch` is ignored there). `relay` without + `jellyfin`: no URL is stored or broadcast; `fetch` needs an http(s) + `url` (else `relay fetch needs an http(s) url`), which the server keeps + to itself, and a server with a password (else `server fetch needs a + server password`: an open server would fetch any address it can reach + for anyone). Any later swap ends the relay. +- `state.media.relay.source`: member id of the source (absent with + `fetch`). `live` (omitted when false): bytes can flow (the source's + channel is connected, or the server could read the link). +- `welcome.mediaUrl` for relayed media without `jellyfin`: this member's + `/r/` URL; absent for the source itself (it plays its own copy). Clients build it after a swap like the `/m` one: + `http:///r/default///` (name URL-encoded). +- `GET /r////`: the media, Range table as for + `/m`. Errors: 401 bad token, 403 kicked, 404 not the current relayed + media (a player still on an older `mediaId` gets 404, never the new + bytes), 410 unknown session, 503 + `retry-after` the source never + connected (60 s). A 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 + closed with code 4001 (the source stops instead of reconnecting). + Frames: + +``` +source -> server {"t":"info","size":} first frame + 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,"start":0,"end":1048575} inclusive + {"t":"ack","id":1,"n":65536} bytes arrived + {"t":"cancel","id":1} stop that pull +``` + + A `size` different from the one before (a reconnect) drops the buffer. + ## Join and reconnect `hello` then `welcome`, then the app opens `mediaUrl`, seeks to the server @@ -164,6 +216,3 @@ drop), `resume` takes it over and the token and media URL stay the same. ## Reserved, not implemented Server-held queue (`queue-add`, `queue-skip`) would auto-advance like a swap. -`RelayConstants.chunkBytes`, `creditWindowChunks`, `maxConcurrentOpen` and the -`open`, `ack`, `cancel`, `eof`, `buffered`, `state`, `event` kinds in -`MessageKind` are unused leftovers (the `state` frame does not use the enum). diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 57cc3fd..dd74d33 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -6,16 +6,25 @@ off by deleting it in the same commit as the code. Wire changes also edit ## Backlog -1. **Windows: first real install.** CI builds and publishes +1. **Windows: first real use.** CI builds and publishes `watch-together--windows-x64.zip` (libmpv bundled) and runs the - updater tests on NTFS, but nobody has launched the app on Windows yet. - Check: playback, fullscreen controls, hosting (firewall prompt), in-app - update from one release to the next. The rest of the desktop tests are - not in CI. + updater tests on NTFS; a friend has joined from it. Still unchecked: + fullscreen controls, hosting (firewall prompt), in-app update from one + release to the next. The rest of the desktop tests are not in CI. -2. **Jellyfin** (epic, below). +2. **Relay follow-ups** (see ARCHITECTURE "Relay"). First real use: owner's + PC -> VPS -> friend, Jellyfin and a file. Then: + - Source takeover: when the source leaves, another member with an + account on that Jellyfin (or the same file: name, size, hash) takes + over instead of the buffer running dry. + - "Bob can't open this link — relay it?": offer the relay to whoever put + a direct link up when a viewer's player fails to open it. + - Show who is relaying and how fast in the member list; warn the source + when its upload is below the media's bitrate. -3. **Smaller items** +3. **Jellyfin** (epic, below). + +4. **Smaller items** - Video output robustness: if media_kit's hardware path regresses to software (see ARCHITECTURE, Media: render mode), vendor a media_kit_video fix that builds the EGL display from GDK's native diff --git a/packages/core/lib/core.dart b/packages/core/lib/core.dart index 099a51e..658950b 100644 --- a/packages/core/lib/core.dart +++ b/packages/core/lib/core.dart @@ -4,6 +4,7 @@ export 'src/logging.dart'; export 'src/clock_sync.dart'; export 'src/config.dart'; export 'src/protocol.dart'; +export 'src/relay.dart'; export 'src/media_source.dart'; export 'src/sync_engine.dart'; export 'src/jellyfin/client.dart'; diff --git a/packages/core/lib/src/constants.dart b/packages/core/lib/src/constants.dart index 0a920f4..aa25572 100644 --- a/packages/core/lib/src/constants.dart +++ b/packages/core/lib/src/constants.dart @@ -1,9 +1,22 @@ abstract class RelayConstants { - static const int protoVersion = 7; + static const int protoVersion = 8; - static const int chunkBytes = 256 * 1024; - static const int creditWindowChunks = 8; - static const int maxConcurrentOpen = 16; + // Relay (`/r/...`): a member's bytes through the server's buffer. + static const int relaySliceBytes = 1 << 20; + static const int relayFrameBytes = 64 * 1024; + /// Defaults for the buffer; they mirror a viewer's mpv buffer (150 s / + /// 256 MiB ahead, 64 MiB back), so whatever a viewer may seek to soon is + /// already on the server. + static const int relayAheadBytes = 256 << 20; + static const int relayBackBytes = 64 << 20; + /// Kept regardless of position: containers keep their index at the start + /// or the end, and players read both on open and on seeks. + static const int relayPinBytes = 2 << 20; + static const int relayMinWindowBytes = 256 * 1024; + static const int relayMaxWindowBytes = 8 << 20; + /// A viewer request waits this long for its bytes (source away) before + /// the server gives up on it. + static const int relayWaitSecs = 60; static const int seekDebounceMs = 300; static const double driftSeekThresholdSecs = 5.0; diff --git a/packages/core/lib/src/protocol.dart b/packages/core/lib/src/protocol.dart index 7e1c1b2..428835e 100644 --- a/packages/core/lib/src/protocol.dart +++ b/packages/core/lib/src/protocol.dart @@ -1,22 +1,6 @@ import 'constants.dart'; import 'errors.dart'; -enum MessageKind { - hello, - welcome, - state, - event, - chat, - buffered, - kicked, - kick, - eof, - err, - open, - ack, - cancel, -} - /// Reads an optional JSON number as a finite double. JSON like 1e400 decodes to /// Infinity, which would later break jsonEncode on every broadcast. double finiteDouble(Object? value, String field, {double fallback = 0.0}) { @@ -111,11 +95,23 @@ class HelloMedia { final String? url; /// Set for Jellyfin media; then there is no [url] (the server drops one). final JellyfinRef? jellyfin; + /// Relayed media (bytes from the source member through the server). + final MediaRelay? relay; const HelloMedia( - {required this.name, required this.size, required this.duration, this.url, this.jellyfin}); + {required this.name, + required this.size, + required this.duration, + this.url, + this.jellyfin, + this.relay}); - HelloMedia copyWith({double? duration}) => HelloMedia( - name: name, size: size, duration: duration ?? this.duration, url: url, jellyfin: jellyfin); + HelloMedia copyWith({double? duration, int? size, MediaRelay? relay}) => HelloMedia( + name: name, + size: size ?? this.size, + duration: duration ?? this.duration, + url: url, + jellyfin: jellyfin, + relay: relay ?? this.relay); Map toJson() => { 'name': name, @@ -123,20 +119,48 @@ class HelloMedia { 'duration': duration, if (url != null) 'url': url, if (jellyfin != null) 'jellyfin': jellyfin!.toJson(), + if (relay != null) 'relay': relay!.toJson(), }; factory HelloMedia.fromJson(Map json) { final jf = json['jellyfin']; + final relay = json['relay']; return HelloMedia( name: (json['name'] as String?) ?? '', size: finiteInt(json['size'], 'size', fallback: 0), duration: finiteDouble(json['duration'], 'duration'), url: json['url'] as String?, jellyfin: jf is Map ? JellyfinRef.fromJson(Map.from(jf)) : null, + relay: relay is Map ? MediaRelay.fromJson(Map.from(relay)) : null, ); } } +/// 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]). +class MediaRelay { + final String source; + final bool fetch; + final bool live; + const MediaRelay({this.source = '', this.fetch = false, this.live = false}); + + MediaRelay copyWith({bool? live}) => MediaRelay(source: source, fetch: fetch, live: live ?? this.live); + + Map toJson() => { + if (source.isNotEmpty) 'source': source, + if (fetch) 'fetch': true, + if (live) 'live': true, + }; + + factory MediaRelay.fromJson(Map json) => MediaRelay( + source: json['source'] is String ? json['source'] as String : '', + fetch: json['fetch'] == true, + live: json['live'] == true); +} + class HelloMessage { final int proto; final String secret; diff --git a/packages/core/lib/src/relay.dart b/packages/core/lib/src/relay.dart new file mode 100644 index 0000000..c383181 --- /dev/null +++ b/packages/core/lib/src/relay.dart @@ -0,0 +1,354 @@ +import 'dart:async'; +import 'dart:collection'; +import 'dart:convert'; +import 'dart:io'; +import 'dart:math'; +import 'dart:typed_data'; + +import 'constants.dart'; +import 'logging.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. +// +// source -> server {"t":"info","size":N} first frame +// 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,"start":0,"end":N} bytes start..end +// {"t":"ack","id":1,"n":65536} n bytes arrived +// {"t":"cancel","id":1} stop this pull +// +// Flow control is on the source: one adaptive in-flight budget per +// connection (see [RelayWindow]); a cancelled pull's unacked bytes are +// written off. + +final _log = createLogger('relay'); + +/// Binary relay frame: 4-byte big-endian pull id, then payload. +Uint8List encodeRelayFrame(int id, List payload) { + final out = Uint8List(4 + payload.length); + ByteData.sublistView(out).setUint32(0, id); + out.setRange(4, out.length, payload); + return out; +} + +(int id, Uint8List payload) decodeRelayFrame(Uint8List frame) { + if (frame.length < 4) throw const FormatException('short relay frame'); + return (ByteData.sublistView(frame, 0, 4).getUint32(0), Uint8List.sublistView(frame, 4)); +} + +/// Where a source reads the media from. +abstract interface class RelayBytes { + /// Total length in bytes. + Future size(); + + /// Bytes [start]..[end] (inclusive), in order. Cancelling the + /// subscription stops reading. + Stream> read(int start, int end); + + Future close(); +} + +class FileRelayBytes implements RelayBytes { + final File file; + FileRelayBytes(this.file); + + @override + Future size() => file.length(); + + @override + Stream> read(int start, int end) => file.openRead(start, end + 1); + + @override + Future close() async {} +} + +/// An HTTP(S) resource with Range support, e.g. a link or a Jellyfin +/// stream; [headers] (auth) never leave this machine. +class HttpRelayBytes implements RelayBytes { + final Uri url; + final Map headers; + final HttpClient _client; + int? _size; + HttpRelayBytes(this.url, {this.headers = const {}, HttpClient? client}) + : _client = client ?? HttpClient(); + + Future _get(String range) async { + final req = await _client.getUrl(url).timeout(const Duration(seconds: 20)); + headers.forEach(req.headers.set); + req.headers + ..set(HttpHeaders.rangeHeader, range) + ..set(HttpHeaders.acceptEncodingHeader, 'identity'); + final res = await req.close().timeout(const Duration(seconds: 20)); + if (res.statusCode != 200 && res.statusCode != 206) { + await res.drain().catchError((_) {}); + throw HttpException('upstream HTTP ${res.statusCode}', uri: url); + } + return res; + } + + @override + Future size() async { + if (_size != null) return _size!; + final res = await _get('bytes=0-0'); + await res.drain().catchError((_) {}); + // A web page (YouTube and co.) only a player can resolve, not a file. + final type = res.headers.contentType?.mimeType ?? ''; + if (type.startsWith('text/')) throw HttpException('not a video file ($type)', uri: url); + // `bytes 0-0/12345`; a server ignoring Range answers 200 with the length. + final total = res.headers.value(HttpHeaders.contentRangeHeader)?.split('/').last; + final n = int.tryParse(total ?? '') ?? (res.statusCode == 200 ? res.contentLength : -1); + if (n <= 0) throw HttpException('upstream size unknown', uri: url); + return _size = n; + } + + @override + Stream> read(int start, int end) async* { + final res = await _get('bytes=$start-$end'); + if (res.statusCode == 200 && start > 0) { + throw HttpException('upstream ignores Range', uri: url); + } + yield* res; + } + + @override + Future close() async => _client.close(force: true); +} + +/// The per-connection in-flight budget: 2 x (delivery rate x min RTT), +/// clamped to [RelayConstants.relayMinWindowBytes]..[RelayConstants.relayMaxWindowBytes]. +/// The rate is the max of the last 2 s of samples, so a pause or a burst of +/// cancels does not collapse it (BBR's bandwidth max-filter); spike results +/// in docs/ARCHITECTURE.md ("Relay"). +class RelayWindow { + var window = RelayConstants.relayMinWindowBytes; + var inflight = 0; + var minRttUs = 1 << 40; + var _sent = 0, _acked = 0, _writtenOff = 0; + final _sw = Stopwatch()..start(); + final _sendTimes = Queue<(int, int)>(); // (cumulative sent, time us) + final _acks = Queue<(int, int)>(); // (time us, cumulative acked) + final _rates = Queue<(int, double)>(); // (time us, bytes/s) + Completer? _room; + + bool get full => inflight >= window; + + void sent(int n) { + _sent += n; + inflight += n; + _sendTimes.add((_sent, _sw.elapsedMicroseconds)); + } + + void acked(int n) { + inflight -= n; + _acked += n; + final now = _sw.elapsedMicroseconds; + while (_sendTimes.isNotEmpty && _sendTimes.first.$1 <= _acked + _writtenOff) { + minRttUs = min(minRttUs, now - _sendTimes.removeFirst().$2); + } + _acks.add((now, _acked)); + while (_acks.length > 2 && now - _acks.first.$1 > max(2 * minRttUs, 200000)) { + _acks.removeFirst(); + } + if (_acks.length >= 2 && minRttUs < (1 << 40)) { + final dt = (now - _acks.first.$1) / 1e6; + if (dt > 0.02) { + _rates.add((now, (_acked - _acks.first.$2) / dt)); + while (now - _rates.first.$1 > 2000000) { + _rates.removeFirst(); + } + final rate = _rates.map((r) => r.$2).reduce(max); + window = (2 * rate * minRttUs / 1e6) + .clamp(RelayConstants.relayMinWindowBytes, RelayConstants.relayMaxWindowBytes) + .toInt(); + } + } + _wake(); + } + + /// Bytes of a cancelled pull that will never be acked. + void writeOff(int n) { + inflight -= n; + _writtenOff += n; + _wake(); + } + + void _wake() { + final r = _room; + _room = null; + r?.complete(); + } + + Future room() async { + while (full) { + await (_room ??= Completer()).future; + } + } +} + +class _Pull { + final int id; + var sent = 0, acked = 0; + var cancelled = false, finished = false; + StreamIterator>? it; + _Pull(this.id); + bool get settled => cancelled || (finished && acked >= sent); +} + +/// Close code the server uses for "you are not (or no longer) the source of +/// the current media": the source stops instead of reconnecting. +const relayCloseNotSource = 4001; + +/// Serves [bytes] on the relay channel at [url] (`ws://host/r///source`) +/// until [close]. Reconnects with backoff when the channel drops; stops for +/// good when the server closes it with [relayCloseNotSource]. +class RelaySource { + final Uri url; + final RelayBytes bytes; + final Duration Function(int attempt) backoff; + RelaySource(this.url, this.bytes, {Duration Function(int attempt)? backoff}) + : backoff = backoff ?? ((a) => Duration(milliseconds: min(8000, 500 * (1 << min(a, 4))))); + + final window = RelayWindow(); + final _pulls = {}; + WebSocket? _ws; + var _closed = false; + final _done = Completer(); + + /// Completes when the source stopped for good (closed or refused). + Future get done => _done.future; + + /// Bytes sent to the server, for stats. + int bytesSent = 0; + + Future start() async { + final size = await bytes.size(); + unawaited(_loop(size)); + } + + Future _loop(int size) async { + var attempt = 0; + while (!_closed) { + try { + final ws = await WebSocket.connect(url.toString()).timeout(const Duration(seconds: 15)); + if (_closed) { + await ws.close(); + break; + } + _ws = ws; + attempt = 0; + ws.add(jsonEncode({'t': 'info', 'size': size})); + await for (final m in ws) { + if (m is String) _onText(ws, m); + } + if (ws.closeCode == relayCloseNotSource) { + _log.info('relay channel: not the source any more, stopping'); + _closed = true; + } + } catch (e) { + _log.fine('relay channel: $e'); + } + _ws = null; + _abandonAll(); + if (_closed) break; + await Future.delayed(backoff(attempt++)); + } + _abandonAll(); + await bytes.close(); + if (!_done.isCompleted) _done.complete(); + } + + void _onText(WebSocket ws, String text) { + final Map j; + try { + j = jsonDecode(text) as Map; + } catch (_) { + return; + } + final id = j['id']; + if (id is! int) return; + switch (j['t']) { + case 'pull': + final start = j['start'], end = j['end']; + if (start is! int || end is! int || start < 0 || end < start) return; + final p = _Pull(id); + _pulls[id] = p; + unawaited(_serve(ws, p, start, end)); + case 'ack': + final p = _pulls[id]; + final n = j['n']; + if (p == null || p.cancelled || n is! int) return; + p.acked += n; + window.acked(n); + if (p.settled) _pulls.remove(id); + case 'cancel': + final p = _pulls.remove(id); + if (p == null) return; + _cancel(p); + } + } + + void _cancel(_Pull p) { + if (p.cancelled) return; + p.cancelled = true; + window.writeOff(p.sent - p.acked); + p.acked = p.sent; + unawaited(p.it?.cancel()); + } + + void _abandonAll() { + for (final p in _pulls.values.toList()) { + _cancel(p); + } + _pulls.clear(); + } + + Future _serve(WebSocket ws, _Pull p, int start, int end) async { + final it = p.it = StreamIterator(bytes.read(start, end)); + final pending = BytesBuilder(copy: false); + const frame = RelayConstants.relayFrameBytes; + try { + var more = true; + while (!p.cancelled) { + while (more && pending.length < frame) { + more = await it.moveNext(); + if (more) pending.add(it.current); + } + if (pending.isEmpty) break; + await window.room(); + // Check and send in one synchronous step: a cancel that lands + // during an await must not be followed by a sent frame. + if (p.cancelled) break; + final all = pending.takeBytes(); + final n = min(frame, all.length); + ws.add(encodeRelayFrame(p.id, Uint8List.sublistView(all, 0, n))); + if (n < all.length) pending.add(Uint8List.sublistView(all, n)); + p.sent += n; + bytesSent += n; + window.sent(n); + } + if (!p.cancelled) ws.add(jsonEncode({'t': 'end', 'id': p.id})); + } catch (e) { + if (!p.cancelled) { + _log.warning('relay read ${p.id} failed: $e'); + try { + ws.add(jsonEncode({'t': 'fail', 'id': p.id, 'reason': '$e'})); + } catch (_) {} + } + } finally { + p.finished = true; + if (p.settled) _pulls.remove(p.id); + await it.cancel(); + } + } + + /// Stops serving; [done] completes once the loop has wound down (a + /// pending reconnect delay runs out first). + Future close() async { + _closed = true; + await _ws?.close(); + } +} diff --git a/packages/server/lib/server.dart b/packages/server/lib/server.dart index ddefd60..44c3640 100644 --- a/packages/server/lib/server.dart +++ b/packages/server/lib/server.dart @@ -4,3 +4,5 @@ export 'src/server_hub.dart'; export 'src/session.dart'; export 'src/sync_server.dart'; export 'src/jellyfin_relay.dart'; +export 'src/relay_buffer.dart'; +export 'src/relay_routes.dart'; diff --git a/packages/server/lib/src/relay_buffer.dart b/packages/server/lib/src/relay_buffer.dart new file mode 100644 index 0000000..27c19ad --- /dev/null +++ b/packages/server/lib/src/relay_buffer.dart @@ -0,0 +1,408 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:math'; +import 'dart:typed_data'; + +import 'package:sync_relay_core/core.dart'; + +final _log = createLogger('relay'); + +/// What feeds a [RelayBuffer]: 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]. + void pull(int id, int start, int end); + 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). +class LocalRelayFeed extends RelayFeed { + final RelayBytes bytes; + final RelayBuffer buffer; + final _subs = >>{}; + LocalRelayFeed(this.bytes, this.buffer); + + @override + void pull(int id, int start, int end) { + _subs[id] = bytes.read(start, end).listen( + (c) => buffer.feedBytes(id, c is Uint8List ? c : Uint8List.fromList(c)), + onError: (Object e) { + _subs.remove(id)?.cancel(); + buffer.feedFail(id, '$e'); + }, + onDone: () { + _subs.remove(id); + buffer.feedEnd(id); + }, + cancelOnError: true, + ); + } + + @override + void cancel(int id) => _subs.remove(id)?.cancel(); + + Future close() async { + for (final s in _subs.values) { + await s.cancel(); + } + _subs.clear(); + await bytes.close(); + } +} + +/// A relay channel (`/r/.../source`) as a feed. +class ChannelRelayFeed extends RelayFeed { + final void Function(String text) send; + ChannelRelayFeed(this.send); + + @override + void pull(int id, int start, int end) => + send(jsonEncode({'t': 'pull', 'id': id, 'start': start, 'end': end})); + + @override + void cancel(int id) => send(jsonEncode({'t': 'cancel', 'id': id})); + + @override + void arrived(int id, int n) => send(jsonEncode({'t': 'ack', 'id': id, 'n': n})); +} + +class _Pull { + final int id, start, end; + int pos; + final buf = BytesBuilder(copy: false); + _Pull(this.id, this.start, this.end) : pos = start; +} + +class _Viewer { + int pos = 0; + int open = 0; // requests in flight + int newest = 0; // sequence of its newest request + DateTime lastActive = DateTime.now(); +} + +/// The server side of the relay for one media: an in-memory buffer of +/// [RelayConstants.relaySliceBytes] slices filled from one [RelayFeed] with +/// one sequential pull at a time, served to any number of viewers. +/// +/// - Viewers only ever talk HTTP to the server; the feed never sees their +/// request pattern (mpv leaves superseded requests open but unread, and +/// jumps to the file's end for the index). +/// - Kept: the file's first and last [pin] bytes, plus [back] behind the +/// slowest and [ahead] past the furthest viewer. Evicted by distance from +/// the viewers, never LRU. +/// - Filled only while someone watches: a viewer blocked on a slice comes +/// first (oldest wait first); then read-ahead. +/// - 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 { + final int ahead, back, pin; + final Duration waitLimit; + RelayBuffer({ + this.ahead = RelayConstants.relayAheadBytes, + this.back = RelayConstants.relayBackBytes, + this.pin = RelayConstants.relayPinBytes, + this.waitLimit = const Duration(seconds: RelayConstants.relayWaitSecs), + }); + + static const slice = RelayConstants.relaySliceBytes; + + RelayFeed? _feed; + int size = 0; + final _slices = {}; + final _waiters = >>{}; + final _blocked = {}; // slice -> wait start (us) + final _viewers = {}; + _Pull? _pull; + var _nextId = 1, _reqSeq = 0; + var _disposed = false; + Completer _kick = Completer(); + final Completer _ready = Completer(); + + /// Totals, for logs and tests. + int bytesFromFeed = 0, bytesToViewers = 0; + + bool get live => _feed != null && size > 0; + 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; + + /// Attaches a (re)connected feed for a file of [size] bytes. A different + /// size means different content: the buffer starts over. + void attach(RelayFeed feed, int size) { + if (_disposed) return; + if (this.size != 0 && this.size != size) { + _log.info('relay source changed size ${this.size} -> $size, dropping buffer'); + _slices.clear(); + } + _feed = feed; + this.size = size; + if (!_ready.isCompleted) _ready.complete(); + if (_fillerRunning) { + _wake(); + } else { + unawaited(_filler()); + } + } + + /// The feed went away; buffered slices keep serving. + void detach(RelayFeed feed) { + if (!identical(_feed, feed)) return; + _feed = null; + _pull = null; + _wake(); + } + + void dispose() { + _disposed = true; + _feed = null; + _pull = null; + _slices.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 + + 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; + var off = 0; + while (off < data.length) { + final s = pl.pos ~/ slice; + final take = min(_sliceEnd(s) + 1 - pl.pos, data.length - off); + if (take <= 0) { + // Past the end the feed claimed: protocol violation. + _endPull(pl); + return; + } + pl.buf.add(Uint8List.sublistView(data, off, off + take)); + off += take; + pl.pos += take; + if (pl.pos == _sliceEnd(s) + 1) { + _slices[s] = pl.buf.takeBytes(); + for (final c in _waiters.remove(s) ?? const >[]) { + c.complete(); + } + if (_shouldStop(pl, s)) { + _endPull(pl); + return; + } + } + } + } + + void feedEnd(int id) { + final pl = _pull; + if (pl != null && pl.id == id) { + _pull = null; + _wake(); + } + } + + void feedFail(int id, String reason) { + _log.warning('relay pull $id failed: $reason'); + feedEnd(id); + } + + // ------------------------------------------------------------- scheduling + + bool _shouldStop(_Pull pl, int done) { + final next = done + 1; + if (next >= _nSlices || next * slice > pl.end) return false; // ends naturally + // Someone is blocked elsewhere: hand the feed over at this boundary. + if (_blocked.keys.any((w) => w != next && w != next + 1 && !_slices.containsKey(w))) { + return true; + } + if (_slices.containsKey(next)) return true; + return !_wanted(next); + } + + void _endPull(_Pull pl) { + if (!identical(_pull, pl)) return; + _pull = null; + if (pl.pos <= pl.end) _feed?.cancel(pl.id); + _wake(); + } + + Iterable<_Viewer> get _active { + final now = DateTime.now(); + return _viewers.values + .where((v) => v.open > 0 || now.difference(v.lastActive) < const Duration(seconds: 60)); + } + + bool _pinned(int s) => s * slice < pin || s * slice >= size - pin; + + bool _wanted(int s) { + if (_pinned(s)) return true; + final pos = _active.map((v) => v.pos).toList(); + if (pos.isEmpty) return false; + final off = s * slice; + return off >= pos.reduce(min) - back && off <= pos.reduce(max) + ahead; + } + + int _distance(int s) { + if (_pinned(s)) return -1; + final off = s * slice; + var d = 1 << 62; + for (final v in _active) { + // Behind a viewer costs more than ahead (it was already watched). + d = min(d, off >= v.pos ? off - v.pos : (v.pos - off) * 4); + } + return d; + } + + void _evict() { + final limit = ahead + back + 2 * pin; + var total = _slices.length * slice; + if (total <= limit) return; + final order = _slices.keys.toList()..sort((a, b) => _distance(b).compareTo(_distance(a))); + for (final s in order) { + if (total <= limit || _distance(s) < 0) break; + if (_waiters.containsKey(s)) continue; + _slices.remove(s); + total -= slice; + } + } + + int? _nextMissing() { + final blocked = _blocked.entries.where((e) => !_slices.containsKey(e.key)).toList() + ..sort((a, b) => a.value.compareTo(b.value)); + if (blocked.isNotEmpty) return blocked.first.key; + final active = _active.toList(); + if (active.isEmpty) return null; // nobody watches: fetch nothing + for (var s = 0; s * slice < pin && s < _nSlices; s++) { + if (!_slices.containsKey(s)) return s; + } + for (var s = max(0, (size - pin) ~/ slice); s < _nSlices; s++) { + if (!_slices.containsKey(s)) return s; + } + final starts = active.map((v) => v.pos).toList()..sort(); + for (final p in starts) { + for (var s = p ~/ slice; s * slice <= p + ahead && s < _nSlices; s++) { + if (!_slices.containsKey(s)) return s; + } + } + return null; + } + + var _fillerRunning = false; + + Future _filler() async { + _fillerRunning = true; + try { + while (!_disposed) { + final feed = _feed; + if (feed == null || _pull != null) { + await _sleep(); + continue; + } + _evict(); + final s = _nextMissing(); + if (s == null) { + await _sleep(); + continue; + } + var e = s; + while (e + 1 < _nSlices && !_slices.containsKey(e + 1) && _wanted(e + 1)) { + e++; + } + final pl = _Pull(_nextId++, s * slice, _sliceEnd(e)); + _pull = pl; + feed.pull(pl.id, pl.start, pl.end); + } + } finally { + _fillerRunning = false; + } + } + + Future _sleep() async { + _kick = Completer(); + // Also wake periodically: viewers age out of [_active]. + await _kick.future.timeout(const Duration(seconds: 5), onTimeout: () {}); + } + + Future _waitSlice(int s, bool drives) async { + final deadline = DateTime.now().add(waitLimit); + while (!_slices.containsKey(s)) { + if (_disposed || DateTime.now().isAfter(deadline)) return false; + if (drives) { + _blocked.putIfAbsent(s, () => DateTime.now().microsecondsSinceEpoch); + // A pull nobody waits on gives way at once; one serving another + // blocked viewer hands over at its next slice boundary. + final pl = _pull; + final at = pl == null ? -1 : pl.pos ~/ slice; + if (pl != null && at != s && !_blocked.containsKey(at)) _endPull(pl); + _wake(); + } + final c = Completer(); + _waiters.putIfAbsent(s, () => []).add(c); + await c.future.timeout(const Duration(seconds: 2), onTimeout: () {}); + } + _blocked.remove(s); + return true; + } + + // ---------------------------------------------------------------- viewers + + /// Waits for the size (the feed's first connect), up to [waitLimit]. + Future ready() async { + if (size > 0) return true; + await _ready.future.timeout(waitLimit, onTimeout: () {}); + return size > 0 && !_disposed; + } + + /// Streams [start]..[end] to [viewer]. Cancelling the subscription (the + /// viewer went away) ends it; a slice that never arrives ends it with an + /// error after [waitLimit]. + Stream read(String viewer, int start, int end, {void Function(int n)? onSent}) async* { + final v = _viewers.putIfAbsent(viewer, _Viewer.new); + final tail = start >= size - pin; + final me = ++_reqSeq; + if (!tail) { + v.newest = me; + v.pos = start; + } + v.open++; + v.lastActive = DateTime.now(); + _wake(); + try { + var off = start; + while (off <= end) { + final s = off ~/ slice; + if (!await _waitSlice(s, tail || v.newest == me)) { + throw StateError('relay: bytes at $off did not arrive'); + } + final data = _slices[s]; + if (data == null) continue; // evicted between wake and read + final from = off - s * slice, to = min(data.length, end - s * slice + 1); + yield Uint8List.sublistView(data, from, to); + bytesToViewers += to - from; + onSent?.call(to - from); + off += to - from; + v.lastActive = DateTime.now(); + if (!tail && v.newest == me) v.pos = off; + _wake(); + } + } finally { + v.open--; + v.lastActive = DateTime.now(); + } + } +} diff --git a/packages/server/lib/src/relay_routes.dart b/packages/server/lib/src/relay_routes.dart new file mode 100644 index 0000000..e9679b0 --- /dev/null +++ b/packages/server/lib/src/relay_routes.dart @@ -0,0 +1,230 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:typed_data'; + +import 'package:relic/relic.dart'; +import 'package:sync_relay_core/core.dart'; +import 'package:web_socket/web_socket.dart'; + +import 'media_endpoint.dart'; +import 'relay_buffer.dart'; +import 'session.dart'; + +final _log = createLogger('relay'); + +/// The relay of the current media: its buffer and where the bytes come +/// from. One per relayed swap; [dispose] on the next swap or server stop +/// (the buffer is memory only, so that is the whole cleanup). +class ActiveRelay { + final int mediaId; + final String name; + /// Token of the member whose machine serves the bytes; null when the + /// server fetches the link itself ([local]). + final String? sourceToken; + final RelayBuffer buffer; + final LocalRelayFeed? local; + final void Function() onLiveChanged; + RelicWebSocket? _channel; + ChannelRelayFeed? _channelFeed; + + ActiveRelay._(this.mediaId, this.name, this.sourceToken, this.buffer, this.local, this.onLiveChanged); + + /// Bytes from member [sourceToken] over its relay channel. + factory ActiveRelay.fromMember(int mediaId, String name, String sourceToken, + {RelayBuffer? buffer, required void Function() onLiveChanged}) => + ActiveRelay._(mediaId, name, sourceToken, buffer ?? RelayBuffer(), null, onLiveChanged); + + /// The server reads [url] itself. + static ActiveRelay fetch(int mediaId, String name, Uri url, + {RelayBuffer? buffer, required void Function() onLiveChanged}) { + final b = buffer ?? RelayBuffer(); + final local = LocalRelayFeed(HttpRelayBytes(url), b); + final r = ActiveRelay._(mediaId, name, null, b, local, onLiveChanged); + unawaited(() async { + try { + final size = await local.bytes.size(); + if (r._disposed) return; + b.attach(local, size); + onLiveChanged(); + } catch (e) { + _log.warning('relay fetch: cannot read the link: $e'); + } + }()); + return r; + } + + var _disposed = false; + bool get live => buffer.live; + + void dispose() { + if (_disposed) return; + _disposed = true; + buffer.dispose(); + unawaited(local?.close()); + final ch = _channel; + _channel = null; + if (ch != null) unawaited(_closeNotSource(ch)); + } + + /// Source member gone for good (kicked): stop its channel. + void dropChannel() { + final ch = _channel; + _channel = null; + if (ch != null) unawaited(_closeNotSource(ch)); + } + + void _attachChannel(RelicWebSocket ws, int size) { + final old = _channel; + if (old != null && !identical(old, ws)) unawaited(old.tryClose()); + _channel = ws; + final feed = _channelFeed = ChannelRelayFeed((t) => ws.trySendText(t)); + buffer.attach(feed, size); + onLiveChanged(); + } + + void _detachChannel(RelicWebSocket ws) { + if (!identical(_channel, ws)) return; + _channel = null; + final feed = _channelFeed; + _channelFeed = null; + if (feed != null) buffer.detach(feed); + onLiveChanged(); + } +} + +Future _closeNotSource(RelicWebSocket ws) async { + try { + await ws.close(relayCloseNotSource, 'not the source'); + } catch (_) { + await ws.tryClose(); + } +} + +/// `GET /r///source` (WebSocket): the source member's relay +/// channel. `GET /r////`: a viewer reads the +/// relayed media, Range as for `/m`. [relay] gives the current relay. +void mountRelayRoutes( + RelicApp app, + SessionMap sessions, + ActiveRelay? Function() relay, { + void Function(String token, MediaLog log)? onLog, +}) { + app.get('/r/:sid/:token/source', (final req) { + final sid = req.pathParameters.raw[#sid] ?? ''; + final token = req.pathParameters.raw[#token] ?? ''; + final session = sessions.get(sid); + final r = relay(); + final ok = session != null && + session.gate(token) == MediaGate.ok && + r != null && + r.sourceToken == token; + return WebSocketUpgrade((final ws) async { + if (!ok) { + await _closeNotSource(ws); + return; + } + final r = relay()!; + ws.pingInterval = Duration(seconds: RelayConstants.pingIntervalSecs); + var attached = false; + try { + await for (final e in ws.events) { + if (r._disposed) break; + if (e is TextDataReceived) { + final Map j; + try { + j = jsonDecode(e.text) as Map; + } catch (_) { + continue; + } + final id = j['id']; + switch (j['t']) { + case 'info': + final size = j['size']; + if (size is int && size > 0 && !attached) { + attached = true; + _log.info('relay channel up (${r.name}, $size bytes)'); + r._attachChannel(ws, size); + } + case 'end': + if (id is int) r.buffer.feedEnd(id); + case 'fail': + if (id is int) r.buffer.feedFail(id, '${j['reason']}'); + } + } else if (e is BinaryDataReceived && attached) { + try { + final (id, payload) = decodeRelayFrame(e.data); + r.buffer.feedBytes(id, payload); + } catch (_) {} + } + } + } catch (_) { + // channel broke + } finally { + r._detachChannel(ws); + _log.info('relay channel down (${r.name})'); + } + }); + }); + + app.get('/r/:sid/:token/:mediaId/:name', (final req) async { + final stopwatch = Stopwatch()..start(); + final sid = req.pathParameters.raw[#sid] ?? ''; + final token = req.pathParameters.raw[#token] ?? ''; + final mediaId = int.tryParse(req.pathParameters.raw[#mediaId] ?? ''); + final session = sessions.get(sid); + if (session == null) return _plain(410); + switch (session.gate(token)) { + case MediaGate.ok: + break; + case MediaGate.forbidden: + return _plain(403); + default: + return _plain(401); + } + final r = relay(); + // A player still on an older media must not get the new one's bytes. + if (r == null || r.mediaId != mediaId) return _plain(404); + if (!await r.buffer.ready()) { + return Response(503, + body: Body.fromString('relay source not connected'), + headers: Headers.build((h) => h['retry-after'] = ['2'])); + } + final size = r.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() ?? ''}'; + } + 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) { + return Response(res.status, body: Body.fromString('status ${res.status}'), headers: headers); + } + final end = res.end ?? size - 1; + final shortToken = token.length > 4 ? '…${token.substring(token.length - 4)}' : token; + onLog?.call( + token, + MediaLog( + method: req.method.name, + path: '/r/$sid/$shortToken/$mediaId/${r.name}', + rangeHeader: rangeHeader, + status: res.status, + bytesWritten: end - res.start + 1, + durationMs: stopwatch.elapsedMilliseconds, + )); + final Stream body = + r.buffer.read(token, res.start, end, onSent: (n) => session.addBytes(token, n)); + MimeType mime; + try { + mime = MimeType.parse(res.headers['content-type'] ?? ''); + } catch (_) { + mime = MimeType.octetStream; + } + return Response(res.status, + body: Body.fromDataStream(body, mimeType: mime, contentLength: end - res.start + 1), + headers: headers); + }); +} + +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 ed8f300..1c08031 100644 --- a/packages/server/lib/src/server_hub.dart +++ b/packages/server/lib/src/server_hub.dart @@ -5,6 +5,7 @@ import 'package:relic/relic.dart'; import 'package:sync_relay_core/core.dart'; import 'package:web_socket/web_socket.dart'; +import 'relay_routes.dart'; import 'session.dart'; final _hubLog = createLogger('server_hub'); @@ -32,6 +33,29 @@ class ServerHub { StateBroadcast stateBroadcast() => engine.broadcast(relay: relayAvailable); + /// The relay of the current media (`/r/...`), if it is relayed. + ActiveRelay? relay; + + /// Replaces the relay (null: the new media is not relayed). The old + /// buffer is dropped with it. + void setRelay(ActiveRelay? next) { + relay?.dispose(); + relay = next; + } + + /// `media.relay` for a relayed swap by [token]: who serves the bytes. + MediaRelay relayFor(String token, MediaRelay requested) => MediaRelay( + source: requested.fetch ? '' : (engine.members[token]?.id ?? ''), fetch: requested.fetch); + + /// The relay's channel came up or went down: `media.relay.live` follows. + void relayLiveChanged(ActiveRelay r) { + final m = engine.media; + if (!identical(relay, r) || m?.relay == null) return; + if (m!.relay!.live == r.live) return; + engine.media = m.copyWith(relay: m.relay!.copyWith(live: r.live)); + changed(); + } + bool isHostKey(String key) => hostKey.isNotEmpty && key == hostKey; /// The member a reconnecting client may take over with its old token: @@ -177,6 +201,7 @@ 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, String host)? relayMediaUrlBuilder, }) { app.get('/sync', (final req) { // Host the client dialled (Host header), so file URLs work for remote joiners. @@ -233,13 +258,17 @@ void mountSyncRoute( // By reference only: no URL (it could carry a token) is // stored or broadcast for Jellyfin media. final ref = sanitizeJellyfinRef(parsed.jellyfin!); + // Relayed for viewers without access: from the swapper's + // machine (its own sign-in never leaves it). + final relayReq = parsed.relay == null || parsed.relay!.fetch ? null : parsed.relay; final media = ref == null ? null : HelloMedia( name: sanitizeMediaName(parsed.name.isEmpty ? ref.title : parsed.name), size: parsed.size, duration: parsed.duration, - jellyfin: ref); + jellyfin: ref, + relay: relayReq == null ? null : hub.relayFor(token, relayReq)); if (media == null || !hub.engine.applySwap(token, media)) { wsErr(ws, 'bad swap media'); return; @@ -250,6 +279,46 @@ void mountSyncRoute( duration: media.duration, jellyfin: ref, mediaId: hub.engine.mediaId); + hub.setRelay(relayReq == null ? null : _memberRelay(hub, token, media.name)); + hub.changed(); + return; + } + final relayReq = parsed.relay; + if (relayReq != null) { + // Relayed file or link: no URL is stored or broadcast (a + // link may be private to the swapper; the server keeps a + // fetched one to itself). + final name = sanitizeMediaName(parsed.name.isEmpty ? 'media' : parsed.name); + final link = Uri.tryParse(parsed.url ?? ''); + if (relayReq.fetch && + (link == null || !(link.isScheme('http') || link.isScheme('https')) || link.host.isEmpty)) { + wsErr(ws, 'relay fetch needs an http(s) url'); + return; + } + // The server would fetch any address it can reach (its own + // network, a VPN) for whoever asks: only for members who + // know the password. + if (relayReq.fetch && session.secret.isEmpty) { + wsErr(ws, 'server fetch needs a server password'); + return; + } + final media = HelloMedia( + name: name, + size: parsed.size, + duration: parsed.duration, + relay: hub.relayFor(token, relayReq)); + if (!hub.engine.applySwap(token, media)) { + wsErr(ws, 'bad swap media'); + return; + } + session.media = SessionMedia( + name: name, size: parsed.size, duration: parsed.duration, mediaId: hub.engine.mediaId); + late final ActiveRelay active; + active = relayReq.fetch + ? ActiveRelay.fetch(hub.engine.mediaId, name, link!, + onLiveChanged: () => hub.relayLiveChanged(active)) + : _memberRelay(hub, token, name); + hub.setRelay(active); hub.changed(); return; } @@ -271,6 +340,7 @@ void mountSyncRoute( wsErr(ws, 'bad swap media'); return; } + hub.setRelay(null); if (matchesFile) { session.media = SessionMedia( name: current.name, @@ -406,6 +476,7 @@ void mountSyncRoute( final activeToken = token; hub.sockets[activeToken] = ws; session.tokens[activeToken]!.onKick = (reason) async { + if (hub.relay?.sourceToken == activeToken) hub.relay!.dropChannel(); ws.trySendText(jsonEncode(KickedMessage(reason: reason).toJson())); await Future.delayed(const Duration(milliseconds: 200)); await ws.tryClose(); @@ -413,10 +484,23 @@ void mountSyncRoute( }; final media = session.media; final relay = hub.relayAvailable; + final relayed = hub.relay; final welcome = WelcomeMessage( token: activeToken, memberId: memberId, - mediaUrl: media?.file != null ? urlBuilder('default', activeToken, media, dialledHost) : media?.url, + // Jellyfin media resolves per member (own account first); only + // relayed files and links start from the /r/ URL, and never for + // the source itself (it plays its own copy). + mediaUrl: relayed != null && + media != null && + media.jellyfin == null && + relayed.sourceToken != activeToken && + relayed.mediaId == media.mediaId && + relayMediaUrlBuilder != null + ? relayMediaUrlBuilder('default', activeToken, relayed.mediaId, relayed.name, dialledHost) + : media?.file != null + ? urlBuilder('default', activeToken, media, dialledHost) + : media?.url, relayUrl: relay && relayUrlBuilder != null ? relayUrlBuilder('default', activeToken, media!.jellyfin!.itemId, dialledHost) : null, @@ -437,3 +521,9 @@ void mountSyncRoute( }); }); } + +ActiveRelay _memberRelay(ServerHub hub, String token, String name) { + late final ActiveRelay r; + r = ActiveRelay.fromMember(hub.engine.mediaId, name, token, onLiveChanged: () => hub.relayLiveChanged(r)); + return r; +} diff --git a/packages/server/lib/src/sync_server.dart b/packages/server/lib/src/sync_server.dart index de29207..899a6a0 100644 --- a/packages/server/lib/src/sync_server.dart +++ b/packages/server/lib/src/sync_server.dart @@ -8,6 +8,7 @@ import 'package:sync_relay_core/core.dart'; import 'jellyfin_relay.dart'; import 'media_endpoint.dart'; +import 'relay_routes.dart'; import 'server_hub.dart'; import 'session.dart'; @@ -62,6 +63,7 @@ class SyncServer { ); }); mountSessionMediaRoute(_app, sessions); + mountRelayRoutes(_app, sessions, () => hub.relay); mountJellyfinRelayRoute( _app, sessions, () => hub.jellyfin, () => _relayClient ??= HttpClient()); // advertiseHost is only an explicit override; otherwise echo what the client dialled. @@ -72,7 +74,9 @@ class SyncServer { final name = Uri.encodeComponent(media?.name ?? 'file'); return 'http://${authority(dialledHost)}/m/$sid/$token/$name'; }, relayUrlBuilder: (sid, token, itemId, dialledHost) => - 'http://${authority(dialledHost)}/j/$sid/$token/${Uri.encodeComponent(itemId)}'); + 'http://${authority(dialledHost)}/j/$sid/$token/${Uri.encodeComponent(itemId)}', + relayMediaUrlBuilder: (sid, token, mediaId, name, dialledHost) => + 'http://${authority(dialledHost)}/r/$sid/$token/$mediaId/${Uri.encodeComponent(name)}'); _app.fallback = (final req) => Response.notFound( body: Body.fromString('not found'), ); @@ -143,6 +147,7 @@ class SyncServer { _heartbeat = null; _relayClient?.close(force: true); _relayClient = null; + hub.setRelay(null); await hub.closeAllSockets(); await _app.close(); _server = null; diff --git a/packages/server/test/relay_test.dart b/packages/server/test/relay_test.dart new file mode 100644 index 0000000..f848fee --- /dev/null +++ b/packages/server/test/relay_test.dart @@ -0,0 +1,386 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; +import 'dart:math'; +import 'dart:typed_data'; + +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; +import 'package:test/test.dart'; + +// The relay end to end inside one process: a real SyncServer, the real +// RelaySource from core, viewers as plain HTTP range requests. + +class _Member { + final WebSocket ws; + final raw = []; + late WelcomeMessage welcome; + _Member(this.ws); + + static Future<_Member> join(int port, String name, {String secret = 'pw'}) async { + final ws = await WebSocket.connect('ws://127.0.0.1:$port/sync'); + final m = _Member(ws); + final w = Completer(); + ws.listen((f) { + m.raw.add('$f'); + if ('$f'.contains('"t":"welcome"') && !w.isCompleted) w.complete('$f'); + }, onError: (_) {}); + ws.add(jsonEncode(HelloMessage( + proto: RelayConstants.protoVersion, + secret: secret, + user: HelloUser(name: name), + media: const HelloMedia(name: '', size: 0, duration: 0), + ).toJson())); + m.welcome = WelcomeMessage.fromJson( + Map.from(jsonDecode(await w.future.timeout(const Duration(seconds: 5))) as Map)); + return m; + } + + void send(Map j) => ws.add(jsonEncode(j)); + + Map? get lastState { + for (final f in raw.reversed) { + final d = jsonDecode(f) as Map; + if (d['t'] == 'state') return Map.from(d); + } + return null; + } + + Map? get media => lastState?['media'] as Map?; +} + +Future _until(bool Function() cond, String what, {int secs = 5}) async { + final deadline = DateTime.now().add(Duration(seconds: secs)); + while (!cond()) { + if (DateTime.now().isAfter(deadline)) fail('timed out waiting for $what'); + await Future.delayed(const Duration(milliseconds: 20)); + } +} + +void main() { + late Directory tmp; + late File file; + late Uint8List bytes; + late SyncServer server; + late HttpClient http; + final cleanup = Function()>[]; + + setUp(() async { + tmp = await Directory.systemTemp.createTemp('relay-test'); + final rnd = Random(7); + // 5.5 slices: exercises the short last slice. + bytes = Uint8List.fromList(List.generate(5 * RelayBuffer.slice + 500000, (_) => rnd.nextInt(256))); + file = File('${tmp.path}/ep 01.mkv')..writeAsBytesSync(bytes); + server = SyncServer(const AppConfig(port: 0, secret: 'pw', name: 't')); + await server.start(); + http = HttpClient(); + }); + + tearDown(() async { + for (final c in cleanup.reversed) { + await c(); + } + cleanup.clear(); + http.close(force: true); + await server.stop(); + await tmp.delete(recursive: true); + }); + + Future<_Member> join(String name) async { + final m = await _Member.join(server.boundPort, name); + cleanup.add(() => m.ws.close()); + return m; + } + + Future startSource(_Member m, {RelayBytes? from}) async { + final src = RelaySource( + Uri.parse('ws://127.0.0.1:${server.boundPort}/r/default/${m.welcome.token}/source'), + from ?? FileRelayBytes(file), + backoff: (_) => const Duration(milliseconds: 100)); + await src.start(); + cleanup.add(src.close); + return src; + } + + Future swapRelay(_Member m, {String name = 'ep 01.mkv', Map relay = const {}, String? url}) async { + final before = m.raw.length; + m.send({ + 't': 'swap', + 'media': {'name': name, 'size': bytes.length, 'duration': 0, if (url != null) 'url': url, 'relay': relay}, + }); + await _until(() => m.raw.skip(before).any((f) => f.contains('"relay"') && f.contains('"t":"state"')), 'relay state'); + } + + String viewerUrl(_Member v, int mediaId) => + 'http://127.0.0.1:${server.boundPort}/r/default/${v.welcome.token}/$mediaId/${Uri.encodeComponent('ep 01.mkv')}'; + + Future<(int, Map, List)> get(String url, {String? range}) async { + final req = await http.getUrl(Uri.parse(url)); + if (range != null) req.headers.set('range', range); + final res = await req.close(); + final body = [for (final c in await res.toList()) ...c]; + final h = {}; + res.headers.forEach((k, v) => h[k] = v.join(',')); + return (res.statusCode, h, body); + } + + int mediaIdOf(_Member m) => m.lastState!['mediaId'] as int; + + group('member as source', () { + test('state names the source, no URL; live once the channel is up', () async { + final alice = await join('alice'); + await swapRelay(alice); + expect(alice.media!['relay'], {'source': alice.welcome.memberId}); + expect(alice.media!.containsKey('url'), isFalse); + await startSource(alice); + await _until(() => (alice.media?['relay'] as Map?)?['live'] == true, 'live'); + }); + + test('viewers get byte-exact ranges, including the short last slice and 416', () async { + final alice = await join('alice'); + final bob = await join('bob'); + await swapRelay(alice); + await startSource(alice); + final url = viewerUrl(bob, mediaIdOf(alice)); + final n = bytes.length; + for (final (range, start, end) in [ + (null, 0, n - 1), + ('bytes=0-99', 0, 99), + ('bytes=${RelayBuffer.slice - 10}-${RelayBuffer.slice + 10}', RelayBuffer.slice - 10, RelayBuffer.slice + 10), + ('bytes=${n - 777}-', n - 777, n - 1), + ('bytes=-333', n - 333, n - 1), + ]) { + final (status, h, body) = await get(url, range: range); + expect(status, range == null ? 200 : 206, reason: '$range'); + expect(body, bytes.sublist(start, end + 1), reason: '$range'); + if (range != null) expect(h['content-range'], 'bytes $start-$end/$n'); + expect(h['content-type'], 'video/x-matroska'); + } + final (status, h, _) = await get(url, range: 'bytes=${n + 5}-'); + expect(status, 416); + expect(h['content-range'], 'bytes */$n'); + }); + + test('a late joiner gets its own /r/ URL in welcome', () async { + final alice = await join('alice'); + await swapRelay(alice); + final carol = await join('carol'); + expect(carol.welcome.mediaUrl, viewerUrl(carol, mediaIdOf(alice))); + }); + + test('nobody watching: the source uploads nothing', () async { + final alice = await join('alice'); + await swapRelay(alice); + final src = await startSource(alice); + await _until(() => server.hub.relay!.live, 'live'); + await Future.delayed(const Duration(milliseconds: 600)); + expect(src.bytesSent, 0); + }); + + test('several viewers: each byte crosses the source link about once', () async { + final alice = await join('alice'); + final viewers = [for (final n in ['b', 'c', 'd']) await join(n)]; + await swapRelay(alice); + final src = await startSource(alice); + final id = mediaIdOf(alice); + final got = await Future.wait([for (final v in viewers) get(viewerUrl(v, id))]); + for (final (status, _, body) in got) { + expect(status, 200); + expect(body.length, bytes.length); + } + expect(src.bytesSent, lessThan(bytes.length * 1.3)); + }); + + test('only the source may open the channel', () async { + final alice = await join('alice'); + final bob = await join('bob'); + await swapRelay(alice); + final impostor = await startSource(bob); + await impostor.done.timeout(const Duration(seconds: 5)); + expect(server.hub.relay!.live, isFalse); + }); + + test('a new swap drops the buffer, old URLs 404, the source stops', () async { + final alice = await join('alice'); + final bob = await join('bob'); + await swapRelay(alice); + final src = await startSource(alice); + final oldId = mediaIdOf(alice); + expect((await get(viewerUrl(bob, oldId), range: 'bytes=0-9')).$1, 206); + final buf = server.hub.relay!.buffer; + alice.send({ + 't': 'swap', + 'media': {'name': 'x', 'size': 0, 'duration': 0, 'url': 'http://example.invalid/x.mkv'}, + }); + await _until(() => mediaIdOf(alice) != oldId, 'swap'); + expect(server.hub.relay, isNull); + expect(buf.cachedBytes, 0); + await src.done.timeout(const Duration(seconds: 5)); + expect((await get(viewerUrl(bob, oldId), range: 'bytes=0-9')).$1, 404); + }); + + test('server stop drops the relay', () async { + final alice = await join('alice'); + await swapRelay(alice); + final buf = server.hub.relay!.buffer; + await server.stop(); + expect(server.hub.relay, isNull); + expect(buf.cachedBytes, 0); + server = SyncServer(const AppConfig(port: 0, secret: 'pw', name: 't')); + await server.start(); + }); + + test('source drops and comes back: buffered bytes keep serving, the rest resumes', () async { + final alice = await join('alice'); + final bob = await join('bob'); + await swapRelay(alice); + final src = await startSource(alice); + final url = viewerUrl(bob, mediaIdOf(alice)); + expect((await get(url, range: 'bytes=0-${RelayBuffer.slice * 2 - 1}')).$3, + bytes.sublist(0, RelayBuffer.slice * 2)); + await src.close(); + await src.done; + await _until(() => !server.hub.relay!.live, 'down'); + expect((await get(url, range: 'bytes=0-99')).$3, bytes.sublist(0, 100)); + final pending = get(url, range: 'bytes=${RelayBuffer.slice * 4}-'); + await Future.delayed(const Duration(milliseconds: 300)); + await startSource(alice); + final (status, _, body) = await pending.timeout(const Duration(seconds: 10)); + expect(status, 206); + expect(body, bytes.sublist(RelayBuffer.slice * 4)); + }); + + test('Jellyfin swap can carry a relay', () async { + final alice = await join('alice'); + final before = alice.raw.length; + alice.send({ + 't': 'swap', + 'media': { + 'name': 'Pilot', + 'size': 0, + 'duration': 0, + 'jellyfin': {'serverId': 'srv1', 'itemId': '0123456789abcdef0123456789abcdef', 'title': 'Pilot'}, + 'relay': {}, + }, + }); + await _until(() => alice.raw.skip(before).any((f) => f.contains('"jellyfin"')), 'state'); + expect(alice.media!['relay'], {'source': alice.welcome.memberId}); + expect(server.hub.relay!.sourceToken, alice.welcome.token); + }); + }); + + group('server fetches the link', () { + late HttpServer origin; + setUp(() async { + origin = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + origin.listen((req) async { + final r = resolveMediaRequest( + rangeHeader: req.headers.value('range'), size: bytes.length, filename: 'x.mkv'); + req.response.statusCode = r.status; + r.headers.forEach(req.response.headers.set); + if (r.status == 200 || r.status == 206) { + final end = r.end ?? bytes.length - 1; + req.response.add(bytes.sublist(r.start, end + 1)); + } + await req.response.close(); + }); + }); + tearDown(() => origin.close(force: true)); + + test('byte-exact, the link is never broadcast', () async { + final alice = await join('alice'); + final link = 'http://127.0.0.1:${origin.port}/secret/file.mkv?api_key=KEY'; + await swapRelay(alice, relay: {'fetch': true}, url: link); + expect(jsonEncode(alice.lastState), isNot(contains('KEY'))); + await _until(() => server.hub.relay!.live, 'live'); + final bob = await join('bob'); + expect(jsonEncode(bob.welcome.toJson()), isNot(contains('KEY'))); + final (status, _, body) = await get(bob.welcome.mediaUrl!, range: 'bytes=1000-'); + expect(status, 206); + expect(body, bytes.sublist(1000)); + }); + + test('an open server (no password) refuses to fetch', () async { + await server.stop(); + server = SyncServer(const AppConfig(port: 0, secret: '', name: 'open')); + await server.start(); + final m = await _Member.join(server.boundPort, 'mallory', secret: ''); + cleanup.add(() => m.ws.close()); + m.send({ + 't': 'swap', + 'media': { + 'name': 'x', 'size': 0, 'duration': 0, + 'url': 'http://127.0.0.1:${origin.port}/x.mkv', 'relay': {'fetch': true}, + }, + }); + await _until(() => m.raw.any((f) => f.contains('server fetch needs a server password')), 'err'); + expect(server.hub.relay, isNull); + }); + + test('a non-http link is refused', () async { + final alice = await join('alice'); + alice.send({ + 't': 'swap', + 'media': {'name': 'x', 'size': 0, 'duration': 0, 'url': 'file:///etc/passwd', 'relay': {'fetch': true}}, + }); + await _until(() => alice.raw.any((f) => f.contains('relay fetch needs')), 'err'); + expect(server.hub.relay, isNull); + }); + }); + + group('buffer', () { + // A feed that serves [bytes] in-process and counts what it sends. + late RelayBuffer buf; + late LocalRelayFeed feed; + late _CountingBytes counted; + setUp(() { + buf = RelayBuffer(ahead: 2 * RelayBuffer.slice, back: RelayBuffer.slice, pin: RelayBuffer.slice); + counted = _CountingBytes(FileRelayBytes(file)); + feed = LocalRelayFeed(counted, buf); + buf.attach(feed, bytes.length); + }); + tearDown(() async { + buf.dispose(); + await feed.close(); + }); + + Future> read(String v, int start, int end) async => + [for (final c in await buf.read(v, start, end).toList()) ...c]; + + test('keeps pinned head/tail plus a window around the viewer', () async { + expect(await read('a', 0, bytes.length - 1), bytes); + await Future.delayed(const Duration(milliseconds: 100)); + // ahead 2 + back 1 + 2 pinned slices: never all 6. + expect(buf.cachedBytes, lessThanOrEqualTo(5 * RelayBuffer.slice)); + }); + + test('a superseded request does not hold the feed', () async { + // Like mpv after a seek: the old request stays open, unread. + final old = buf.read('a', 0, bytes.length - 1).listen((_) {})..pause(); + final s = 3 * RelayBuffer.slice; + expect(await read('a', s, s + 999).timeout(const Duration(seconds: 5)), bytes.sublist(s, s + 1000)); + await old.cancel(); + }); + + test('dispose frees everything and ends waiting readers', () async { + await read('a', 0, 10); + buf.dispose(); + expect(buf.cachedBytes, 0); + }); + }); +} + +class _CountingBytes implements RelayBytes { + final RelayBytes inner; + int sent = 0; + _CountingBytes(this.inner); + @override + Future size() => inner.size(); + @override + Stream> read(int start, int end) => inner.read(start, end).map((c) { + sent += c.length; + return c; + }); + @override + Future close() => inner.close(); +} diff --git a/packages/server/test/session_test.dart b/packages/server/test/session_test.dart index a9737c9..462ff7c 100644 --- a/packages/server/test/session_test.dart +++ b/packages/server/test/session_test.dart @@ -80,8 +80,14 @@ void main() { }); group('core welcome + proto', () { - test('proto 7: Jellyfin refs and relay', () { - expect(RelayConstants.protoVersion, 7); + test('proto 8: relayed media', () { + expect(RelayConstants.protoVersion, 8); + const m = HelloMedia(name: 'a', size: 1, duration: 0, relay: MediaRelay(source: 'm1', live: true)); + final back = HelloMedia.fromJson(Map.from(m.toJson())); + expect(back.relay!.source, 'm1'); + expect(back.relay!.live, isTrue); + expect(back.relay!.fetch, isFalse); + expect(const MediaRelay().toJson(), isEmpty); }); test('welcome round-trips with token + member id', () { diff --git a/tool/relay_e2e.dart b/tool/relay_e2e.dart new file mode 100644 index 0000000..5a5cbf5 --- /dev/null +++ b/tool/relay_e2e.dart @@ -0,0 +1,71 @@ +// Relay harness: a real SyncServer, a member that swaps a file in as +// relayed media, and that member's RelaySource dialling the server through +// [--via] (e.g. a link shaper). Prints `URL ` for players, and +// `STATS ...` every 5 s. Usage: +// dart run tool/relay_e2e.dart --file x.mkv [--port 9100] [--via 9200] +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:args/args.dart'; +import 'package:sync_relay_core/core.dart'; +import 'package:sync_relay_server/server.dart'; + +Future main(List argv) async { + final a = (ArgParser() + ..addOption('file', mandatory: true) + ..addOption('port', defaultsTo: '9100') + ..addOption('via') + ..addOption('viewers', defaultsTo: '1')) + .parse(argv); + final file = File(a['file'] as String); + final port = int.parse(a['port'] as String); + final server = SyncServer(AppConfig(port: port, secret: '', name: 'e2e')); + await server.start(); + + Future<(WebSocket, WelcomeMessage)> join(String name) async { + final ws = await WebSocket.connect('ws://127.0.0.1:$port/sync'); + final welcome = Completer(); + ws.listen((f) { + final j = jsonDecode('$f') as Map; + if (j['t'] == 'welcome' && !welcome.isCompleted) { + welcome.complete(WelcomeMessage.fromJson(Map.from(j))); + } + }); + ws.add(jsonEncode(HelloMessage( + proto: RelayConstants.protoVersion, + secret: '', + user: HelloUser(name: name), + media: const HelloMedia(name: '', size: 0, duration: 0)) + .toJson())); + // Stay alive like a real client (silent members get dropped). + Timer.periodic(const Duration(seconds: 1), (_) => ws.add(jsonEncode({'t': 'report', 'position': 0, 'buffered': 0, 'eventId': 0}))); + return (ws, await welcome.future); + } + + final (srcWs, srcW) = await join('source'); + final name = file.uri.pathSegments.last; + srcWs.add(jsonEncode(SwapIntent( + media: HelloMedia(name: name, size: await file.length(), duration: 0, relay: const MediaRelay())) + .toJson())); + await Future.delayed(const Duration(milliseconds: 300)); + final via = a['via'] as String? ?? '$port'; + var source = RelaySource(Uri.parse('ws://127.0.0.1:$via/r/default/${srcW.token}/source'), FileRelayBytes(file)); + await source.start(); + final urls = []; + for (var i = 0; i < int.parse(a['viewers'] as String); i++) { + urls.add((await join('viewer$i')).$2.mediaUrl!); + } + stdout.writeln('URL ${urls.join(',')}'); + Timer.periodic(const Duration(seconds: 5), (_) { + final b = server.hub.relay?.buffer; + stdout.writeln('STATS sourceSent=${source.bytesSent} toViewers=${b?.bytesToViewers} ' + 'cached=${b?.cachedBytes} window=${source.window.window}'); + }); + // SIGUSR1: drop the source; SIGUSR2: bring it back (outage tests). + ProcessSignal.sigusr1.watch().listen((_) => source.close()); + ProcessSignal.sigusr2.watch().listen((_) async { + source = RelaySource(Uri.parse('ws://127.0.0.1:$via/r/default/${srcW.token}/source'), FileRelayBytes(file)); + await source.start(); + }); +}