diff --git a/README.md b/README.md index 13c9c65..c2078e0 100644 --- a/README.md +++ b/README.md @@ -166,8 +166,10 @@ right-click a Jellyfin card or episode, a video in Files, a recent link or an episode on the Schedule for **Add to queue** / **Play next**; the Jellyfin details panel has a queue button (right-click it to play next), and the card for a dropped or pasted video or link has both. A link is -checked first, like Put it up. The next item is kept ready on the app -that will put it up, so skipping to it starts at once. Everyone shares one Up next and anyone can change it; what +checked first, like Put it up. Every app keeps the next item ready, +its first seconds already loaded, so skipping to it starts at once +(not yet for a link passed on, someone else's file or one the server +fetches: those load when they go up). Everyone shares one Up next and anyone can change it; what you added stays when you leave. In the player, Up next sits in the top bar next to the chat button: the diff --git a/apps/desktop/lib/src/playback/playback_controller.dart b/apps/desktop/lib/src/playback/playback_controller.dart index 6d06c33..73205f9 100644 --- a/apps/desktop/lib/src/playback/playback_controller.dart +++ b/apps/desktop/lib/src/playback/playback_controller.dart @@ -3,7 +3,7 @@ import 'dart:ffi' show Int8; import 'dart:io'; import 'package:ffi/ffi.dart' show StringUtf8Pointer, malloc; -import 'package:flutter/foundation.dart' show listEquals; +import 'package:flutter/foundation.dart' show listEquals, mapEquals; import 'package:flutter/widgets.dart'; import 'package:media_kit/media_kit.dart'; import 'package:media_kit_video/media_kit_video.dart'; @@ -82,6 +82,7 @@ class PlaybackController { final p = _player; if (p == null) return; try { + _ytdlPath = path; final dynamic platform = p.platform; if (!identical(p, _ytdlSetOn)) { await platform?.setProperty('script-opts', 'ytdl_hook-ytdl_path=$path'); @@ -101,6 +102,7 @@ class PlaybackController { Player? _ytdlSetOn; String _ytdlRaw = ''; + String? _ytdlPath; Player get player => _player!; @@ -188,16 +190,18 @@ class PlaybackController { return AppSettings.instance.hwdec.peek(); } + SyncPlayer _newPlayer({required int bufferSize, required MPVLogLevel logLevel}) => SyncPlayer( + configuration: PlayerConfiguration( + bufferSize: bufferSize, + libass: true, + vo: Platform.environment['SYNC_RELAY_VO'], + logLevel: logLevel, + )) + ..onUserSeek = (d) => onUserSeek?.call(d); + Future init() async { if (_player != null) return; - _player = SyncPlayer( - configuration: PlayerConfiguration( - bufferSize: maxBytes, - libass: true, - vo: Platform.environment['SYNC_RELAY_VO'], - logLevel: MPVLogLevel.warn, // see [logLevel] - )) - ..onUserSeek = (d) => onUserSeek?.call(d); + _player = _newPlayer(bufferSize: maxBytes, logLevel: MPVLogLevel.warn); // see [logLevel] _shaders = ''; renderMode.value = RenderMode.unknown; shaderNotice.value = null; @@ -223,9 +227,9 @@ class PlaybackController { } } - Future _pin(String property, String value) async { + Future _pin(String property, String value, [Player? on]) async { try { - final dynamic platform = player.platform; + final dynamic platform = (on ?? player).platform; await platform?.setProperty(property, value); final got = '${await platform?.getProperty(property)}'; if (!pinMatches(value, got)) { @@ -238,17 +242,17 @@ class PlaybackController { } } - Future _pinBufferProps() async { - pinMismatches.clear(); - await _pin('demuxer-readahead-secs', '$readAheadSecs'); - await _pin('cache-secs', '$readAheadSecs'); - await _pin('demuxer-max-bytes', '$maxBytes'); - await _pin('demuxer-max-back-bytes', '$maxBackBytes'); + Future _pinBufferProps([Player? on, int ahead = maxBytes, int back = maxBackBytes]) async { + if (on == null) pinMismatches.clear(); + await _pin('demuxer-readahead-secs', '$readAheadSecs', on); + await _pin('cache-secs', '$readAheadSecs', on); + await _pin('demuxer-max-bytes', '$ahead', on); + await _pin('demuxer-max-back-bytes', '$back', on); final hwdec = _hwdecOverride; if (hwdec != null) { - await _pin('hwdec', hwdec); + await _pin('hwdec', hwdec, on); } - await _pin('stream-lavf-o', streamLavfOptions); + await _pin('stream-lavf-o', streamLavfOptions, on); } /// ffmpeg's HTTP options (mpv `stream-lavf-o`, comma form): reconnect @@ -371,10 +375,7 @@ class PlaybackController { return; } // Nothing playable opened (a track of mpv's yt-dlp EDL refused, ...). - if (!_loadErrorSent && - duration.peek() == Duration.zero && - ((log.prefix == 'cplayer' && log.level == 'fatal') || - (log.prefix == 'demux' && log.level == 'error'))) { + if (!_loadErrorSent && _loadFailed(log, duration.peek())) { _loadErrorSent = true; final http = _httpError; onLoadError?.call(http == null ? 'the video stream could not be opened' : 'the video server answered $http'); @@ -399,6 +400,15 @@ class PlaybackController { } } + /// [log] says the media could not be opened: before a [duration] is + /// known, mpv's fatal "nothing to play" or a demuxer error, or a yt-dlp + /// hook error. + static bool _loadFailed(PlayerLog log, Duration duration) { + if (log.prefix == 'ytdl_hook') return log.level == 'error' || log.level == 'fatal'; + return duration == Duration.zero && + ((log.prefix == 'cplayer' && log.level == 'fatal') || (log.prefix == 'demux' && log.level == 'error')); + } + void _setRenderMode(RenderMode mode) { renderMode.value = mode; lastRenderMode.value = mode; @@ -428,8 +438,14 @@ class PlaybackController { /// Opens [url] paused. [headers] go with every request for it (mpv /// `http-header-fields`): a Jellyfin token travels there, not in the URL. - Future openUrl(String url, {Map? headers}) async { - if (!_ready) return; + /// When [preload] holds the same [url] and [headers], that player + /// becomes this one instead (true: promoted, its buffer comes along). + Future openUrl(String url, {Map? headers}) async { + if (!_ready) return false; + final pre = _pre; + if (pre != null && pre.matches(url, headers)) { + if (await _promote(pre)) return true; + } try { _chapterJson = ''; _httpError = null; @@ -439,6 +455,7 @@ class PlaybackController { buffered.value = Duration.zero; // An empty map, not null: media_kit caches headers per URI. await player.open(Media(url, httpHeaders: headers ?? const {}), play: false); + return false; } catch (e, st) { debugPrint('open ${Uri.tryParse(url)?.replace(query: '') ?? 'media'} failed: $e\n$st'); rethrow; @@ -607,6 +624,7 @@ class PlaybackController { /// Release the player. The view drops its texture first ([alive] goes /// false, one frame passes), then mpv is torn down. Future dispose() async { + await dropPreload(); final p = _player; if (p == null) return; await _logSub?.cancel(); @@ -639,6 +657,173 @@ class PlaybackController { } catch (_) {} } + // ------------------------------------------------------------ preload + // + // A second, hidden player holding what plays next (Up next's first + // item), open paused at its start and muted, with a small cache: its + // first seconds are in before anyone asks. [openUrl] for the same URL + // promotes it to the session player (see ARCHITECTURE.md "Getting + // ready"). One at a time. + + /// Preloading on (env `SYNC_RELAY_PRELOAD=0` turns it off; tests set it). + bool preloadEnabled = Platform.environment['SYNC_RELAY_PRELOAD'] != '0'; + + /// The preload's forward cache: its first seconds and the container's + /// index, at little memory. Raised to [maxBytes] when promoted. + static const preloadBytes = 48 * 1024 * 1024; + static const _preloadBackBytes = 8 * 1024 * 1024; + + _Preload? _pre; + + /// The URL preloaded now (null: none). + String? get preloadUrl => _pre?.url; + + /// Opens [url] in the hidden preload player (replacing any other), and + /// completes once [need] seconds are buffered or the whole media is: + /// true, false when it failed (or was dropped or replaced meanwhile). + /// The same [url] and [headers] again is the same preload unless + /// [fresh]. [ytdlPath]: mpv's yt-dlp hook for web pages. + Future preload(String url, + {Map? headers, String? ytdlPath, double need = 4, bool fresh = false}) async { + if (!preloadEnabled || !_ready) return false; + final had = _pre; + if (had != null && had.matches(url, headers) && !had.failed && !fresh) return had.ready.future; + // Set before any await: a second call for the same URL finds it. + _pre = null; + if (had != null) unawaited(_disposePreload(had)); + final p = _newPlayer(bufferSize: preloadBytes, logLevel: MPVLogLevel.error); + final pre = _pre = _Preload(p, url, headers ?? const {}); + pre.logSub = p.stream.log.listen((log) { + // ffmpeg logs through one mpv per process: the first one, and once + // that is gone (a promotion disposes it) the next one created, which + // can be this preload. Its HTTP errors still name the session's. + if (log.prefix == 'ffmpeg') return _onLog(log); + if (_loadFailed(log, p.state.duration)) pre.fail(); + }); + try { + await _pinBufferProps(p, preloadBytes, _preloadBackBytes); + await _pin('volume-max', '$maxVolume', p); + await _pin('mute', 'yes', p); + final path = ytdlPath ?? _ytdlPath; + if (path != null) { + final dynamic platform = p.platform; + await platform?.setProperty('script-opts', 'ytdl_hook-ytdl_path=$path'); + await platform?.setProperty('ytdl-format', ytdlFormat); + pre.ytdlRaw = await AppYtDlp.instance.mpvRawOptions(); + await platform?.setProperty('ytdl-raw-options', pre.ytdlRaw); + pre.ytdl = true; + } + // With a view up, its own video output: mpv decodes (media_kit + // starts players with vid=no) and the first frame is there. + if (_video != null) { + pre.video = VideoController(p, configuration: VideoControllerConfiguration(hwdec: _hwdecOverride)); + await pre.video!.platform.future; + } + if (!identical(_pre, pre)) { + pre.fail(); + return false; + } + await p.open(Media(url, httpHeaders: headers ?? const {}), play: false); + if (!pre.opened.isCompleted) pre.opened.complete(); + } catch (e) { + debugPrint('preload failed: $e'); + pre.fail(); + return false; + } + unawaited(_watchPreload(pre, need)); + return pre.ready.future; + } + + Future _watchPreload(_Preload pre, double need) async { + final end = DateTime.now().add(const Duration(seconds: 60)); + while (identical(_pre, pre) && !pre.failed && DateTime.now().isBefore(end)) { + final st = pre.player.state; + final secs = st.duration.inMilliseconds / 1000; + final buffered = (st.buffer - st.position).inMilliseconds / 1000; + if (secs > 0 && (buffered >= need || st.buffer.inMilliseconds / 1000 >= secs - 0.5)) { + if (!pre.ready.isCompleted) pre.ready.complete(true); + return; + } + await Future.delayed(const Duration(milliseconds: 200)); + } + if (!pre.ready.isCompleted) pre.ready.complete(false); + } + + /// Drops the preload (nothing to get ready, or something else is). + Future dropPreload() async { + final pre = _pre; + if (pre == null) return; + _pre = null; + await _disposePreload(pre); + } + + Future _disposePreload(_Preload pre) async { + pre.fail(); + await pre.logSub?.cancel(); + try { + await pre.player.dispose(); + } catch (_) {} + } + + /// [pre] becomes the session player: the view swaps to its video + /// output ([generation]), volume, shaders, log level and the full cache + /// carry over, and the old player goes once the view let go of it. + Future _promote(_Preload pre) async { + await pre.opened.future; + final old = _player; + if (!identical(_pre, pre) || old == null) return false; + _pre = null; + if (pre.failed) { + unawaited(_disposePreload(pre)); + return false; + } + await pre.logSub?.cancel(); + if (!pre.ready.isCompleted) pre.ready.complete(true); + final p = pre.player; + final volume = old.state.volume; + await _logSub?.cancel(); + for (final sub in _timeSubs) { + await sub.cancel(); + } + _timeSubs.clear(); + _player = p; + _video = pre.video; + _ytdlSetOn = pre.ytdl ? p : null; + _ytdlRaw = pre.ytdlRaw; + _chapterJson = ''; + _httpError = null; + _loadErrorSent = false; + _shaders = ''; + final at = p.state.position.inMilliseconds; + position.value = Duration(milliseconds: at - at % 250); + duration.value = p.state.duration; + final ms = p.state.buffer.inMilliseconds; + buffered.value = Duration(milliseconds: ms - ms % 250); + chapters.value = const []; + tracks.value = const []; + _logSub = p.stream.log.listen(_onLog); + _requestLog(p, logLevel); + _watchMarks(p); + generation.value++; + try { + await p.setVolume(volume); + await _pin('mute', 'no', p); + await _pinBufferProps(p); + } catch (e) { + debugPrint('promoted player settings: $e'); + } + unawaited(applyShaders()); + unawaited(_refreshTracks()); + // The view lets go of the old texture first (as in [dispose]). + unawaited(() async { + await _settleUi(); + try { + await old.dispose(); + } catch (_) {} + }()); + return true; + } + Future stop() async { if (!_ready) return; try { @@ -671,3 +856,27 @@ bool pinMatches(String wanted, String got) { final a = double.tryParse(wanted), b = double.tryParse(got); return a != null && b != null && a == b; } + +class _Preload { + final SyncPlayer player; + final String url; + final Map headers; + VideoController? video; + StreamSubscription? logSub; + /// Completes once the media is handed to mpv (or the preload failed). + final opened = Completer(); + bool ytdl = false; + String ytdlRaw = ''; + bool failed = false; + final ready = Completer(); + _Preload(this.player, this.url, this.headers); + + bool matches(String url, Map? headers) => + url == this.url && mapEquals(headers ?? const {}, this.headers); + + void fail() { + failed = true; + if (!ready.isCompleted) ready.complete(false); + if (!opened.isCompleted) opened.complete(); + } +} diff --git a/apps/desktop/lib/src/queue/queue_controller.dart b/apps/desktop/lib/src/queue/queue_controller.dart index 3b84421..8aab2f3 100644 --- a/apps/desktop/lib/src/queue/queue_controller.dart +++ b/apps/desktop/lib/src/queue/queue_controller.dart @@ -34,7 +34,7 @@ class UpNextCountdown { const UpNextCountdown({required this.mediaId, required this.item, this.secondsLeft}); } -/// The first item's readiness, on the app that puts it up. +/// The first item's readiness on this app. enum WarmState { idle, warming, ready, failed } class QueueController { @@ -51,8 +51,8 @@ class QueueController { final countdown = signal(null); - /// The first item's id and how ready it is here (only on the app that - /// puts it up; null elsewhere). + /// The first item's id and how ready it is here (null: nothing to get + /// ready on this app). final warm = signal<({String id, WarmState state})?>(null); /// The natural continuation of what plays: the next episodes of the @@ -88,9 +88,12 @@ class QueueController { final plan = coord.jellyfinPlan.value; unawaited(_loadContinuation(next, plan)); }); - // The continuation and the state decide what to get ready. + // The continuation and the state decide what to get ready; the media + // playing having started (or sitting out) lets the preload go. effect(() { shownContinuation.value; + coord.settledMediaId.value; + coord.selfOut.value; final r = coord.serverState.value; untracked(() => _warm(r)); }); @@ -326,6 +329,7 @@ class QueueController { // --------------------------------------------------------- per state void _onState(StateBroadcast r) { + _lastState = r; _prune(r); final queue = r.queue; final play = r.queuePlay; @@ -490,58 +494,98 @@ class QueueController { String? _warmUrl; Timer? _warmTimer; int _warmGen = 0; + // What the preload player should hold once the media playing has + // started (null: nothing); [_preloadFresh]: open it again (a link's + // refresh). + ({String url, Map? headers})? _preloadWant; + bool _preloadFresh = false; + String _preloadKey = ''; + bool _preloadBusy = false; + bool _preloadOk = true; + int _preloadGen = 0; + StateBroadcast? _lastState; - /// What this app gets ready: Up next's first item when it is the one - /// that puts it up; with Up next empty, the first continuation episode - /// (only here when this app plays the series with its own account, and - /// its ▶ puts it up from here). Null: nothing. - ({String id, String? url, JellyfinRef? ref})? _warmTarget(StateBroadcast? r) { + bool _ownAccount(String serverId) => + coord.jellyfin.account(serverId) != null && coord.jellyfin.statusOf(serverId) != JellyfinAccountStatus.expired; + + /// What this app gets ready: Up next's first item, else (Up next empty) + /// the first continuation episode (only here when this app plays the + /// series with its own account). Every app does, each its own copy (see + /// [_WarmTarget]); while sitting out only the probe of a link this app + /// puts up (its swap reads it). Null: nothing. + _WarmTarget? _warmTarget(StateBroadcast? r) { + final t = _warmTargetHere(r); + if (t == null || !coord.selfOut.peek()) return t; + return t.probe == null ? null : _WarmTarget(t.id, probe: t.probe); + } + + _WarmTarget? _warmTargetHere(StateBroadcast? r) { if (r == null) return null; final head = r.queue.firstOrNull; if (head != null) { - if (queuePerformer(head, r.members).id != _me) return null; + final performer = queuePerformer(head, r.members).id == _me; final local = _local[head.key]?.pick; - return switch (head.how) { - QueueHow.direct => (id: head.id, url: head.media.url, ref: null), - QueueHow.through => (id: head.id, url: local is LinkMedia ? local.url : null, ref: null), - QueueHow.jellyfin => (id: head.id, url: null, ref: head.media.jellyfin), - QueueHow.file || QueueHow.server => (id: head.id, url: null, ref: null), - }; + switch (head.how) { + case QueueHow.jellyfin: + final ref = head.media.jellyfin; + if (ref == null) return null; + if (_ownAccount(ref.serverId)) return _WarmTarget(head.id, ref: ref); + // No account here: the server's login passes it on (`/j`). + final relay = r.relayNext ? coord.jellyfinRelayUrl(ref) : null; + return relay == null ? null : _WarmTarget(head.id, open: relay); + case QueueHow.direct: + final url = head.media.url; + if (url == null) return null; + return _WarmTarget(head.id, probe: performer ? url : null, open: url); + case QueueHow.through: + // Only its adder has the link; the others' relay starts with the swap. + if (!performer) return null; + final url = local is LinkMedia ? local.url : null; + return _WarmTarget(head.id, probe: url, open: url); + case QueueHow.file: + if (!performer) return null; + // The host's own file plays from its `/m` URL once it is up. + return _WarmTarget(head.id, open: local is FileMedia && coord.hostFile == null ? local.path : null); + case QueueHow.server: + return performer ? _WarmTarget(head.id) : null; + } } final next = shownContinuation.peek().firstOrNull; if (next == null) return null; - return (id: 'next:${next.item.id}', url: null, ref: mediaForPick(next.pick).jellyfin); + return _WarmTarget('next:${next.item.id}', ref: mediaForPick(next.pick).jellyfin); } - /// Keeps the one item this app puts up next ready, so a skip or the end - /// of what plays starts it at once: a link probed (`AppYtDlp.probe`, - /// kept while it stays first), a Jellyfin item resolved with this app's - /// account ([SessionCoordinator.warmJellyfin]). Done as soon as it is - /// first here, again before that runs out, dropped when it is no longer - /// first. A file or a link the server fetches has nothing to get ready. + /// Keeps the one item that plays next ready on this app, so a skip or + /// the end of what plays starts it at once: a link probed when this app + /// puts it up (`AppYtDlp.probe`, kept while it stays first), a Jellyfin + /// item resolved with this app's account + /// ([SessionCoordinator.warmJellyfin]), and then opened in the hidden + /// preload player ([SessionCoordinator.preloadMedia]) once the media + /// playing has started. Done as soon as it is first, again before that + /// runs out, dropped when it is no longer first. void _warm(StateBroadcast? r) { final t = _warmTarget(r); - final ref = t?.ref; - final key = t == null ? '' : '${t.id}|${t.url}|${ref?.serverId}/${ref?.itemId}/${ref?.mediaSourceId}'; - if (key == _warmKey) return; - _dropWarm(r); - _warmKey = key; - if (t == null) { - if (warm.peek() != null) warm.value = null; - return; + final key = t?.key ?? ''; + if (key != _warmKey) { + _dropWarm(r); + _warmKey = key; + if (t == null) { + if (warm.peek() != null) warm.value = null; + } else if (t.empty) { + warm.value = (id: t.id, state: WarmState.ready); + } else { + warm.value = (id: t.id, state: WarmState.warming); + _resolveWarm(t, fresh: false); + } } - if (t.url == null && ref == null) { - warm.value = (id: t.id, state: WarmState.ready); - return; - } - warm.value = (id: t.id, state: WarmState.warming); - _resolveWarm(t, fresh: false); + _syncPreload(r); } void _dropWarm(StateBroadcast? r) { _warmTimer?.cancel(); _warmTimer = null; _warmGen++; + _preloadWant = null; final url = _warmUrl; _warmUrl = null; if (url != null) coord.ytDlp.release(url); @@ -550,33 +594,89 @@ class QueueController { if (w != null && !sameJellyfinItem(w.ref, refOfStateMedia(r?.media))) coord.warmJellyfin = null; } - void _resolveWarm(({String id, String? url, JellyfinRef? ref}) t, {required bool fresh}) { + void _resolveWarm(_WarmTarget t, {required bool fresh}) { final gen = ++_warmGen; unawaited(() async { - var ok = false; + var ok = true; + ({String url, Map? headers})? open; try { - final url = t.url, ref = t.ref; - if (url != null) { - _warmUrl = url; - final p = await coord.ytDlp.probe(url, keep: _warmKeep, fresh: fresh).timeout(const Duration(seconds: 60)); + final probe = t.probe, ref = t.ref, url = t.open; + if (probe != null) { + _warmUrl = probe; + final p = await coord.ytDlp.probe(probe, keep: _warmKeep, fresh: fresh).timeout(const Duration(seconds: 60)); ok = p.problem == null; - } else if (ref != null) { + } + if (ref != null) { final plan = await resolveJellyfinRef(coord.jellyfin, ref, relay: false, relayUrl: null); if (gen != _warmGen) return; + ok = plan is OwnPlay; if (plan is OwnPlay) { coord.warmJellyfin = (ref: ref, plan: plan, until: DateTime.now().add(_warmKeep)); - ok = true; + open = (url: plan.url, headers: plan.headers); } + } else if (url != null && ok) { + open = (url: url, headers: null); } } catch (e) { + ok = false; debugPrint('queue: getting ${t.id} ready: $e'); } if (gen != _warmGen) return; - warm.value = (id: t.id, state: ok ? WarmState.ready : WarmState.failed); + _preloadWant = ok ? open : null; + // A refresh opens a link again (a page's stream URLs expire); the + // same Jellyfin stream stays as it is. + _preloadFresh = fresh && ok && t.ref == null && open != null; + final waits = ok && open != null && coord.playback.preloadEnabled; + warm.value = (id: t.id, state: !ok ? WarmState.failed : (waits ? WarmState.warming : WarmState.ready)); + _syncPreload(coord.serverState.peek()); _warmTimer = Timer(ok ? _warmRefresh : _warmRetry, () => _resolveWarm(t, fresh: true)); }()); } + /// Puts [_preloadWant] in the preload player, or drops what it holds; + /// only once the media playing has started (its own open goes first, + /// and a preload a swap is about to take is never replaced before), at + /// once while sitting out (nothing preloads then). + void _syncPreload(StateBroadcast? r) { + // The newest state seen here: the coordinator's lags behind while the + // same `state` is still being handed to it. + final seen = _lastState; + if (r == null || (seen != null && seen.mediaId > r.mediaId)) r = seen; + if (r == null) return; + final out = coord.selfOut.peek(); + if (!out && r.media != null && coord.settledMediaId.peek() != r.mediaId) return; + final want = out ? null : _preloadWant; + final playback = coord.playback; + if (want == null) { + _preloadKey = ''; + _preloadGen++; + if (playback.preloadUrl != null) unawaited(playback.dropPreload()); + return; + } + if (!playback.preloadEnabled) return; + final key = '${want.url}|${want.headers}'; + final fresh = _preloadFresh; + // Asked for already: in flight, there, or failed (tried again with the + // next refresh). One that went away (Reload video) is asked again. + if (!fresh && key == _preloadKey && (_preloadBusy || !_preloadOk || playback.preloadUrl == want.url)) return; + _preloadFresh = false; + _preloadKey = key; + _preloadBusy = true; + final gen = ++_preloadGen; + final id = warm.peek()?.id; + unawaited(coord.preloadMedia(want.url, headers: want.headers, fresh: fresh).then((ok) { + if (gen != _preloadGen) return; + _preloadBusy = false; + _preloadOk = ok; + // A preload that fails leaves the usual open: nothing to say. + if (!ok) debugPrint('queue: preload of $id failed'); + final w = warm.peek(); + if (w != null && w.id == id && w.state == WarmState.warming) { + warm.value = (id: w.id, state: WarmState.ready); + } + })); + } + // -------------------------------------------------------- continuation /// The next few episodes after the one playing (own account only). @@ -610,3 +710,18 @@ class QueueController { } } } + +/// What [QueueController] gets ready for one item on this app: [probe] a +/// link (the app that puts it up), resolve [ref] with this app's own +/// account, and preload [open] (or the resolved stream). None of it: +/// nothing to do here, and it is ready ([empty]). +class _WarmTarget { + final String id; + final String? probe; + final JellyfinRef? ref; + final String? open; + const _WarmTarget(this.id, {this.probe, this.ref, this.open}); + + bool get empty => probe == null && ref == null && open == null; + String get key => '$id|$probe|$open|${ref?.serverId}/${ref?.itemId}/${ref?.mediaSourceId}'; +} diff --git a/apps/desktop/lib/src/session/session_coordinator.dart b/apps/desktop/lib/src/session/session_coordinator.dart index e0f64da..51db682 100644 --- a/apps/desktop/lib/src/session/session_coordinator.dart +++ b/apps/desktop/lib/src/session/session_coordinator.dart @@ -31,6 +31,7 @@ import 'idle.dart'; import 'media_art.dart'; import 'now_playing.dart'; import 'now_playing_info.dart'; +import 'switch_trace.dart'; /// Thrown when a connect is superseded by cancel/leave before it finished. class ConnectCancelled implements Exception { @@ -180,6 +181,8 @@ class SessionCoordinator { /// The current media's poster and banner by `state.media.art` hash. late final ArtCache art = ArtCache(sync); final PlaybackController playback = PlaybackController(); + /// Timings of each media change here (see [SwitchTrace]). + final SwitchTrace switchTrace = SwitchTrace(); final StateDiff diff = StateDiff(); final health = @@ -367,6 +370,7 @@ class SessionCoordinator { playback.onYtdlError = _setPlayerError; playback.onLoadError = _setPlayerError; playback.onUserSeek = (d) => _userSeek(d.inMicroseconds / 1e6); + _bufferSub ??= playback.buffered.subscribe((_) => _onBuffered()); await playback.init(); _playerReady = true; } catch (e, st) { @@ -418,6 +422,7 @@ class SessionCoordinator { } bool _reloadingVideo = false; + void Function()? _bufferSub; /// "Reload video" (button, Ctrl+R, status menu): fresh /// player + video output, current media reopened. While it runs the @@ -460,6 +465,7 @@ class SessionCoordinator { diff.reset(); _mediaId = -1; _playerMediaId = -1; + settledMediaId.value = -1; _openUrl = null; playerError.value = ''; _memberErrors = const {}; @@ -553,40 +559,85 @@ class SessionCoordinator { _openHeaders = headers; diff.reset(); _opening = true; + final mediaId = _mediaId; + var ok = false; try { await _prepareYtDlp(url); - await playback.openUrl(url, headers: headers); + switchTrace.mark(mediaId, SwitchTrace.open); + final promoted = await playback.openUrl(url, headers: headers); + switchTrace.mark(mediaId, promoted ? SwitchTrace.preloaded : SwitchTrace.opened); health.value = const HealthStatus(HealthSeverity.ok, 'connected'); - return true; + ok = true; } catch (e) { debugPrint('open media failed: $e'); _setPlayerError('$e'); - return false; } finally { _opening = false; } + // Tell the server at once (a preloaded media already has its buffer). + final r = serverState.peek(); + if (ok && r != null && r.mediaId == mediaId && url == _openUrl && sync.token.isNotEmpty && _playerReady) { + _sendReport(r); + } + return ok; + } + + // The media whose buffer reached the gate's need was reported at once. + int _bufferReportedFor = -1; + + /// While held, the report that the buffer reached the gate's need goes + /// out as it happens, not on the next tick. + void _onBuffered() { + final r = serverState.peek(); + if (r == null || r.transport != 'held' || _bufferReportedFor == r.mediaId) return; + if (_lastOut || _opening || _openUrl == null || _playerMediaId != r.mediaId || !_playerReady) return; + if (sync.token.isEmpty || playback.bufferedSecsAhead < r.bufferNeed) return; + _bufferReportedFor = r.mediaId; + _sendReport(r); } /// A web link (not this server's): mpv's yt-dlp hook gets the app's /// managed yt-dlp, downloaded first if needed; a media file name does /// not wait for that. Future _prepareYtDlp(String url) async { + final path = await _ytDlpPathFor(url); + if (path != null) await playback.useYtDlp(path); + } + + /// The yt-dlp mpv's hook needs for [url] (null: none, not a web link). + Future _ytDlpPathFor(String url) async { final uri = Uri.tryParse(url); - if (uri == null || !(uri.isScheme('http') || uri.isScheme('https'))) return; - if (uri.host == _wsHost && uri.port == _wsPort) return; + if (uri == null || !(uri.isScheme('http') || uri.isScheme('https'))) return null; + if (uri.host == _wsHost && uri.port == _wsPort) return null; try { final y = await ytDlp.ytDlp; final ready = y.ensure(); - final path = hasMediaExtension(uri) + return hasMediaExtension(uri) ? y.path : await ready.timeout(const Duration(seconds: 90), onTimeout: () => null); - if (path != null) await playback.useYtDlp(path); } catch (e) { debugPrint('yt-dlp not ready: $e'); + return null; } } + /// The media of the latest `state` is open here (or has nothing to + /// open, could not, or waits for the server): its mediaId. Up next + /// preloads what comes next only then, so the two never compete, and a + /// preload a swap is about to take is never replaced first. + final settledMediaId = signal(-1); + + /// Opens [url] in the hidden preload player ([PlaybackController.preload]); + /// [openUrl] for the same URL and headers then starts from it. True + /// once its first seconds are in. + Future preloadMedia(String url, {Map? headers, bool fresh = false}) async { + if (!_playerReady) return false; + final path = await _ytDlpPathFor(url); + if (!_playerReady) return false; + return playback.preload(url, headers: headers, ytdlPath: path, fresh: fresh); + } + bool _relayWaitShown = false; /// The server-fetched media this viewer opens once it is live. @@ -680,6 +731,7 @@ class SessionCoordinator { void _onState(StateBroadcast r) { _lastStateAt = DateTime.now().toUtc(); serverState.value = r; + if (r.transport == 'playing') switchTrace.mark(r.mediaId, SwitchTrace.playing); _toastMemberErrors(r); _warnSlowSource(r); _checkPlayerError(r.mediaId); @@ -717,6 +769,7 @@ class SessionCoordinator { _lastEventId = r.eventId; if (r.mediaId != _mediaId) { _mediaId = r.mediaId; + switchTrace.start(r.mediaId); final ref = refOfStateMedia(r.media); // Back after a dropped connection on the same item played with our // own account: nothing to resolve or reopen. @@ -779,6 +832,7 @@ class SessionCoordinator { } _seekGate.reset(); _clearPendingSeek(); + if (_mediaId == r.mediaId) settledMediaId.value = r.mediaId; } else if (_awaitLiveFor == r.mediaId && (stateMediaRelay(r)?.live ?? false)) { _awaitLiveFor = -1; await _openMedia( @@ -998,6 +1052,31 @@ class SessionCoordinator { _ticker = Timer.periodic(const Duration(seconds: 1), (_) => _tick()); } + /// The 1 s `report` (also sent at once when a new media is open, so the + /// server's gate does not wait for the next tick). + void _sendReport(StateBroadcast r) { + // Duration only once the open media is the server's current one. + // Sitting out: still a report (it keeps this member fresh), empty. + final known = !_lastOut && !_opening && _openUrl != null && _mediaId == r.mediaId; + // The buffer counts only while the player holds the server's current + // media: before a swap's media is open it is the previous one's (or + // none), and the start must wait for this member. + final holds = !_lastOut && !_opening && _openUrl != null && _playerMediaId == r.mediaId; + _checkPlayerError(r.mediaId); + final error = playerError.peek(); + final buffered = holds ? playback.bufferedSecsAhead : 0.0; + if (holds && buffered >= r.bufferNeed) switchTrace.mark(r.mediaId, SwitchTrace.buffered); + sync.sendReport( + position: _lastOut ? 0 : playback.positionSecs, + buffered: buffered, + eventId: r.eventId, + duration: known ? playback.durationSecs : 0, + mediaId: known ? r.mediaId : 0, + jf: _jfAccessFor(r), + error: known && _playerErrorFor == r.mediaId ? error : '', + ); + } + Future _tick() async { final now = DateTime.now().toUtc(); toasts.value = @@ -1005,24 +1084,7 @@ class SessionCoordinator { final r = serverState.peek(); if (sync.token.isNotEmpty) sync.sendPing(); if (sync.token.isNotEmpty && r != null && _playerReady) { - // Duration only once the open media is the server's current one. - // Sitting out: still a report (it keeps this member fresh), empty. - final known = !_lastOut && !_opening && _openUrl != null && _mediaId == r.mediaId; - // The buffer counts only while the player holds the server's current - // media: before a swap's media is open it is the previous one's (or - // none), and the start must wait for this member. - final holds = !_lastOut && !_opening && _openUrl != null && _playerMediaId == r.mediaId; - _checkPlayerError(r.mediaId); - final error = playerError.peek(); - sync.sendReport( - position: _lastOut ? 0 : playback.positionSecs, - buffered: holds ? playback.bufferedSecsAhead : 0, - eventId: r.eventId, - duration: known ? playback.durationSecs : 0, - mediaId: known ? r.mediaId : 0, - jf: _jfAccessFor(r), - error: known && _playerErrorFor == r.mediaId ? error : '', - ); + _sendReport(r); if (!_lastOut) { _checkSeekLanded(r); unawaited(_tickJellyfin(r)); @@ -1580,6 +1642,9 @@ class SessionCoordinator { ); } + /// This member's `/j/` URL for [ref] (the server's Jellyfin login). + String? jellyfinRelayUrl(JellyfinRef ref) => _relayUrlFor(ref); + String? _relayUrlFor(JellyfinRef ref) { if (sync.token.isEmpty) return null; final welcome = sync.relayUrl; @@ -1632,6 +1697,7 @@ class SessionCoordinator { if (gen == _jfGen) _opening = false; } if (gen != _jfGen) return; + switchTrace.mark(r.mediaId, SwitchTrace.resolved); // 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) { @@ -2046,6 +2112,7 @@ class SessionCoordinator { await _nudgeSub?.cancel(); await _errSub?.cancel(); await _statusSub?.cancel(); + _bufferSub?.call(); _aniList?.close(); await sync.dispose(); await playback.dispose(); diff --git a/apps/desktop/lib/src/session/switch_trace.dart b/apps/desktop/lib/src/session/switch_trace.dart new file mode 100644 index 0000000..9f03cb2 --- /dev/null +++ b/apps/desktop/lib/src/session/switch_trace.dart @@ -0,0 +1,53 @@ +import 'dart:io'; + +import 'package:flutter/foundation.dart'; + +/// Timings of one media change on this client, from the `state` naming +/// the new media to the server playing it (see ARCHITECTURE.md "Getting +/// ready"). Printed when [enabled] (env `SYNC_RELAY_TRACE_SWITCH=1`); +/// marks are always kept for the last few media, tests read them. +class SwitchTrace { + /// Phases, in the order they usually happen. + static const state = 'state'; + static const resolved = 'resolved'; + static const open = 'open'; + static const opened = 'opened'; + static const preloaded = 'preloaded'; + static const buffered = 'buffered'; + static const playing = 'playing'; + + static bool enabled = (Platform.environment['SYNC_RELAY_TRACE_SWITCH'] ?? '').isNotEmpty; + + final String who; + SwitchTrace([this.who = '']); + + final Map> _marks = {}; + final Map _clocks = {}; + + /// Milliseconds from [state] to each phase of media [mediaId]. + Map of(int mediaId) => Map.unmodifiable(_marks[mediaId] ?? const {}); + + /// Starts the clock for [mediaId] (its first `state`). + void start(int mediaId) { + if (_clocks.containsKey(mediaId)) return; + _clocks[mediaId] = Stopwatch()..start(); + _marks[mediaId] = {state: 0}; + while (_clocks.length > 4) { + final oldest = _clocks.keys.first; + _clocks.remove(oldest); + _marks.remove(oldest); + } + } + + /// Records [phase] once for [mediaId]; [playing] ends the trace. + void mark(int mediaId, String phase) { + final clock = _clocks[mediaId]; + final marks = _marks[mediaId]; + if (clock == null || marks == null || marks.containsKey(phase)) return; + marks[phase] = clock.elapsedMilliseconds; + if (phase == playing && enabled) { + debugPrint('switch${who.isEmpty ? '' : ' $who'} m$mediaId: ' + '${[for (final e in marks.entries) '${e.key} +${e.value} ms'].join(', ')}'); + } + } +} diff --git a/apps/desktop/test/queue_switch_e2e_test.dart b/apps/desktop/test/queue_switch_e2e_test.dart new file mode 100644 index 0000000..eb49517 --- /dev/null +++ b/apps/desktop/test/queue_switch_e2e_test.dart @@ -0,0 +1,327 @@ +import 'dart:async'; +import 'dart:convert'; +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/session_media.dart'; +import 'package:watch_together/src/jellyfin/track_prefs.dart'; +import 'package:watch_together/src/playback/yt_dlp_app.dart'; +import 'package:watch_together/src/queue/queue_controller.dart' show WarmState; +import 'package:watch_together/src/screens/home/picked_media.dart'; +import 'package:watch_together/src/session/session_coordinator.dart'; +import 'package:watch_together/src/session/switch_trace.dart'; + +// How long an Up next switch takes, end to end, over real mpv and a real +// sync server: two Jellyfin account holders and a guest (through the +// server's /j login) watch an episode, then ⏭ puts up the next episode, +// then a direct link. The fake Jellyfin and the link's host answer with a +// delay and a capped rate, like a server elsewhere. Prints each client's +// phases (SwitchTrace) and the wall time from ⏭ to the server playing. +// +// WT_SWITCH_PRELOAD=0 turns the preload off (the cold path, for comparison); +// WT_SWITCH_DELAY_MS and WT_SWITCH_RATE_KBS set the remote's delay and rate. + +final _tmp = Platform.environment['WT_TEST_TMP'] ?? Directory.systemTemp.path; +const _sid = 'ab12ab12ab12ab12ab12ab12ab12ab12'; +const _series = '5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e'; +const _ep1 = 'e1e1e1e1e1e1e1e1e1e1e1e1e1e1e1e1'; +const _ep2 = 'e2e2e2e2e2e2e2e2e2e2e2e2e2e2e2e2'; + +/// Per API request, and before a stream's first byte (WT_SWITCH_DELAY_MS). +final _delay = Duration(milliseconds: int.tryParse(Platform.environment['WT_SWITCH_DELAY_MS'] ?? '') ?? 120); +/// Per stream connection, default 24 Mbit/s (WT_SWITCH_RATE_KBS). +final _bytesPerSec = (int.tryParse(Platform.environment['WT_SWITCH_RATE_KBS'] ?? '') ?? 3072) * 1024; + +/// 60 s of 720p at about 4 Mbit/s with sound (headless mpv needs audio). +Future _media() async { + final path = '$_tmp/queue_switch_60s.mkv'; + if (await File(path).exists()) return path; + final r = await Process.run('ffmpeg', [ + '-y', '-v', 'error', + '-f', 'lavfi', '-i', 'testsrc2=size=1280x720:rate=24', + '-f', 'lavfi', '-i', 'sine=frequency=440:sample_rate=48000', + '-t', '60', '-c:v', 'libx264', '-preset', 'ultrafast', '-b:v', '4M', '-g', '48', '-pix_fmt', 'yuv420p', + '-c:a', 'aac', path, + ]); + if (r.exitCode != 0) fail('ffmpeg failed: ${r.stderr}'); + return path; +} + +/// Serves [file] with ranges, after [_delay], at most [_bytesPerSec]. +Future _serveFile(HttpRequest req, String file) async { + final f = File(file); + final size = await f.length(); + final res = req.response; + res.headers.set('Accept-Ranges', 'bytes'); + res.headers.contentType = ContentType('video', 'x-matroska'); + var start = 0, end = size - 1; + final range = RegExp(r'bytes=(\d*)-(\d*)').firstMatch(req.headers.value('range') ?? ''); + if (range != null) { + start = int.tryParse(range.group(1)!) ?? 0; + end = int.tryParse(range.group(2)!) ?? size - 1; + if (end >= size) end = size - 1; + res.statusCode = 206; + res.headers.set('Content-Range', 'bytes $start-$end/$size'); + } + res.contentLength = end - start + 1; + await Future.delayed(_delay); + try { + final raf = await f.open(); + try { + await raf.setPosition(start); + var at = start; + const chunk = 64 * 1024; + final clock = Stopwatch()..start(); + var sent = 0; + while (at <= end) { + final n = (end - at + 1).clamp(0, chunk); + res.add(await raf.read(n)); + await res.flush(); + at += n; + sent += n; + final due = Duration(microseconds: sent * 1000000 ~/ _bytesPerSec); + if (due > clock.elapsed) await Future.delayed(due - clock.elapsed); + } + } finally { + await raf.close(); + } + await res.close(); + } catch (_) {} +} + +class _FakeJellyfin { + final String file; + final tokens = {'ATOKEN': 'ua', 'CTOKEN': 'uc', 'HOSTTOKEN': 'uh'}; + late HttpServer _server; + _FakeJellyfin(this.file); + + String get base => 'http://127.0.0.1:${_server.port}'; + + Future start() async { + _server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + _server.listen(_handle); + } + + Future stop() => _server.close(force: true); + + Map _item(String id) => { + 'Id': id, + 'Name': id == _ep1 ? 'Pilot' : 'Second', + 'Type': 'Episode', + 'ServerId': _sid, + 'SeriesId': _series, + 'SeriesName': 'Fake Show', + 'ParentIndexNumber': 1, + 'IndexNumber': id == _ep1 ? 1 : 2, + 'RunTimeTicks': 60 * jellyfinTicksPerSecond, + }; + + Future _handle(HttpRequest req) async { + final auth = req.headers.value('authorization') ?? ''; + await req.drain(); + final res = req.response; + final path = req.uri.path; + if (RegExp(r'^/Videos/\w+/stream$').hasMatch(path)) return _serveFile(req, file); + await Future.delayed(_delay); + Future json(Object v) async { + res.headers.contentType = ContentType.json; + res.write(jsonEncode(v)); + await res.close(); + } + + if (path == '/System/Info/Public') return json({'Id': _sid, 'ServerName': 'Fake', 'Version': '10.11.11'}); + final token = RegExp(r'Token="([^"]*)"').firstMatch(auth)?.group(1); + final user = tokens[token]; + if (user == null) { + res.statusCode = 401; + return res.close(); + } + final m = RegExp(r'^/Items/(\w+)(/PlaybackInfo)?$').firstMatch(path); + if (m != null && m.group(2) != null) { + return json({ + 'PlaySessionId': 'ps-$user-${m.group(1)}', + 'MediaSources': [ + { + 'Id': m.group(1), + 'SupportsDirectPlay': true, + 'MediaStreams': [ + {'Index': 0, 'Type': 'Video', 'Codec': 'h264'}, + {'Index': 1, 'Type': 'Audio', 'Language': 'jpn', 'Codec': 'aac'}, + ], + } + ], + }); + } + if (m != null) return json(_item(m.group(1)!)); + if (path == '/Shows/$_series/Episodes') { + return json({'Items': [_item(_ep1), _item(_ep2)], 'TotalRecordCount': 2}); + } + if (path.startsWith('/Sessions/Playing')) { + res.statusCode = 204; + return res.close(); + } + if (path.startsWith('/UserPlayedItems/') || path.startsWith('/UserItems/')) { + return json({'Played': false, 'PlaybackPositionTicks': 0}); + } + res.statusCode = 404; + await res.close(); + } +} + +Future _poll(FutureOr Function() cond, String what, + {Duration timeout = const Duration(seconds: 40), Duration every = const Duration(milliseconds: 200)}) async { + final deadline = DateTime.now().add(timeout); + while (DateTime.now().isBefore(deadline)) { + if (await cond()) return; + await Future.delayed(every); + } + fail('timed out waiting for $what'); +} + +Future _account(String base, String token, String userId, String device) async { + final a = JellyfinAccounts(store: MemoryJellyfinStore(), deviceName: device); + await a.ensureLoaded(validate: false); + await a.link( + baseUrl: base, + server: const JellyfinServerInfo(id: _sid, name: 'Fake', version: '10.11.11'), + auth: JellyfinAuthResult(accessToken: token, user: JellyfinUser(id: userId, name: userId), serverId: _sid), + ); + return a; +} + +void main() { + test( + 'Up next: ⏭ to a Jellyfin episode, then to a direct link, timed per client', + timeout: const Timeout(Duration(minutes: 4)), + () async { + MediaKit.ensureInitialized(); + final preload = Platform.environment['WT_SWITCH_PRELOAD'] != '0'; + final file = await _media(); + final jelly = _FakeJellyfin(file); + await jelly.start(); + final link = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + link.listen((req) => _serveFile(req, file)); + final linkUrl = 'http://127.0.0.1:${link.port}/clip.mkv'; + final server = SyncServer(AppConfig( + port: 0, + secret: 'q', + name: 'q', + jellyfin: JellyfinCredential( + baseUrl: jelly.base, + serverId: _sid, + token: 'HOSTTOKEN', + client: 'Watch Together', + device: 'Host', + deviceId: 'host-1', + version: '9.9.9'), + )); + await server.start(); + final ytDlp = AppYtDlp(YtDlp('$_tmp/queue_switch_ytdlp', enabled: false)); + final accA = await _account(jelly.base, 'ATOKEN', 'ua', 'A-PC'); + final accC = await _account(jelly.base, 'CTOKEN', 'uc', 'C-PC'); + final accB = JellyfinAccounts(store: MemoryJellyfinStore()); + await accB.ensureLoaded(validate: false); + SessionCoordinator coord(JellyfinAccounts acc) => SessionCoordinator( + jellyfin: acc, trackPrefs: TrackPrefs(store: MemoryJellyfinStore()), ytDlp: ytDlp) + ..playback.preloadEnabled = preload; + final a = coord(accA), b = coord(accB), c = coord(accC); + final all = {'alice': a, 'guest': b, 'carol': c}; + ConnectParams params(String name, [HelloMedia? media]) => + ConnectParams(host: '127.0.0.1', port: server.boundPort, secret: 'q', name: name, media: media); + final report = []; + + Future switchOnce(String what) async { + final before = a.serverState.peek()!.mediaId; + // Time to get ready (Up next's first item, on every client). + await Future.delayed(const Duration(seconds: 8)); + if (preload) { + for (final x in all.values) { + expect(x.queue.warm.peek()?.state, WarmState.ready, reason: 'ready on every client before ⏭'); + expect(x.playback.preloadUrl, isNotNull, reason: 'preloaded on every client'); + } + } + final clock = Stopwatch()..start(); + a.queue.skip(); + await _poll(() { + final r = a.serverState.peek(); + return r != null && r.mediaId > before && r.transport == 'playing'; + }, '$what playing', every: const Duration(milliseconds: 10)); + final total = clock.elapsedMilliseconds; + final id = a.serverState.peek()!.mediaId; + report.add('$what (${preload ? 'preload' : 'cold'}): ⏭ to server playing $total ms'); + for (final e in all.entries) { + report.add(' ${e.key}: ${e.value.switchTrace.of(id)}'); + } + await _poll(() => all.values.every((x) => x.playback.player.state.playing), 'all play $what'); + // The promoted player is the session player: marks, tracks and + // the volume carried over, the clock moves. + for (final x in all.values) { + expect(x.playback.durationSecs, greaterThan(59)); + expect(x.playback.player.state.volume, 0, reason: 'volume kept'); + await _poll(() => x.playback.tracks.peek().isNotEmpty, 'tracks listed'); + } + final at = a.playback.positionSecs; + await Future.delayed(const Duration(seconds: 1)); + expect(a.playback.positionSecs, greaterThan(at + 0.5)); + } + + try { + await a.connect(params('alice', const HelloMedia( + name: 'Pilot', size: 0, duration: 0, jellyfin: JellyfinRef(serverId: _sid, itemId: _ep1, title: 'Pilot')))); + await b.connect(params('guest')); + await c.connect(params('carol')); + for (final x in all.values) { + await x.playback.player.setVolume(0); + } + await _poll(() => all.values.every((x) => x.playback.durationSecs > 59), 'all opened episode 1'); + expect(await a.queue.add(pickForRef(accA, const JellyfinRef(serverId: _sid, itemId: _ep2, title: 'Second'))!), + isNull); + expect(await a.queue.add(LinkMedia(linkUrl, relay: LinkRelay.direct)), isNull); + await _poll(() => (a.serverState.peek()?.queue.length ?? 0) == 2, 'two queued'); + for (final x in all.values) { + x.toggleReady(); + } + await _poll(() => all.values.every((x) => x.playback.player.state.playing), 'all playing episode 1'); + + await switchOnce('episode 2'); + expect(refOfStateMedia(a.serverState.peek()!.media)?.itemId, _ep2); + expect(b.jellyfinPlan.peek(), isA()); + if (preload) { + for (final x in all.values) { + expect(x.switchTrace.of(x.serverState.peek()!.mediaId), contains(SwitchTrace.preloaded)); + } + } + await switchOnce('direct link'); + expect(stateMediaUrl(a.serverState.peek()!), linkUrl); + // ignore: avoid_print + print(report.join('\n')); + + // Nothing queued: no preload. Queued and taken off again: dropped. + if (preload) { + await _poll(() => all.values.every((x) => x.playback.preloadUrl == null), 'nothing preloaded'); + expect(await a.queue.add(LinkMedia(linkUrl, relay: LinkRelay.direct)), isNull); + await _poll(() => all.values.every((x) => x.playback.preloadUrl == linkUrl), 'preloaded again'); + a.queue.remove(a.serverState.peek()!.queue.first.id); + await _poll(() => all.values.every((x) => x.playback.preloadUrl == null), 'dropped when taken off'); + } + // Preloaded everywhere: every client starts from its preload. + if (preload) { + for (final x in all.values) { + expect(x.switchTrace.of(x.serverState.peek()!.mediaId), contains(SwitchTrace.preloaded)); + } + } + } finally { + for (final x in all.values) { + await x.dispose(); + } + await server.stop(); + await jelly.stop(); + await link.close(force: true); + } + }, + ); +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 067a855..d57cb6a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -103,6 +103,9 @@ Loosely modeled on Syncplay, with a different readiness design. media still resolving, being fetched by the server or opening), and the server counts a report naming an earlier `mediaId` as 0, so the old media's buffer never passes the gate for the new one. + Besides the 1 s tick, a report goes out right after a new media is + open and, while HELD, the moment the buffer reaches `need`, so the gate + does not wait for the next tick. - **Ready gate** (HELD to PLAYING): the in-set is non-empty and every in-set member is ready, fresh and has `buffered >= need`. `need` is 4 s (`minBufferSecs`), or near the end what is left minus 2 s @@ -277,8 +280,9 @@ or a 5xx is retried, at most 3 times per read and 4 s apart at most; a fatal "No video or audio streams selected", or an EDL track that cannot be opened) before a duration is known is reported once per open (`onLoadError`), with ffmpeg's HTTP status when this player gets -ffmpeg's log (ffmpeg logs through one mpv instance per process, the -first one alive). That, a yt-dlp hook error and a failed `openUrl` become +ffmpeg's log (ffmpeg logs through one mpv instance per process: the +first one, and once that is disposed the next one created; the preload +player passes the HTTP errors it gets on to the session player). That, a yt-dlp hook error and a failed `openUrl` become the coordinator's `playerError` in plain words (`playErrorText`, core `link_probe.dart`: 412/precondition/login/cookies/sign in → "This site needs cookies — Settings → Player → yt-dlp → Cookies", 403 → "Not allowed @@ -336,14 +340,16 @@ can silently fall back to software. Player lifecycle: the `Player` and its `VideoController` live for one session: created on join, disposed on leave, replaced by `recreate()` for -"Reload video" (button or Ctrl+R; volume kept). `openUrl` opens with `play: false`; play/pause/seek/rate come only from +"Reload video" (button or Ctrl+R; volume kept), or by a promoted preload +when Up next's first item goes up (see "Getting ready" under "Up next"). `openUrl` opens with `play: false`; play/pause/seek/rate come only from `StateDiff` actions, the play gate (`localPlayAllowed`: self in-set and ready) or the user's own seeks (see Sync model). A server seek within 0.5 s of the current position is skipped. `leave()` stops the player (otherwise audio keeps playing). On media change (`mediaId` bump) the coordinator stops the previous media at once (no old picture or sound while the new one resolves, is fetched or opens; a reconnect to the media the player holds keeps it), then opens the new URL -and `StateDiff` seeks to 0, holds and resets rate. +(or promotes the preload holding it) and `StateDiff` seeks to 0, holds +and resets rate. Chapters, segments, tracks (`PlaybackController` signals, values from `playback/media_marks.dart`, no widgets; any controls UI reads them): @@ -436,8 +442,10 @@ relay stays off. The token is never logged (`toString` and config parse errors leave it out). - Available (`state.relay`) only while the current media is a ref whose - `serverId` equals the credential's. Only that item is relayed, only to a - valid viewer token: no open proxy. + `serverId` equals the credential's; `state.relayNext` likewise for Up + next's first item, which `/j` also serves so guests can preload it + (see "Getting ready" under "Up next"). Only those two items are + relayed, only to a valid viewer token: no open proxy. - Each request opens `/Videos/{id}/stream?static=true&mediaSourceId=` with `Authorization: MediaBrowser` built from exactly the configured fields (Jellyfin rewrites a token's device fields from the header), forwards @@ -873,23 +881,64 @@ every `state` and feed), the pure rules in `queue_logic.dart`, the UI in with the reason. A swap without `queueId` (a manual one) leaves the queue alone. A member who puts its own link or file up while sitting out comes back first; any other queued swap leaves the swapper out. -- **Getting ready** (`QueueController._warm`): the one item this app - puts up next is kept ready all the time, so a skip, Play now or the - end of the media starts it at once. That is the first Up next item on - the app that puts it up (`queuePerformer`), else, with Up next empty, - the first "Continues with" episode on an app that has it. It is - resolved as soon as it is first there (or this app becomes the one), - again every 30 minutes, and dropped when it stops being first: a link - is probed (`AppYtDlp.probe` with `keep`: the result stays 35 minutes - instead of 10 and a refresh replaces it only once it is in; swaps read - it from there); a Jellyfin item is resolved with this app's account +- **Getting ready** (`QueueController._warm`): every app keeps the one + item that plays next ready all the time, its own copy, so a skip, Play + now or the end of the media starts it at once. That is the first Up + next item, else, with Up next empty, the first "Continues with" + episode (on the apps that have it); while sitting out only the probe + of a link this app puts up (its swap reads it). Per + kind: a Jellyfin item is resolved with this app's account (`resolveJellyfinRef`: PlaybackInfo and the item) into `SessionCoordinator.warmJellyfin`, which `_playJellyfin` takes instead - of resolving when the same ref goes up within those 35 minutes and the - account is unchanged. A file or a server-fetched link has nothing to - get ready. One item, one app, nothing on the server. Its row shows a - dot on its picture's corner (a tooltip says which): hollow waiting, pulsing getting ready, filled ready, red failed - (tried again every 2 minutes). + of resolving when the same ref goes up within 35 minutes and the + account is unchanged; without an account it is the server's `/j` URL + when `state.relayNext` says the server's login passes it on (see + "Jellyfin relay"). A link everyone opens is that link; the app that + puts it up also probes it (`AppYtDlp.probe` with `keep`: kept 35 + minutes instead of 10, a refresh replaces it only once it is in; swaps + read it from there). A link passed on "through me" and a file only on + their adder's app (the link, the path; not the host's own file, which + plays from `/m`). A server-fetched link and others' relayed items have + nothing to get ready (the relay starts with the swap; see ROADMAP). It + is resolved as soon as it is first there, again every 30 minutes, and + dropped when it stops being first. + **Preload**: once resolved, the stream is opened in a second, hidden + mpv player (`PlaybackController.preload`, through + `SessionCoordinator.preloadMedia`): paused at the start, muted, with + its own video output when the session view is up (media_kit starts a + player with `vid=no`; the output decodes the first frame), the same + pins, hwdec and yt-dlp hook as the session player but a 48 MiB forward + and 8 MiB back cache (`preloadBytes`), mpv log at `error`. One at a + time; it waits until the media playing has started here + (`SessionCoordinator.settledMediaId`: opened, failed or nothing to + open), so it never competes with that open and a preload a swap is + about to take is never replaced first; a removed, reordered or replaced + first item swaps or drops it then, a manual swap to something else + keeps it (still next). A link is opened again at the 30-minute refresh + (stream URLs expire); a failed preload is not retried until then and + changes nothing (the usual open follows). When the coordinator opens + the same URL and headers (`openUrl`), the preload is promoted instead: + it becomes the session player (log, marks and track observers move + over, volume, unmute, the full cache pins, shaders), the view swaps to + its video output (`generation`), and the old player is disposed once + the view let go of its texture (as on leave). Its buffer comes along, + so the report sent right after the open already passes the gate: the + start no longer waits for anyone's open. A promoted player is no + longer the first mpv of the process, so ffmpeg's HTTP status reaches + it only when it happens to be the instance ffmpeg logs through (load + errors themselves are still detected; see "Media"). Off with env + `SYNC_RELAY_PRELOAD=0`. Measured (`queue_switch_e2e_test`, a fake + Jellyfin and link host answering after 120 ms at 24 Mbit/s, two + account holders and a `/j` guest): ⏭ to the server playing 59 to + 86 ms for the next episode and 35 ms for a direct link, against about + 1.5 s cold; at 300 ms and 12 Mbit/s 52 and 103 ms against about 2.9 s. + `SwitchTrace` (env `SYNC_RELAY_TRACE_SWITCH=1` prints it) times each + phase of a media change: state, resolved, open, opened or preloaded, + buffered, playing. + The row of the item shows a dot on its picture's corner (a tooltip + says which): hollow waiting, pulsing getting ready (resolving, then + preloading its first 4 s), filled ready, red failed (tried again every + 2 minutes). - **Continues with**: not queued, computed here like the next episode: with an own-account Jellyfin episode playing, the next three episodes of the series (`/Shows/{id}/Episodes`, cached), greyed under Up next, diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index 1cc877a..68ab928 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -39,7 +39,9 @@ the dedicated binary. Prints `READY port=... name=...` and stops on SIGINT or SIGTERM. Then Join from one or more apps at `127.0.0.1:8080`. Or just Host from the app. mpv env override: `SYNC_RELAY_HWDEC` (wins over the Settings choice). `SYNC_RELAY_VO` has no -effect with embedded video: media_kit forces `vo=libmpv`. +effect with embedded video: media_kit forces `vo=libmpv`. `SYNC_RELAY_PRELOAD=0` +turns Up next's preload player off; `SYNC_RELAY_TRACE_SWITCH=1` prints +the timings of each media change (`SwitchTrace`). Harnesses in `tool/` (own `pubspec.yaml`, `dart run tool/.dart`): `range_logger` (serve a file and log every Range request), `shaper` @@ -52,7 +54,10 @@ and `e2e_check` (simulated viewers against a live server). - `cd apps/desktop && flutter test test/`. - Real-mpv tests: `sync_playback_e2e`, `coord_presence`, `video_reload_e2e`, `seek_drag_e2e`, `end_of_file_e2e`, `readahead_e2e`, `jellyfin_e2e` (a - local fake Jellyfin over HTTP; tests never talk to a real Jellyfin). The sync and + local fake Jellyfin over HTTP; tests never talk to a real Jellyfin), + `queue_switch_e2e` (times Up next switches against a delayed, rate-capped + fake Jellyfin and link host; `WT_SWITCH_PRELOAD=0` for the cold path, + `WT_SWITCH_DELAY_MS` / `WT_SWITCH_RATE_KBS` for the link). The sync and end-of-file ones expect `~/Videos/chiyo.mp4`; others generate media with ffmpeg into `WT_TEST_TMP`. Headless mpv (no video output attached) won't open video-only files, so test media needs an audio track. Tests always log diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 9b2e8d6..ab695f0 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -1,4 +1,4 @@ -# Protocol (proto = 12) +# Protocol (proto = 13) Wire format only. Behavior (gate, clock, timeouts) is in [ARCHITECTURE.md](ARCHITECTURE.md). Source of truth: @@ -21,7 +21,8 @@ with different `proto` refuse each other at `hello` (`proto mismatch`). - `GET /health` returns JSON (`ok`, `name`, `proto`, `viewers`, `bytesServed`, `uptimeSecs`; unauthenticated). `GET /m///` serves the host's file (Range table in ARCHITECTURE.md). - `GET /j///` relays the current Jellyfin media (see + `GET /j///` relays the current Jellyfin media and + Up next's first one (see "Jellyfin media" below). `/r///...` is the relay of a member's media (see "Relay" below). `GET /a///` serves a member's picture (see "Avatars" below), `GET /i///` @@ -30,7 +31,7 @@ with different `proto` refuse each other at `hello` (`proto mismatch`). ## Client to server ``` -{"t":"hello","proto":12,"secret":"","user":{"name":"","avatar":""}, +{"t":"hello","proto":13,"secret":"","user":{"name":"","avatar":""}, "media":{"name":"","size":0,"duration":0,"url":null},"hostKey":"...", "resume":""} {"t":"ready","ready":true,"source":"manual"} source: manual | auto @@ -133,7 +134,8 @@ Field notes: "media":{name,size,duration,url?,jellyfin?,relay?,swapBy?,art?}|absent, "members":[{"id","name","avatarHash"?,"ready","inSet","buffered","stale","jf"?, "host"?,"out"?,"error"?,"rate"?}], - "seekBy":"m1","seekSeq":3,"relay":true,"queue":[]?,"queuePlay":"q3"?} + "seekBy":"m1","seekSeq":3,"relay":true,"relayNext":true,"queue":[]?, + "queuePlay":"q3"?} {"t":"pong","c":1767225600000,"s":1767225600012} {"t":"feed","actor":"","action":"","target":"","reason":"","at":""} {"t":"chat","text":"...","from":""} @@ -159,7 +161,9 @@ Field notes: member left, and for the media the server started with). `media.art`: see "Media art". `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 + relay the current Jellyfin media. `relayNext` (omitted when false, proto + 13): it can relay Up next's first item too (a Jellyfin ref on the same + login), so viewers without access preload it from `/j`. `members[].jf` is the member's last `report.jf`, omitted when it never sent one. `members[].avatarHash` (omitted when the member has no picture) names the picture's bytes (see "Avatars"). `members[].host` (omitted when false): the member sent the @@ -243,12 +247,13 @@ state media.art = {"poster":"","banner":""} without one plays the relay URL when `state.relay` is true: `welcome.relayUrl`, or after a swap built like the `/m` one from the host it dialled: `http:///j/default//`. -- `GET /j///`: the server fetches +- `GET /j///`: for the current item or, when + `state.relayNext`, Up next's first one, the server fetches `/Videos//stream?static=true&mediaSourceId=` (the original file) with its own credential and streams it through. `Range` is forwarded as sent; status, `Content-Range`, `Content-Length`, `Content-Type` and `Accept-Ranges` come back as Jellyfin sent them. Errors: 401 bad token, - 403 not the current item, 404 no relay for the current media, + 403 neither of those items, 404 no relay for either, 410 unknown session, 502 Jellyfin unreachable or rejecting the server's credential. Bytes count toward `bytesServed`. - Pasted Jellyfin links sent as plain `url` stay plain links. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 6bc5083..d6624a0 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -56,6 +56,25 @@ off by deleting it in the same commit as the code. Wire changes also edit 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. + - Preload relayed Up next items (see ARCHITECTURE "Getting ready"): + today a link passed on "through me", another member's file, a + server-fetched link and the host's own file are opened cold by + everyone but their source when they go up (their relay or `/m` + starts with the swap), so the gate waits for the slowest open. Plan + (wire change): the server holds a second, small relay for Up next's + first item, keyed by its queue id. Its adder's app dials a second + source for it as soon as it is first (`/r/.../source` naming the + queue id); a server-fetched item starts its fetch then; + `state.queue[0].relay` (`live`, `tracks`) tells viewers when and + how to open `/r///q//` in their preload + player. The buffer keeps only the head and tail pins plus about + 32 MiB ahead of the viewers until the swap, pulls nothing while + nobody preloads, and is dropped when the item stops being first. A + swap with `queueId` turns it into the media's relay (same buffer; + the `q/` URL keeps working for that media, so the preloads match + and are promoted). The host's file: the hosting app hands the + server the first item's path early and `/m` serves it under a + `next/` name until the swap. 8. **Jellyfin** (epic, below). diff --git a/packages/core/lib/src/constants.dart b/packages/core/lib/src/constants.dart index f5686c0..8efe4d5 100644 --- a/packages/core/lib/src/constants.dart +++ b/packages/core/lib/src/constants.dart @@ -1,5 +1,5 @@ abstract class RelayConstants { - static const int protoVersion = 12; + static const int protoVersion = 13; // Relay (`/r/...`): a member's bytes through the server's buffer. static const int relaySliceBytes = 1 << 20; diff --git a/packages/core/lib/src/protocol.dart b/packages/core/lib/src/protocol.dart index 8386c85..272d129 100644 --- a/packages/core/lib/src/protocol.dart +++ b/packages/core/lib/src/protocol.dart @@ -711,6 +711,10 @@ class StateBroadcast { /// The server can relay the current Jellyfin media (`/j/...`) for viewers /// without their own access to it. final bool relay; + /// The server can relay Up next's first item the same way (it is a + /// Jellyfin ref on the server's login), so viewers without access get + /// it ready before it goes up. + final bool relayNext; /// Up next, in order (see `QueueItem`); empty when nothing is queued. final List queue; /// The queued item someone asked to play now, waiting for the member @@ -727,6 +731,7 @@ class StateBroadcast { this.seekBy, this.seekSeq, this.relay = false, + this.relayNext = false, this.queue = const [], this.queuePlay, }); @@ -743,6 +748,7 @@ class StateBroadcast { if (seekBy != null) 'seekBy': seekBy, if (seekSeq != null) 'seekSeq': seekSeq, if (relay) 'relay': true, + if (relayNext) 'relayNext': true, if (queue.isNotEmpty) 'queue': [for (final q in queue) q.toJson()], if (queuePlay != null) 'queuePlay': queuePlay, }; @@ -767,6 +773,7 @@ class StateBroadcast { seekBy: json['seekBy'] as String?, seekSeq: json['seekSeq'] == null ? null : finiteInt(json['seekSeq'], 'seekSeq'), relay: (json['relay'] as bool?) ?? false, + relayNext: json['relayNext'] == true, queue: json['queue'] is List ? (json['queue'] as List).map(QueueItem.fromJson).whereType().toList() : const [], diff --git a/packages/core/lib/src/sync_engine.dart b/packages/core/lib/src/sync_engine.dart index d36480b..9654394 100644 --- a/packages/core/lib/src/sync_engine.dart +++ b/packages/core/lib/src/sync_engine.dart @@ -535,7 +535,7 @@ class SyncEngine { /// [relay]: whether the server can relay the current Jellyfin media; the /// engine does not know the server's credential. - StateBroadcast broadcast({bool relay = false}) { + StateBroadcast broadcast({bool relay = false, bool relayNext = false}) { final bySeek = _seekEventId == eventId; return StateBroadcast( eventId: eventId, @@ -554,6 +554,7 @@ class SyncEngine { seekBy: bySeek ? _seekBy : null, seekSeq: bySeek ? _seekSeq : null, relay: relay, + relayNext: relayNext, queue: queue.items, queuePlay: queue.play, ); diff --git a/packages/core/test/jellyfin_ref_test.dart b/packages/core/test/jellyfin_ref_test.dart index 7d63875..ed56ff7 100644 --- a/packages/core/test/jellyfin_ref_test.dart +++ b/packages/core/test/jellyfin_ref_test.dart @@ -66,13 +66,16 @@ void main() { MemberView(id: 'm2', name: 'b', ready: false, inSet: false, buffered: 0, stale: false), ], relay: true, + relayNext: true, ); final json = wire(state.toJson()); expect(json['relay'], true); + expect(json['relayNext'], true); expect((json['members'] as List)[0], containsPair('jf', {'abc123': false})); expect((json['members'] as List)[1], isNot(contains('jf'))); final back = StateBroadcast.fromJson(json); expect(back.relay, isTrue); + expect(back.relayNext, isTrue); expect(back.members[0].jf, {'abc123': false}); expect(back.members[1].jf, isEmpty); expect(HelloMedia.fromJson(back.media!).jellyfin, ref); @@ -80,7 +83,9 @@ void main() { final plain = StateBroadcast( eventId: 1, transport: 'held', position: 0, serverTime: DateTime.utc(2026), mediaId: 1, members: const []); expect(plain.toJson().containsKey('relay'), isFalse); + expect(plain.toJson().containsKey('relayNext'), isFalse); expect(StateBroadcast.fromJson(wire(plain.toJson())).relay, isFalse); + expect(StateBroadcast.fromJson(wire(plain.toJson())).relayNext, isFalse); }); test('welcome relayUrl round-trips, omitted when null', () { diff --git a/packages/server/lib/src/jellyfin_relay.dart b/packages/server/lib/src/jellyfin_relay.dart index fd5691d..704c3d2 100644 --- a/packages/server/lib/src/jellyfin_relay.dart +++ b/packages/server/lib/src/jellyfin_relay.dart @@ -25,8 +25,9 @@ Response _plain(int status) => Response(status, body: Body.fromString('status $s /// `GET /j///`: relays byte ranges of the current /// Jellyfin media from [credential]'s server to a viewer, for guests without -/// their own access. Only the active ref's item on the credential's server, -/// only for a valid viewer token: no open proxy. Range goes up as sent; +/// their own access. Only the active ref's item, or Up next's first one +/// ([next], so viewers can get it ready), on the credential's server, only +/// for a valid viewer token: no open proxy. Range goes up as sent; /// status, Content-Range, Content-Length, Content-Type and Accept-Ranges /// come back as Jellyfin sent them (a 401/403 from Jellyfin is the server's /// credential failing, so the viewer gets 502). Bytes stream through without @@ -39,6 +40,7 @@ void mountJellyfinRelayRoute( SessionMap sessions, JellyfinCredential? Function() credential, HttpClient Function() client, { + JellyfinRef? Function()? next, String pattern = '/j/:sid/:token/:itemId', void Function(String token, MediaLog log)? onLog, Duration timeout = const Duration(seconds: 20), @@ -56,12 +58,14 @@ void mountJellyfinRelayRoute( default: return _plain(401); } - final ref = session.media?.jellyfin; final cred = credential(); - if (ref == null || cred == null || cred.serverId.isEmpty || cred.serverId != ref.serverId) { - return _plain(404); - } - if (!_sameItemId(itemId, ref.itemId)) return _plain(403); + if (cred == null || cred.serverId.isEmpty) return _plain(404); + JellyfinRef? onLogin(JellyfinRef? r) => r != null && cred.serverId == r.serverId ? r : null; + final current = onLogin(session.media?.jellyfin); + final upNext = onLogin(next?.call()); + if (current == null && upNext == null) return _plain(404); + final ref = [current, upNext].firstWhere((r) => r != null && _sameItemId(itemId, r.itemId), orElse: () => null); + if (ref == null) return _plain(403); final range = req.headers['range']?.join(','); final HttpClientRequest upReq; diff --git a/packages/server/lib/src/server_hub.dart b/packages/server/lib/src/server_hub.dart index 5026bb9..1eb437c 100644 --- a/packages/server/lib/src/server_hub.dart +++ b/packages/server/lib/src/server_hub.dart @@ -30,8 +30,16 @@ class ServerHub { : engine = engine ?? SyncEngine(); /// The current media is a Jellyfin ref on the server [jellyfin] logs in to. - bool get relayAvailable { - final ref = session.media?.jellyfin; + bool get relayAvailable => _onLogin(session.media?.jellyfin); + + /// Up next's first item is a Jellyfin ref on that server: `/j` serves it + /// too, so viewers without access get it ready. + bool get relayNextAvailable => _onLogin(nextJellyfinRef); + + /// Up next's first item's Jellyfin ref (null: none, or not a ref). + JellyfinRef? get nextJellyfinRef => engine.queue.items.firstOrNull?.media.jellyfin; + + bool _onLogin(JellyfinRef? ref) { final cred = jellyfin; return ref != null && cred != null && cred.serverId.isNotEmpty && cred.serverId == ref.serverId; } @@ -53,7 +61,7 @@ class ServerHub { : '$head: $why'; } - StateBroadcast stateBroadcast() => engine.broadcast(relay: relayAvailable); + StateBroadcast stateBroadcast() => engine.broadcast(relay: relayAvailable, relayNext: relayNextAvailable); /// The relay of the current media (`/r/...`), if it is relayed. ActiveRelay? relay; diff --git a/packages/server/lib/src/sync_server.dart b/packages/server/lib/src/sync_server.dart index 417aabb..61dfa56 100644 --- a/packages/server/lib/src/sync_server.dart +++ b/packages/server/lib/src/sync_server.dart @@ -98,7 +98,7 @@ class SyncServer { mountAvatarRoute(_app, hub); mountArtRoute(_app, hub); mountJellyfinRelayRoute( - _app, sessions, () => hub.jellyfin, () => _relayClient ??= HttpClient()); + _app, sessions, () => hub.jellyfin, () => _relayClient ??= HttpClient(), next: () => hub.nextJellyfinRef); // advertiseHost is only an explicit override; otherwise echo what the client dialled. String authority(String dialledHost) => config.advertiseHost.isNotEmpty ? '${config.advertiseHost}:$boundPort' diff --git a/packages/server/test/jellyfin_relay_test.dart b/packages/server/test/jellyfin_relay_test.dart index 83cb176..34e9d16 100644 --- a/packages/server/test/jellyfin_relay_test.dart +++ b/packages/server/test/jellyfin_relay_test.dart @@ -11,6 +11,7 @@ const _serverId = 'fakeserver01'; const _item = '0123456789abcdef0123456789abcdef'; const _source = 'fedcba9876543210fedcba9876543210'; const _slowItem = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'; +const _nextItem = 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb'; const _token = 'JF-SECRET-TOKEN'; const _cred = JellyfinCredential( @@ -368,6 +369,28 @@ void main() { expect(jf.uris, isEmpty, reason: 'nothing reached Jellyfin'); }); + test('Up next first item on the login: relayNext, and /j serves it too', () async { + await setUpRef(_item); + final a = clients.first; + expect(a.lastState!.containsKey('relayNext'), isFalse); + expect((await relayGet(token, _nextItem)).statusCode, 403); + a.send(const QueueIntent.add(QueueItem( + media: HelloMedia( + name: 'Next', + size: 0, + duration: 0, + jellyfin: JellyfinRef(serverId: _serverId, itemId: _nextItem, mediaSourceId: _source)), + how: QueueHow.jellyfin, + title: 'Next')) + .toJson()); + await _until(() => a.lastState?['relayNext'] == true, 'relayNext'); + final res = await relayGet(token, _nextItem, range: 'bytes=0-99'); + expect(res.statusCode, 206); + expect(await readAll(res), bytes.sublist(0, 100)); + expect((await relayGet(token, _item, range: 'bytes=0-9')).statusCode, 206, reason: 'the current one still'); + expect((await relayGet(token, _slowItem)).statusCode, 403, reason: 'nothing else'); + }); + test('non-Jellyfin media: 404', () async { await startServer(cred()); final (a, w) = await join('alice'); diff --git a/packages/server/test/session_test.dart b/packages/server/test/session_test.dart index 33470c4..27cf67c 100644 --- a/packages/server/test/session_test.dart +++ b/packages/server/test/session_test.dart @@ -65,7 +65,7 @@ void main() { group('core welcome + proto', () { test('proto 12: avatars by hash, profile intent', () { - expect(RelayConstants.protoVersion, 12); + expect(RelayConstants.protoVersion, 13); const v = MemberView(id: 'm1', name: 'a', ready: false, inSet: false, buffered: 0, stale: false); expect(v.toJson().containsKey('avatarHash'), isFalse); expect(v.toJson().containsKey('avatar'), isFalse);