haeaeae/plezy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(watch-together): flat engine callbacks instead of bundle classes

Browse Source 
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.
This commit is contained in:
edde746
2026-08-17 19:02:22 +02:00
parent 99b0528623
commit 0e1af9faaf
5 changed files with 81 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/watch_together/services/guest_playback_reconciler.dart
+27 -35
View File
@@ -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<String> 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<String> 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);
}
// ---------------------------------------------------------------------
lib/watch_together/services/host_playback_coordinator.dart
+23 -31
View File
@@ -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<String> peerIds)? onWaitingOnChanged;
/// The safety timeout excused these peers and the room resumed.
final void Function(List<String> 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<String> peerIds)? onWaitingOnChanged;
/// The safety timeout excused these peers and the room resumed.
final void Function(List<String> 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);
lib/watch_together/services/watch_together_controller.dart
+10 -14
View File
@@ -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();
test/watch_together/guest_playback_reconciler_test.dart
+12 -16
View File
@@ -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 = <bool>[];
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 = <bool>[];
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 = <String>[];
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 = <String>[];
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());
test/watch_together/host_playback_coordinator_test.dart
+9 -7
View File
@@ -17,7 +17,8 @@ class _Harness {
_Harness(
FakeAsync async, {
ControlMode controlMode = ControlMode.hostOnly,
HostCoordinatorCallbacks callbacks = const HostCoordinatorCallbacks(),
void Function(List<String>)? 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 = <List<String>>[];
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);