From 0e1af9faaf6052baca0eeed8a962493a58cb70f4 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Mon, 17 Aug 2026 01:50:31 +0200 Subject: [PATCH] refactor(watch-together): flat engine callbacks instead of bundle classes HostCoordinatorCallbacks and GuestReconcilerCallbacks existed only to be unpacked by the controller through ten forwarding lambdas into its own flat fields. The engines now take individual nullable callbacks; the controller keeps its late-binding lambdas (the provider assigns the public fields after construction), and the test harnesses forward the callbacks they exercise. --- .../services/guest_playback_reconciler.dart | 62 ++++++++----------- .../services/host_playback_coordinator.dart | 54 +++++++--------- .../services/watch_together_controller.dart | 24 +++---- .../guest_playback_reconciler_test.dart | 28 ++++----- .../host_playback_coordinator_test.dart | 16 ++--- 5 files changed, 81 insertions(+), 103 deletions(-) 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);