diff --git a/lib/watch_together/services/guest_playback_reconciler.dart b/lib/watch_together/services/guest_playback_reconciler.dart index 189f07be7..b134e3b24 100644 --- a/lib/watch_together/services/guest_playback_reconciler.dart +++ b/lib/watch_together/services/guest_playback_reconciler.dart @@ -8,33 +8,6 @@ import '../primitives.dart'; import 'attached_player.dart'; import 'clock_sync.dart'; -/// Callbacks the reconciler surfaces to the provider/UI layer. -class GuestReconcilerCallbacks { - /// The host's state names media we don't have loaded — navigate/reload. - /// Fires on EVERY such state (the heartbeat is the retry channel for - /// failed switches); the provider's dispatcher dedups. - final void Function(String ratingKey, String serverId, String? mediaTitle)? onMediaSwitchNeeded; - - final void Function(ControlMode mode)? onControlModeChanged; - final void Function(PlaybackPhase phase)? onPhaseChanged; - final void Function(List waitingOn)? onWaitingOnChanged; - - /// A hard correction is in flight (drives the syncing pill). - final void Function(bool correcting)? onCorrectingChanged; - - /// Another peer caused a transition (drives action toasts). - final void Function(String peerId, PlaybackActionHint hint)? onRemoteAction; - - const GuestReconcilerCallbacks({ - this.onMediaSwitchNeeded, - this.onControlModeChanged, - this.onPhaseChanged, - this.onWaitingOnChanged, - this.onCorrectingChanged, - this.onRemoteAction, - }); -} - /// Guest-side reconciliation loop: converges the local player onto the /// host's authoritative [PlaybackState]. /// @@ -50,11 +23,31 @@ class GuestPlaybackReconciler { required this.myPeerId, required this._sendToHost, required ClockSync clockSync, - this._callbacks = const GuestReconcilerCallbacks(), + this.onMediaSwitchNeeded, + this.onControlModeChanged, + this.onPhaseChanged, + this.onWaitingOnChanged, + this.onCorrectingChanged, + this.onRemoteAction, int Function()? nowMs, }) : _clock = clockSync, _nowMs = nowMs ?? watchTogetherSystemNowMs; + /// The host's state names media we don't have loaded — navigate/reload. + /// Fires on EVERY such state (the heartbeat is the retry channel for + /// failed switches); the provider's dispatcher dedups. + final void Function(String ratingKey, String serverId, String? mediaTitle)? onMediaSwitchNeeded; + + final void Function(ControlMode mode)? onControlModeChanged; + final void Function(PlaybackPhase phase)? onPhaseChanged; + final void Function(List waitingOn)? onWaitingOnChanged; + + /// A hard correction is in flight (drives the syncing pill). + final void Function(bool correcting)? onCorrectingChanged; + + /// Another peer caused a transition (drives action toasts). + final void Function(String peerId, PlaybackActionHint hint)? onRemoteAction; + // Tuning constants. static const int tickMs = 500; static const int deadbandMs = 350; @@ -75,7 +68,6 @@ class GuestPlaybackReconciler { final String myPeerId; final void Function(SyncMessage message) _sendToHost; final ClockSync _clock; - final GuestReconcilerCallbacks _callbacks; final int Function() _nowMs; PlaybackState? _latestState; @@ -234,18 +226,18 @@ class GuestPlaybackReconciler { if (state.controlMode != _reportedControlMode) { _reportedControlMode = state.controlMode; - _callbacks.onControlModeChanged?.call(state.controlMode); + onControlModeChanged?.call(state.controlMode); } if (state.phase != _reportedPhase) { _reportedPhase = state.phase; - _callbacks.onPhaseChanged?.call(state.phase); + onPhaseChanged?.call(state.phase); } if (!orderedStringListsEqual(state.waitingOn, _reportedWaitingOn)) { _reportedWaitingOn = state.waitingOn; - _callbacks.onWaitingOnChanged?.call(state.waitingOn); + onWaitingOnChanged?.call(state.waitingOn); } if (state.actionHint != null && state.actorPeerId != null && state.actorPeerId != myPeerId) { - _callbacks.onRemoteAction?.call(state.actorPeerId!, state.actionHint!); + onRemoteAction?.call(state.actorPeerId!, state.actionHint!); } // Close the optimistic window only on an explicit transition (the host @@ -266,7 +258,7 @@ class GuestPlaybackReconciler { // else) — hand off to the switch flow on every state so a failed switch // retries on the next heartbeat. The provider's dispatcher dedups. if (_attachedMediaKey == null || state.mediaKey != _attachedMediaKey) { - _callbacks.onMediaSwitchNeeded?.call(state.ratingKey, state.serverId, state.mediaTitle); + onMediaSwitchNeeded?.call(state.ratingKey, state.serverId, state.mediaTitle); return; } @@ -562,7 +554,7 @@ class GuestPlaybackReconciler { void _setCorrecting(bool value) { if (_correcting == value) return; _correcting = value; - _callbacks.onCorrectingChanged?.call(value); + onCorrectingChanged?.call(value); } // --------------------------------------------------------------------- diff --git a/lib/watch_together/services/host_playback_coordinator.dart b/lib/watch_together/services/host_playback_coordinator.dart index 50d405825..7674b5bef 100644 --- a/lib/watch_together/services/host_playback_coordinator.dart +++ b/lib/watch_together/services/host_playback_coordinator.dart @@ -8,28 +8,6 @@ import '../models/watch_session.dart'; import '../primitives.dart'; import 'attached_player.dart'; -/// Callbacks the coordinator surfaces to the provider/UI layer. -class HostCoordinatorCallbacks { - /// Phase transitions (drives the waiting pill and chrome). - final void Function(PlaybackPhase phase)? onPhaseChanged; - - /// The set of peers the room is waiting on changed. - final void Function(List peerIds)? onWaitingOnChanged; - - /// The safety timeout excused these peers and the room resumed. - final void Function(List peerIds)? onResumedWithout; - - /// A guest's control request was applied (drives action toasts). - final void Function(String peerId, PlaybackActionHint hint)? onRemoteAction; - - const HostCoordinatorCallbacks({ - this.onPhaseChanged, - this.onWaitingOnChanged, - this.onResumedWithout, - this.onRemoteAction, - }); -} - /// Host-side policy engine: owns the authoritative [PlaybackState]. /// /// Inputs are local player signals (via [AttachedPlayer]'s intent-classified @@ -45,10 +23,25 @@ class HostPlaybackCoordinator { required this.myPeerId, required this._controlMode, required this._sendState, - this._callbacks = const HostCoordinatorCallbacks(), + this.onPhaseChanged, + this.onWaitingOnChanged, + this.onResumedWithout, + this.onRemoteAction, int Function()? nowMs, }) : _nowMs = nowMs ?? watchTogetherSystemNowMs; + /// Phase transitions (drives the waiting pill and chrome). + final void Function(PlaybackPhase phase)? onPhaseChanged; + + /// The set of peers the room is waiting on changed. + final void Function(List peerIds)? onWaitingOnChanged; + + /// The safety timeout excused these peers and the room resumed. + final void Function(List peerIds)? onResumedWithout; + + /// A guest's control request was applied (drives action toasts). + final void Function(String peerId, PlaybackActionHint hint)? onRemoteAction; + // Tuning constants. static const int stallGraceMs = 500; static const int recoveryHysteresisMs = 400; @@ -66,7 +59,6 @@ class HostPlaybackCoordinator { final String myPeerId; final void Function(PlaybackState state, {String? toPeerId}) _sendState; - final HostCoordinatorCallbacks _callbacks; final int Function() _nowMs; ControlMode _controlMode; @@ -450,7 +442,7 @@ class HostPlaybackCoordinator { if (_phase == PlaybackPhase.playing) return; _intendedPlaying = true; _pendingActor = actor; - if (actor != myPeerId) _callbacks.onRemoteAction?.call(actor, PlaybackActionHint.play); + if (actor != myPeerId) onRemoteAction?.call(actor, PlaybackActionHint.play); final player = _player; if (!_localReady) { @@ -483,7 +475,7 @@ class HostPlaybackCoordinator { _pendingActor = null; _cancelPendingStart(); _cancelSafety(); - if (actor != myPeerId) _callbacks.onRemoteAction?.call(actor, PlaybackActionHint.pause); + if (actor != myPeerId) onRemoteAction?.call(actor, PlaybackActionHint.pause); final player = _player; if (player != null && player.playing) { @@ -501,7 +493,7 @@ class HostPlaybackCoordinator { if (player == null || !player.seekable) return; final durationMs = player.duration.inMilliseconds; if (durationMs <= 0 || targetMs < 0 || targetMs > durationMs) return; - _callbacks.onRemoteAction?.call(actor, PlaybackActionHint.seek); + onRemoteAction?.call(actor, PlaybackActionHint.seek); unawaited( player.seek(Duration(milliseconds: targetMs)).then((didSeek) { if (didSeek) _afterHostSeek(targetMs, actor: actor); @@ -521,7 +513,7 @@ class HostPlaybackCoordinator { void _applyRemoteRate(double rate, {required String actor}) { final player = _player; if (player == null || !rate.isFinite || rate < _minimumRemoteRate || rate > _maximumRemoteRate) return; - _callbacks.onRemoteAction?.call(actor, PlaybackActionHint.rate); + onRemoteAction?.call(actor, PlaybackActionHint.rate); unawaited( player.setRate(rate).then((didSet) { if (!didSet) return; @@ -695,7 +687,7 @@ class HostPlaybackCoordinator { _excused.addAll(gating); _stalledPeers.removeAll(gating); appLogger.w('WatchTogether: Resuming without ${gating.join(', ')} after ${safetyTimeoutMs ~/ 1000}s'); - _callbacks.onResumedWithout?.call(gating.toList()..sort()); + onResumedWithout?.call(gating.toList()..sort()); _scheduleAllReadyCheck(0); }); } @@ -763,7 +755,7 @@ class HostPlaybackCoordinator { void _setPhase(PlaybackPhase phase) { if (_phase == phase) return; _phase = phase; - _callbacks.onPhaseChanged?.call(phase); + onPhaseChanged?.call(phase); _restartHeartbeat(); } @@ -806,7 +798,7 @@ class HostPlaybackCoordinator { final previousWaiting = _lastBroadcast?.waitingOn ?? const []; _lastBroadcast = state; if (!orderedStringListsEqual(previousWaiting, waitingOn)) { - _callbacks.onWaitingOnChanged?.call(waitingOn); + onWaitingOnChanged?.call(waitingOn); } } _sendState(state, toPeerId: toPeerId); diff --git a/lib/watch_together/services/watch_together_controller.dart b/lib/watch_together/services/watch_together_controller.dart index d52c5d7f5..f40782459 100644 --- a/lib/watch_together/services/watch_together_controller.dart +++ b/lib/watch_together/services/watch_together_controller.dart @@ -35,12 +35,10 @@ class WatchTogetherController { myPeerId: peerService.myPeerId ?? '', controlMode: session.controlMode, sendState: _sendState, - callbacks: HostCoordinatorCallbacks( - onPhaseChanged: (phase) => onPhaseChanged?.call(phase), - onWaitingOnChanged: (peers) => onWaitingOnChanged?.call(peers), - onResumedWithout: (peers) => onResumedWithout?.call(peers), - onRemoteAction: (peer, hint) => onRemoteAction?.call(peer, hint), - ), + onPhaseChanged: (phase) => onPhaseChanged?.call(phase), + onWaitingOnChanged: (peers) => onWaitingOnChanged?.call(peers), + onResumedWithout: (peers) => onResumedWithout?.call(peers), + onRemoteAction: (peer, hint) => onRemoteAction?.call(peer, hint), nowMs: _nowMs, ); } else { @@ -49,14 +47,12 @@ class WatchTogetherController { myPeerId: peerService.myPeerId ?? '', sendToHost: _sendToHost, clockSync: _clockSync!, - callbacks: GuestReconcilerCallbacks( - onMediaSwitchNeeded: (ratingKey, serverId, title) => onMediaStateReceived?.call(ratingKey, serverId, title), - onControlModeChanged: (mode) => onControlModeReceived?.call(mode), - onPhaseChanged: (phase) => onPhaseChanged?.call(phase), - onWaitingOnChanged: (peers) => onWaitingOnChanged?.call(peers), - onCorrectingChanged: (correcting) => onCorrectingChanged?.call(correcting), - onRemoteAction: (peer, hint) => onRemoteAction?.call(peer, hint), - ), + onMediaSwitchNeeded: (ratingKey, serverId, title) => onMediaStateReceived?.call(ratingKey, serverId, title), + onControlModeChanged: (mode) => onControlModeReceived?.call(mode), + onPhaseChanged: (phase) => onPhaseChanged?.call(phase), + onWaitingOnChanged: (peers) => onWaitingOnChanged?.call(peers), + onCorrectingChanged: (correcting) => onCorrectingChanged?.call(correcting), + onRemoteAction: (peer, hint) => onRemoteAction?.call(peer, hint), nowMs: _nowMs, ); _clockSync!.start(); diff --git a/test/watch_together/guest_playback_reconciler_test.dart b/test/watch_together/guest_playback_reconciler_test.dart index fb82b0ad2..9a72b8b40 100644 --- a/test/watch_together/guest_playback_reconciler_test.dart +++ b/test/watch_together/guest_playback_reconciler_test.dart @@ -12,14 +12,19 @@ import '../test_helpers/watch_together_fakes.dart'; const _epochMs = 1000000; class _Harness { - _Harness(this.async, {GuestReconcilerCallbacks callbacks = const GuestReconcilerCallbacks()}) { + _Harness( + this.async, { + void Function(bool)? onCorrectingChanged, + void Function(String, String, String?)? onMediaSwitchNeeded, + }) { player = FakeSyncPlayer(position: const Duration(minutes: 2)); clock = ClockSync(sendPing: pings.add, nowMs: nowMs); reconciler = GuestPlaybackReconciler( myPeerId: 'guest', sendToHost: outgoing.add, clockSync: clock, - callbacks: callbacks, + onCorrectingChanged: onCorrectingChanged, + onMediaSwitchNeeded: onMediaSwitchNeeded, nowMs: nowMs, ); attached = AttachedPlayer(player: player, onLost: () {}, nowMs: nowMs); @@ -195,7 +200,7 @@ void main() { test('large drift hard-seeks with lead, settle window, and cooldown', () { fakeAsync((async) { final correcting = []; - final h = _Harness(async, callbacks: GuestReconcilerCallbacks(onCorrectingChanged: correcting.add)); + final h = _Harness(async, onCorrectingChanged: correcting.add); h.attachReady(); h.player.emitPlaying(true); async.flushMicrotasks(); @@ -230,7 +235,7 @@ void main() { test('settle falls back to the timeout when no playback-restart arrives', () { fakeAsync((async) { final correcting = []; - final h = _Harness(async, callbacks: GuestReconcilerCallbacks(onCorrectingChanged: correcting.add)); + final h = _Harness(async, onCorrectingChanged: correcting.add); h.player.emitRestartOnSeek = false; h.attachReady(); h.player.emitPlaying(true); @@ -346,10 +351,7 @@ void main() { test('epoch mismatch hands off to the media-switch flow and stops correcting', () { fakeAsync((async) { final switches = <(String, String, String?)>[]; - final h = _Harness( - async, - callbacks: GuestReconcilerCallbacks(onMediaSwitchNeeded: (rk, sid, title) => switches.add((rk, sid, title))), - ); + final h = _Harness(async, onMediaSwitchNeeded: (rk, sid, title) => switches.add((rk, sid, title))); h.attachReady(); h.player.emitPlaying(true); async.flushMicrotasks(); @@ -367,10 +369,7 @@ void main() { test('detached guest is re-notified on every state (heartbeat retry channel)', () { fakeAsync((async) { final switches = []; - final h = _Harness( - async, - callbacks: GuestReconcilerCallbacks(onMediaSwitchNeeded: (rk, sid, title) => switches.add(rk)), - ); + final h = _Harness(async, onMediaSwitchNeeded: (rk, sid, title) => switches.add(rk)); // Never attached: every heartbeat re-offers the switch so a failed // navigation can retry (the provider's dispatcher dedups). @@ -388,10 +387,7 @@ void main() { test('attached to matching media never fires the switch callback', () { fakeAsync((async) { final switches = []; - final h = _Harness( - async, - callbacks: GuestReconcilerCallbacks(onMediaSwitchNeeded: (rk, sid, title) => switches.add(rk)), - ); + final h = _Harness(async, onMediaSwitchNeeded: (rk, sid, title) => switches.add(rk)); h.attachReady(); h.reconciler.onState(h.state()); diff --git a/test/watch_together/host_playback_coordinator_test.dart b/test/watch_together/host_playback_coordinator_test.dart index 25746c202..143584e06 100644 --- a/test/watch_together/host_playback_coordinator_test.dart +++ b/test/watch_together/host_playback_coordinator_test.dart @@ -17,7 +17,8 @@ class _Harness { _Harness( FakeAsync async, { ControlMode controlMode = ControlMode.hostOnly, - HostCoordinatorCallbacks callbacks = const HostCoordinatorCallbacks(), + void Function(List)? onResumedWithout, + void Function(String, PlaybackActionHint)? onRemoteAction, Duration duration = const Duration(minutes: 45), bool seekable = true, }) { @@ -27,7 +28,8 @@ class _Harness { myPeerId: 'host', controlMode: controlMode, sendState: (state, {toPeerId}) => sent.add((state, toPeerId)), - callbacks: callbacks, + onResumedWithout: onResumedWithout, + onRemoteAction: onRemoteAction, nowMs: nowMs, ); attached = AttachedPlayer(player: player, onLost: () {}, nowMs: nowMs); @@ -260,7 +262,7 @@ void main() { test('guest stall: room pauses, safety timeout excuses them, resume fires', () { fakeAsync((async) { final resumedWithout = >[]; - final h = _Harness(async, callbacks: HostCoordinatorCallbacks(onResumedWithout: resumedWithout.add)); + final h = _Harness(async, onResumedWithout: resumedWithout.add); h.coordinator.onPeerJoined('guest', compatible: true); h.attachForMedia(async); h.guestReports(async); @@ -418,7 +420,7 @@ void main() { final h = _Harness( async, controlMode: ControlMode.anyone, - callbacks: HostCoordinatorCallbacks(onRemoteAction: (peer, hint) => actions.add((peer, hint))), + onRemoteAction: (peer, hint) => actions.add((peer, hint)), ); h.coordinator.onPeerJoined('guest', compatible: true); h.attachForMedia(async); @@ -456,7 +458,7 @@ void main() { final h = _Harness( async, controlMode: ControlMode.anyone, - callbacks: HostCoordinatorCallbacks(onRemoteAction: (peer, hint) => actions.add((peer, hint))), + onRemoteAction: (peer, hint) => actions.add((peer, hint)), ); h.attachForMedia(async); h.hostBecomesReady(async); @@ -491,7 +493,7 @@ void main() { final h = _Harness( async, controlMode: ControlMode.anyone, - callbacks: HostCoordinatorCallbacks(onRemoteAction: (peer, hint) => actions.add((peer, hint))), + onRemoteAction: (peer, hint) => actions.add((peer, hint)), ); h.attachForMedia(async); h.hostBecomesReady(async); @@ -538,7 +540,7 @@ void main() { controlMode: ControlMode.anyone, duration: config.duration, seekable: config.seekable, - callbacks: HostCoordinatorCallbacks(onRemoteAction: (peer, hint) => actions.add((peer, hint))), + onRemoteAction: (peer, hint) => actions.add((peer, hint)), ); h.attachForMedia(async); h.hostBecomesReady(async);