refactor(watch-together): host-authoritative declarative sync protocol
Replaces the imperative play/pause/seek/positionSync message soup with a single host-authored PlaybackState (seq-ordered, anchor-extrapolated, phase machine: loading/waitingForPeers/paused/playing) that doubles as the heartbeat, plus guest status reports and host-applied control requests. Fixes the guest seek-back loop while the host loads (readiness was keyed on a pre-load !buffering snapshot and heartbeats broadcast frozen positions), adds real group buffering coordination (stall grace, scheduled simultaneous resumes, 15s safety timeout), rate-nudge drift correction with passthrough-aware seek fallback, session-scoped message handling (no lost messages during episode-switch detach gaps), and an expected-state ledger replacing the racy remote-action flag.
This commit is contained in:
edde746
@@ -0,0 +1,265 @@
|
||||
import 'package:collection/collection.dart';
|
||||
|
||||
import 'watch_session.dart';
|
||||
|
||||
/// Playback lifecycle phase broadcast by the host.
|
||||
///
|
||||
/// Serialized as the enum index — append new values only.
|
||||
enum PlaybackPhase { loading, waitingForPeers, paused, playing }
|
||||
|
||||
/// What caused a state transition (drives participant toasts).
|
||||
///
|
||||
/// Serialized as the enum index — append new values only.
|
||||
enum PlaybackActionHint { play, pause, seek, rate, mediaSwitch }
|
||||
|
||||
/// Authoritative playback state, broadcast by the host on every transition
|
||||
/// and as the periodic heartbeat. Receivers keep the highest [seq] seen and
|
||||
/// drop anything older, so missed or reordered messages self-heal on the
|
||||
/// next broadcast.
|
||||
///
|
||||
/// A `phase == playing` state whose [anchorHostTimeMs] lies in the future is
|
||||
/// a scheduled group start: [targetPositionMs] clamps elapsed time to >= 0,
|
||||
/// so peers hold at [anchorPositionMs] until the start moment and then
|
||||
/// extrapolate from a shared origin.
|
||||
class PlaybackState {
|
||||
final int seq;
|
||||
final String ratingKey;
|
||||
final String serverId;
|
||||
final String? mediaTitle;
|
||||
final PlaybackPhase phase;
|
||||
|
||||
/// Timeline-adjusted position at [anchorHostTimeMs].
|
||||
final int anchorPositionMs;
|
||||
|
||||
/// Host wall-clock time (Unix ms) the anchor was captured — or, when in
|
||||
/// the future with `phase == playing`, the scheduled group-start moment.
|
||||
final int anchorHostTimeMs;
|
||||
|
||||
final double rate;
|
||||
final ControlMode controlMode;
|
||||
|
||||
/// Peers the room is currently waiting on (readiness or buffering).
|
||||
final List<String> waitingOn;
|
||||
|
||||
/// Peer that caused this transition (host's own id for local actions).
|
||||
final String? actorPeerId;
|
||||
final PlaybackActionHint? actionHint;
|
||||
|
||||
const PlaybackState({
|
||||
required this.seq,
|
||||
required this.ratingKey,
|
||||
required this.serverId,
|
||||
required this.phase,
|
||||
required this.anchorPositionMs,
|
||||
required this.anchorHostTimeMs,
|
||||
required this.rate,
|
||||
required this.controlMode,
|
||||
this.mediaTitle,
|
||||
this.waitingOn = const [],
|
||||
this.actorPeerId,
|
||||
this.actionHint,
|
||||
});
|
||||
|
||||
String get mediaKey => mediaKeyFor(ratingKey: ratingKey, serverId: serverId);
|
||||
|
||||
static String mediaKeyFor({required String ratingKey, required String serverId}) => '$serverId:$ratingKey';
|
||||
|
||||
/// Where the room should be at [nowHostMs] (host clock).
|
||||
int targetPositionMs(int nowHostMs) {
|
||||
if (phase != PlaybackPhase.playing) return anchorPositionMs;
|
||||
final elapsed = nowHostMs - anchorHostTimeMs;
|
||||
if (elapsed <= 0) return anchorPositionMs;
|
||||
return anchorPositionMs + (elapsed * rate).round();
|
||||
}
|
||||
|
||||
PlaybackState copyWith({
|
||||
int? seq,
|
||||
String? ratingKey,
|
||||
String? serverId,
|
||||
String? mediaTitle,
|
||||
PlaybackPhase? phase,
|
||||
int? anchorPositionMs,
|
||||
int? anchorHostTimeMs,
|
||||
double? rate,
|
||||
ControlMode? controlMode,
|
||||
List<String>? waitingOn,
|
||||
String? actorPeerId,
|
||||
PlaybackActionHint? actionHint,
|
||||
}) {
|
||||
return PlaybackState(
|
||||
seq: seq ?? this.seq,
|
||||
ratingKey: ratingKey ?? this.ratingKey,
|
||||
serverId: serverId ?? this.serverId,
|
||||
mediaTitle: mediaTitle ?? this.mediaTitle,
|
||||
phase: phase ?? this.phase,
|
||||
anchorPositionMs: anchorPositionMs ?? this.anchorPositionMs,
|
||||
anchorHostTimeMs: anchorHostTimeMs ?? this.anchorHostTimeMs,
|
||||
rate: rate ?? this.rate,
|
||||
controlMode: controlMode ?? this.controlMode,
|
||||
waitingOn: waitingOn ?? this.waitingOn,
|
||||
actorPeerId: actorPeerId ?? this.actorPeerId,
|
||||
actionHint: actionHint ?? this.actionHint,
|
||||
);
|
||||
}
|
||||
|
||||
Map<String, dynamic> toMap() => {
|
||||
'q': seq,
|
||||
'rk': ratingKey,
|
||||
'sid': serverId,
|
||||
if (mediaTitle != null) 'ti': mediaTitle,
|
||||
'ph': phase.index,
|
||||
'ap': anchorPositionMs,
|
||||
'at': anchorHostTimeMs,
|
||||
'r': rate,
|
||||
'cm': controlMode.index,
|
||||
if (waitingOn.isNotEmpty) 'w': waitingOn,
|
||||
if (actorPeerId != null) 'ab': actorPeerId,
|
||||
if (actionHint != null) 'ah': actionHint!.index,
|
||||
};
|
||||
|
||||
factory PlaybackState.fromMap(Map<String, dynamic> map) {
|
||||
return PlaybackState(
|
||||
seq: map['q'] as int,
|
||||
ratingKey: map['rk'] as String,
|
||||
serverId: map['sid'] as String,
|
||||
mediaTitle: map['ti'] as String?,
|
||||
phase: _enumFromIndex(PlaybackPhase.values, map['ph'] as int) ?? PlaybackPhase.paused,
|
||||
anchorPositionMs: map['ap'] as int,
|
||||
anchorHostTimeMs: map['at'] as int,
|
||||
rate: (map['r'] as num).toDouble(),
|
||||
controlMode: _enumFromIndex(ControlMode.values, map['cm'] as int) ?? ControlMode.hostOnly,
|
||||
waitingOn: (map['w'] as List?)?.cast<String>() ?? const [],
|
||||
actorPeerId: map['ab'] as String?,
|
||||
actionHint: map['ah'] != null ? _enumFromIndex(PlaybackActionHint.values, map['ah'] as int) : null,
|
||||
);
|
||||
}
|
||||
|
||||
@override
|
||||
bool operator ==(Object other) =>
|
||||
other is PlaybackState &&
|
||||
other.seq == seq &&
|
||||
other.ratingKey == ratingKey &&
|
||||
other.serverId == serverId &&
|
||||
other.mediaTitle == mediaTitle &&
|
||||
other.phase == phase &&
|
||||
other.anchorPositionMs == anchorPositionMs &&
|
||||
other.anchorHostTimeMs == anchorHostTimeMs &&
|
||||
other.rate == rate &&
|
||||
other.controlMode == controlMode &&
|
||||
const ListEquality<String>().equals(other.waitingOn, waitingOn) &&
|
||||
other.actorPeerId == actorPeerId &&
|
||||
other.actionHint == actionHint;
|
||||
|
||||
@override
|
||||
int get hashCode => Object.hash(
|
||||
seq,
|
||||
ratingKey,
|
||||
serverId,
|
||||
mediaTitle,
|
||||
phase,
|
||||
anchorPositionMs,
|
||||
anchorHostTimeMs,
|
||||
rate,
|
||||
controlMode,
|
||||
Object.hashAll(waitingOn),
|
||||
actorPeerId,
|
||||
actionHint,
|
||||
);
|
||||
|
||||
@override
|
||||
String toString() =>
|
||||
'PlaybackState(seq: $seq, media: $mediaKey, phase: ${phase.name}, '
|
||||
'anchor: ${anchorPositionMs}ms@$anchorHostTimeMs, rate: $rate, waitingOn: $waitingOn)';
|
||||
}
|
||||
|
||||
/// A peer's report of its own player to the host.
|
||||
class PeerStatus {
|
||||
/// The media this peer currently has loaded or is loading.
|
||||
final String mediaKey;
|
||||
|
||||
/// File loaded and first frame rendered (plus startup gates cleared).
|
||||
final bool ready;
|
||||
final bool buffering;
|
||||
final int positionMs;
|
||||
|
||||
/// The peer's measured min RTT to the host (sizes scheduled-start delays).
|
||||
final int? rttMs;
|
||||
|
||||
const PeerStatus({
|
||||
required this.mediaKey,
|
||||
required this.ready,
|
||||
required this.buffering,
|
||||
required this.positionMs,
|
||||
this.rttMs,
|
||||
});
|
||||
|
||||
Map<String, dynamic> toMap() => {
|
||||
'mk': mediaKey,
|
||||
'rdy': ready,
|
||||
'buf': buffering,
|
||||
'pos': positionMs,
|
||||
if (rttMs != null) 'rtt': rttMs,
|
||||
};
|
||||
|
||||
factory PeerStatus.fromMap(Map<String, dynamic> map) => PeerStatus(
|
||||
mediaKey: map['mk'] as String,
|
||||
ready: map['rdy'] as bool,
|
||||
buffering: map['buf'] as bool,
|
||||
positionMs: map['pos'] as int,
|
||||
rttMs: map['rtt'] as int?,
|
||||
);
|
||||
|
||||
@override
|
||||
bool operator ==(Object other) =>
|
||||
other is PeerStatus &&
|
||||
other.mediaKey == mediaKey &&
|
||||
other.ready == ready &&
|
||||
other.buffering == buffering &&
|
||||
other.positionMs == positionMs &&
|
||||
other.rttMs == rttMs;
|
||||
|
||||
@override
|
||||
int get hashCode => Object.hash(mediaKey, ready, buffering, positionMs, rttMs);
|
||||
|
||||
@override
|
||||
String toString() => 'PeerStatus($mediaKey, ready: $ready, buffering: $buffering, pos: ${positionMs}ms)';
|
||||
}
|
||||
|
||||
/// Serialized as the enum index — append new values only.
|
||||
enum ControlRequestKind { play, pause, seek, rate }
|
||||
|
||||
/// A guest's request for the host to apply a playback action (anyone mode).
|
||||
class ControlRequest {
|
||||
final ControlRequestKind kind;
|
||||
final int? positionMs;
|
||||
final double? rate;
|
||||
|
||||
const ControlRequest({required this.kind, this.positionMs, this.rate});
|
||||
|
||||
Map<String, dynamic> toMap() => {
|
||||
'k': kind.index,
|
||||
if (positionMs != null) 'pos': positionMs,
|
||||
if (rate != null) 'r': rate,
|
||||
};
|
||||
|
||||
factory ControlRequest.fromMap(Map<String, dynamic> map) => ControlRequest(
|
||||
kind: _enumFromIndex(ControlRequestKind.values, map['k'] as int) ?? ControlRequestKind.pause,
|
||||
positionMs: map['pos'] as int?,
|
||||
rate: (map['r'] as num?)?.toDouble(),
|
||||
);
|
||||
|
||||
@override
|
||||
bool operator ==(Object other) =>
|
||||
other is ControlRequest && other.kind == kind && other.positionMs == positionMs && other.rate == rate;
|
||||
|
||||
@override
|
||||
int get hashCode => Object.hash(kind, positionMs, rate);
|
||||
|
||||
@override
|
||||
String toString() => 'ControlRequest(${kind.name}, pos: $positionMs, rate: $rate)';
|
||||
}
|
||||
|
||||
/// Index-safe enum decode: out-of-range values (from a newer protocol
|
||||
/// version) return null instead of throwing.
|
||||
T? _enumFromIndex<T extends Enum>(List<T> values, int index) =>
|
||||
index >= 0 && index < values.length ? values[index] : null;
|
||||
@@ -1,26 +1,20 @@
|
||||
import 'dart:convert';
|
||||
|
||||
import 'watch_session.dart';
|
||||
import 'playback_state.dart';
|
||||
|
||||
/// Types of sync messages sent over the WebRTC data channel
|
||||
/// Types of sync messages sent over the relay data channel (protocol v2).
|
||||
enum SyncMessageType {
|
||||
/// Start playback
|
||||
play,
|
||||
/// Authoritative playback state broadcast by the host
|
||||
state,
|
||||
|
||||
/// Pause playback
|
||||
pause,
|
||||
/// A peer's player status report to the host
|
||||
status,
|
||||
|
||||
/// Seek to position
|
||||
seek,
|
||||
/// A guest's playback control request to the host
|
||||
control,
|
||||
|
||||
/// Buffering state changed
|
||||
buffering,
|
||||
|
||||
/// Periodic position update (for drift correction)
|
||||
positionSync,
|
||||
|
||||
/// Playback rate changed
|
||||
rate,
|
||||
/// Request the current playback state from the host
|
||||
requestState,
|
||||
|
||||
/// Participant joined the session
|
||||
join,
|
||||
@@ -28,45 +22,30 @@ enum SyncMessageType {
|
||||
/// Participant left the session
|
||||
leave,
|
||||
|
||||
/// Session configuration (sent by host on join)
|
||||
sessionConfig,
|
||||
|
||||
/// Ping for latency measurement
|
||||
/// Ping for clock-offset measurement
|
||||
ping,
|
||||
|
||||
/// Pong response
|
||||
pong,
|
||||
|
||||
/// Media switch (host changed content)
|
||||
mediaSwitch,
|
||||
|
||||
/// Host exited the video player
|
||||
hostExitedPlayer,
|
||||
|
||||
/// Player is ready (attached and loaded)
|
||||
playerReady,
|
||||
|
||||
/// Request session config from host (guest recovery)
|
||||
requestSessionConfig,
|
||||
}
|
||||
|
||||
/// A message sent over the WebRTC data channel for synchronization
|
||||
/// A message sent over the relay data channel for synchronization
|
||||
class SyncMessage {
|
||||
/// Current sync protocol version, carried on join messages. Peers with a
|
||||
/// different version are excluded from readiness gating and surfaced as
|
||||
/// needing an update.
|
||||
static const int protocolVersion = 2;
|
||||
|
||||
/// Type of this message
|
||||
final SyncMessageType type;
|
||||
|
||||
/// Timestamp when this message was created (Unix ms)
|
||||
/// Timestamp when this message was created (Unix ms). For pong messages
|
||||
/// this is the responder's "clock now" used for offset estimation.
|
||||
final int timestamp;
|
||||
|
||||
/// Position in milliseconds (for seek, positionSync)
|
||||
final int? positionMs;
|
||||
|
||||
/// Buffering state (for buffering message)
|
||||
final bool? bufferingState;
|
||||
|
||||
/// Playback rate (for rate message)
|
||||
final double? rate;
|
||||
|
||||
/// Peer ID of the sender
|
||||
final String? peerId;
|
||||
|
||||
@@ -76,98 +55,74 @@ class SyncMessage {
|
||||
/// Whether the sender is the host (for join message)
|
||||
final bool? isHost;
|
||||
|
||||
/// Control mode (for sessionConfig message)
|
||||
final ControlMode? controlMode;
|
||||
|
||||
/// Ping ID for matching pong responses
|
||||
final int? pingId;
|
||||
|
||||
/// Rating key of the media (for mediaSwitch message)
|
||||
final String? ratingKey;
|
||||
/// Authoritative playback state (for state message)
|
||||
final PlaybackState? state;
|
||||
|
||||
/// Server ID of the media (for mediaSwitch message)
|
||||
final String? serverId;
|
||||
/// Peer player status report (for status message)
|
||||
final PeerStatus? status;
|
||||
|
||||
/// Title of the media (for mediaSwitch message)
|
||||
final String? mediaTitle;
|
||||
/// Playback control request (for control message)
|
||||
final ControlRequest? control;
|
||||
|
||||
/// Whether playback is currently playing (for positionSync heartbeat)
|
||||
final bool? isPlaying;
|
||||
/// Sync protocol version (for join message)
|
||||
final int? version;
|
||||
|
||||
const SyncMessage({
|
||||
required this.type,
|
||||
required this.timestamp,
|
||||
this.positionMs,
|
||||
this.bufferingState,
|
||||
this.rate,
|
||||
this.peerId,
|
||||
this.displayName,
|
||||
this.isHost,
|
||||
this.controlMode,
|
||||
this.pingId,
|
||||
this.ratingKey,
|
||||
this.serverId,
|
||||
this.mediaTitle,
|
||||
this.isPlaying,
|
||||
this.state,
|
||||
this.status,
|
||||
this.control,
|
||||
this.version,
|
||||
});
|
||||
|
||||
/// Create a PLAY message
|
||||
factory SyncMessage.play({String? peerId, Duration? position}) {
|
||||
/// Create a STATE message carrying the host's authoritative playback state
|
||||
factory SyncMessage.state(PlaybackState state, {String? peerId}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.play,
|
||||
type: SyncMessageType.state,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
peerId: peerId,
|
||||
positionMs: position?.inMilliseconds,
|
||||
state: state,
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a PAUSE message
|
||||
factory SyncMessage.pause({String? peerId}) {
|
||||
return SyncMessage(type: SyncMessageType.pause, timestamp: DateTime.now().millisecondsSinceEpoch, peerId: peerId);
|
||||
/// Create a STATUS message reporting this peer's player state to the host
|
||||
factory SyncMessage.status(PeerStatus status, {String? peerId}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.status,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
peerId: peerId,
|
||||
status: status,
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a SEEK message
|
||||
factory SyncMessage.seek(Duration position, {String? peerId}) {
|
||||
/// Create a CONTROL message requesting a playback action from the host
|
||||
factory SyncMessage.control(ControlRequest control, {String? peerId}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.seek,
|
||||
type: SyncMessageType.control,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
peerId: peerId,
|
||||
control: control,
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a REQUEST_STATE message asking the host to re-send its state
|
||||
factory SyncMessage.requestState({String? peerId}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.requestState,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
positionMs: position.inMilliseconds,
|
||||
peerId: peerId,
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a BUFFERING message
|
||||
factory SyncMessage.buffering(bool isBuffering, {String? peerId}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.buffering,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
bufferingState: isBuffering,
|
||||
peerId: peerId,
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a POSITION_SYNC message (heartbeat with optional play/pause state)
|
||||
factory SyncMessage.positionSync(Duration position, {String? peerId, bool? isPlaying}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.positionSync,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
positionMs: position.inMilliseconds,
|
||||
peerId: peerId,
|
||||
isPlaying: isPlaying,
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a RATE message
|
||||
factory SyncMessage.rate(double playbackRate, {String? peerId}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.rate,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
rate: playbackRate,
|
||||
peerId: peerId,
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a JOIN message
|
||||
/// Create a JOIN message (carries the sender's protocol version)
|
||||
factory SyncMessage.join({required String peerId, required String displayName, required bool isHost}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.join,
|
||||
@@ -175,6 +130,7 @@ class SyncMessage {
|
||||
peerId: peerId,
|
||||
displayName: displayName,
|
||||
isHost: isHost,
|
||||
version: protocolVersion,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -183,44 +139,6 @@ class SyncMessage {
|
||||
return SyncMessage(type: SyncMessageType.leave, timestamp: DateTime.now().millisecondsSinceEpoch, peerId: peerId);
|
||||
}
|
||||
|
||||
/// Create a SESSION_CONFIG message (sent by host to new guests)
|
||||
///
|
||||
/// Optionally includes current media info so guests can catch up
|
||||
/// if they missed a mediaSwitch broadcast.
|
||||
factory SyncMessage.sessionConfig({
|
||||
required ControlMode controlMode,
|
||||
required Duration currentPosition,
|
||||
required bool isPlaying,
|
||||
required double playbackRate,
|
||||
String? peerId,
|
||||
String? ratingKey,
|
||||
String? serverId,
|
||||
String? mediaTitle,
|
||||
}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.sessionConfig,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
controlMode: controlMode,
|
||||
positionMs: currentPosition.inMilliseconds,
|
||||
isPlaying: isPlaying,
|
||||
bufferingState: !isPlaying, // Legacy compat: false = playing, true = paused
|
||||
rate: playbackRate,
|
||||
peerId: peerId,
|
||||
ratingKey: ratingKey,
|
||||
serverId: serverId,
|
||||
mediaTitle: mediaTitle,
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a REQUEST_SESSION_CONFIG message (sent by guest to request current config from host)
|
||||
factory SyncMessage.requestSessionConfig({String? peerId}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.requestSessionConfig,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
peerId: peerId,
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a PING message
|
||||
factory SyncMessage.ping(int pingId, {String? peerId}) {
|
||||
return SyncMessage(
|
||||
@@ -241,23 +159,6 @@ class SyncMessage {
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a MEDIA_SWITCH message (sent by host when changing content)
|
||||
factory SyncMessage.mediaSwitch({
|
||||
required String ratingKey,
|
||||
required String serverId,
|
||||
required String mediaTitle,
|
||||
String? peerId,
|
||||
}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.mediaSwitch,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
ratingKey: ratingKey,
|
||||
serverId: serverId,
|
||||
mediaTitle: mediaTitle,
|
||||
peerId: peerId,
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a HOST_EXITED_PLAYER message (sent by host when exiting video player)
|
||||
factory SyncMessage.hostExitedPlayer({String? peerId}) {
|
||||
return SyncMessage(
|
||||
@@ -267,59 +168,38 @@ class SyncMessage {
|
||||
);
|
||||
}
|
||||
|
||||
/// Create a PLAYER_READY message (sent when player is attached and ready)
|
||||
factory SyncMessage.playerReady({required String peerId, required bool ready}) {
|
||||
return SyncMessage(
|
||||
type: SyncMessageType.playerReady,
|
||||
timestamp: DateTime.now().millisecondsSinceEpoch,
|
||||
peerId: peerId,
|
||||
bufferingState: ready, // Reuse bufferingState field for ready status
|
||||
);
|
||||
}
|
||||
|
||||
/// Position as Duration (convenience getter)
|
||||
Duration? get position => positionMs != null ? Duration(milliseconds: positionMs!) : null;
|
||||
|
||||
SyncMessage copyWith({String? peerId}) {
|
||||
return SyncMessage(
|
||||
type: type,
|
||||
timestamp: timestamp,
|
||||
positionMs: positionMs,
|
||||
bufferingState: bufferingState,
|
||||
rate: rate,
|
||||
peerId: peerId ?? this.peerId,
|
||||
displayName: displayName,
|
||||
isHost: isHost,
|
||||
controlMode: controlMode,
|
||||
pingId: pingId,
|
||||
ratingKey: ratingKey,
|
||||
serverId: serverId,
|
||||
mediaTitle: mediaTitle,
|
||||
isPlaying: isPlaying,
|
||||
state: state,
|
||||
status: status,
|
||||
control: control,
|
||||
version: version,
|
||||
);
|
||||
}
|
||||
|
||||
/// Serialize to JSON string for sending over data channel
|
||||
/// Serialize to JSON string for sending over the data channel
|
||||
String toJson() {
|
||||
final map = <String, dynamic>{'t': type.name, 'ts': timestamp};
|
||||
|
||||
if (positionMs != null) map['pos'] = positionMs;
|
||||
if (bufferingState != null) map['buf'] = bufferingState;
|
||||
if (rate != null) map['r'] = rate;
|
||||
if (peerId != null) map['pid'] = peerId;
|
||||
if (displayName != null) map['name'] = displayName;
|
||||
if (isHost != null) map['host'] = isHost;
|
||||
if (controlMode != null) map['ctrl'] = controlMode!.index;
|
||||
if (pingId != null) map['ping'] = pingId;
|
||||
if (ratingKey != null) map['rk'] = ratingKey;
|
||||
if (serverId != null) map['sid'] = serverId;
|
||||
if (mediaTitle != null) map['title'] = mediaTitle;
|
||||
if (isPlaying != null) map['pl'] = isPlaying;
|
||||
if (state != null) map['st'] = state!.toMap();
|
||||
if (status != null) map['su'] = status!.toMap();
|
||||
if (control != null) map['co'] = control!.toMap();
|
||||
if (version != null) map['v'] = version;
|
||||
|
||||
return jsonEncode(map);
|
||||
}
|
||||
|
||||
/// Parse from JSON string received from data channel
|
||||
/// Parse from JSON string received from the data channel
|
||||
factory SyncMessage.fromJson(String jsonString) {
|
||||
final map = jsonDecode(jsonString) as Map<String, dynamic>;
|
||||
|
||||
@@ -330,24 +210,20 @@ class SyncMessage {
|
||||
return SyncMessage(
|
||||
type: type,
|
||||
timestamp: map['ts'] as int,
|
||||
positionMs: map['pos'] as int?,
|
||||
bufferingState: map['buf'] as bool?,
|
||||
rate: (map['r'] as num?)?.toDouble(),
|
||||
peerId: map['pid'] as String?,
|
||||
displayName: map['name'] as String?,
|
||||
isHost: map['host'] as bool?,
|
||||
controlMode: map['ctrl'] != null ? ControlMode.values[map['ctrl'] as int] : null,
|
||||
pingId: map['ping'] as int?,
|
||||
ratingKey: map['rk'] as String?,
|
||||
serverId: map['sid'] as String?,
|
||||
mediaTitle: map['title'] as String?,
|
||||
isPlaying: map['pl'] as bool?,
|
||||
state: map['st'] != null ? PlaybackState.fromMap((map['st'] as Map).cast<String, dynamic>()) : null,
|
||||
status: map['su'] != null ? PeerStatus.fromMap((map['su'] as Map).cast<String, dynamic>()) : null,
|
||||
control: map['co'] != null ? ControlRequest.fromMap((map['co'] as Map).cast<String, dynamic>()) : null,
|
||||
version: map['v'] as int?,
|
||||
);
|
||||
}
|
||||
|
||||
@override
|
||||
String toString() {
|
||||
return 'SyncMessage(type: $type, timestamp: $timestamp, positionMs: $positionMs, '
|
||||
'bufferingState: $bufferingState, rate: $rate, peerId: $peerId)';
|
||||
return 'SyncMessage(type: $type, timestamp: $timestamp, peerId: $peerId, '
|
||||
'state: $state, status: $status, control: $control)';
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,10 +7,11 @@ import 'package:flutter/foundation.dart';
|
||||
import '../../mpv/mpv.dart';
|
||||
import '../../services/settings_service.dart';
|
||||
import '../../utils/app_logger.dart';
|
||||
import '../models/playback_state.dart';
|
||||
import '../models/sync_message.dart';
|
||||
import '../models/watch_session.dart';
|
||||
import '../services/watch_together_controller.dart';
|
||||
import '../services/watch_together_peer_service.dart';
|
||||
import '../services/watch_together_sync_manager.dart';
|
||||
|
||||
/// Callback type for when media switches (for guest navigation)
|
||||
typedef MediaSwitchCallback = void Function(String ratingKey, ServerId serverId, String mediaTitle);
|
||||
@@ -26,10 +27,12 @@ typedef MediaSwitchCallback = void Function(String ratingKey, ServerId serverId,
|
||||
class WatchTogetherProvider with ChangeNotifier {
|
||||
WatchSession? _session;
|
||||
WatchTogetherPeerService? _peerService;
|
||||
WatchTogetherSyncManager? _syncManager;
|
||||
WatchTogetherController? _controller;
|
||||
final List<Participant> _participants = [];
|
||||
bool _isSyncing = false;
|
||||
bool _isDeferredPlay = false;
|
||||
bool _isWaitingForPeers = false;
|
||||
List<String> _waitingOnPeerIds = const [];
|
||||
PlaybackPhase? _playbackPhase;
|
||||
String _displayName = 'User';
|
||||
String? _lastHandledCurrentPlaybackKey;
|
||||
|
||||
@@ -87,15 +90,38 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
bool get isHost => _session?.isHost ?? false;
|
||||
bool get isConnected => _session?.isConnected ?? false;
|
||||
bool get isSyncing => _isSyncing;
|
||||
bool get isDeferredPlay => _isDeferredPlay;
|
||||
WatchSession? get session => _session;
|
||||
List<Participant> get participants => List.unmodifiable(_participants);
|
||||
int get participantCount => _participants.length;
|
||||
ControlMode get controlMode => _session?.controlMode ?? ControlMode.hostOnly;
|
||||
String? get sessionId => _session?.sessionId;
|
||||
WatchTogetherSyncManager? get syncManager => _syncManager;
|
||||
bool get isWaitingForHostReconnect => _isWaitingForHostReconnect;
|
||||
|
||||
/// Whether the room is held up waiting on peers (readiness or stalls) —
|
||||
/// drives the "Waiting for …" pill.
|
||||
bool get isWaitingForPeers => _isWaitingForPeers;
|
||||
|
||||
/// Display names of the peers the room is waiting on (excluding self).
|
||||
List<String> get waitingOnNames {
|
||||
final myPeerId = _peerService?.myPeerId;
|
||||
if (_waitingOnPeerIds.isEmpty) {
|
||||
// Guests waiting on a still-loading host have an empty digest.
|
||||
if (!isHost && _playbackPhase == PlaybackPhase.loading) {
|
||||
final hostName = _participants.where((p) => p.isHost).map((p) => p.displayName).firstOrNull;
|
||||
return [?hostName];
|
||||
}
|
||||
return const [];
|
||||
}
|
||||
return [
|
||||
for (final peerId in _waitingOnPeerIds)
|
||||
if (peerId != myPeerId)
|
||||
_participants.where((p) => p.peerId == peerId).map((p) => p.displayName).firstOrNull ?? '?',
|
||||
];
|
||||
}
|
||||
|
||||
/// Whether a player is currently attached to the sync controller.
|
||||
bool get hasAttachedPlayer => _controller?.hasPlayer ?? false;
|
||||
|
||||
// Participant join/leave event stream
|
||||
final StreamController<ParticipantEvent> _participantEventController = StreamController<ParticipantEvent>.broadcast();
|
||||
Stream<ParticipantEvent> get participantEvents => _participantEventController.stream;
|
||||
@@ -162,38 +188,96 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
}
|
||||
|
||||
void requestCurrentPlaybackSnapshot() {
|
||||
if (isHost || _peerService == null || _session == null || _peerService!.myPeerId == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
final request = SyncMessage.requestSessionConfig(peerId: _peerService!.myPeerId);
|
||||
if (_session!.hostPeerId != null) {
|
||||
appLogger.d('WatchTogether: Requesting current playback snapshot from host');
|
||||
_peerService!.sendTo(_session!.hostPeerId!, request);
|
||||
} else {
|
||||
appLogger.d('WatchTogether: Host peer unknown, broadcasting current playback snapshot request');
|
||||
_peerService!.broadcast(request);
|
||||
}
|
||||
if (isHost) return;
|
||||
appLogger.d('WatchTogether: Requesting current playback state from host');
|
||||
_controller?.requestState();
|
||||
}
|
||||
|
||||
/// Wire up reconnection handler to re-announce join and readiness after reconnect
|
||||
/// Wire up reconnection handler to re-announce join and re-sync state
|
||||
void _wireReconnectHandler() {
|
||||
_peerService!.onReconnected = () {
|
||||
_syncManager?.announceJoin(_displayName);
|
||||
_syncManager?.reannounceReadyIfNeeded();
|
||||
_controller?.announceJoin(_displayName);
|
||||
_controller?.onReconnected();
|
||||
};
|
||||
}
|
||||
|
||||
/// Wire up sync manager's state change callback to update provider state
|
||||
void _wireSyncStateChanges() {
|
||||
_syncManager!.onSyncStateChanged = (isSyncing) {
|
||||
_isSyncing = isSyncing;
|
||||
/// Wire the controller's callbacks into provider/UI state
|
||||
void _wireController() {
|
||||
final controller = _controller!;
|
||||
|
||||
controller.onCorrectingChanged = (correcting) {
|
||||
_isSyncing = correcting;
|
||||
notifyListeners();
|
||||
};
|
||||
_syncManager!.onDeferredPlayChanged = (isDeferredPlay) {
|
||||
_isDeferredPlay = isDeferredPlay;
|
||||
|
||||
controller.onPhaseChanged = (phase) {
|
||||
_playbackPhase = phase;
|
||||
_updateWaitingState();
|
||||
};
|
||||
|
||||
controller.onWaitingOnChanged = (peerIds) {
|
||||
_waitingOnPeerIds = peerIds;
|
||||
for (var i = 0; i < _participants.length; i++) {
|
||||
final isWaitedOn = peerIds.contains(_participants[i].peerId);
|
||||
if (_participants[i].isBuffering != isWaitedOn) {
|
||||
_participants[i] = _participants[i].copyWith(isBuffering: isWaitedOn);
|
||||
if (isWaitedOn) {
|
||||
_emitActionEvent(_participants[i].peerId, ParticipantEventType.buffering);
|
||||
}
|
||||
}
|
||||
}
|
||||
_updateWaitingState();
|
||||
};
|
||||
|
||||
controller.onControlModeReceived = (mode) {
|
||||
if (isHost || _session == null) return;
|
||||
if (_session!.controlMode == mode) return;
|
||||
_session = _session!.copyWith(controlMode: mode);
|
||||
controller.updateSession(_session!);
|
||||
notifyListeners();
|
||||
};
|
||||
|
||||
controller.onMediaStateReceived = _handleMediaStateReceived;
|
||||
|
||||
controller.onRemoteAction = (peerId, hint) {
|
||||
final type = switch (hint) {
|
||||
PlaybackActionHint.play => ParticipantEventType.resumed,
|
||||
PlaybackActionHint.pause => ParticipantEventType.paused,
|
||||
PlaybackActionHint.seek => ParticipantEventType.seeked,
|
||||
PlaybackActionHint.rate || PlaybackActionHint.mediaSwitch => null,
|
||||
};
|
||||
if (type != null) _emitActionEvent(peerId, type);
|
||||
};
|
||||
|
||||
controller.onPeerNeedsUpdate = (peerId) {
|
||||
final name = _participants.where((p) => p.peerId == peerId).map((p) => p.displayName).firstOrNull;
|
||||
_participantEventController.add(
|
||||
ParticipantEvent(displayName: name ?? peerId, type: ParticipantEventType.needsUpdate),
|
||||
);
|
||||
};
|
||||
|
||||
controller.onResumedWithout = (peerIds) {
|
||||
for (final peerId in peerIds) {
|
||||
final name = _participants.where((p) => p.peerId == peerId).map((p) => p.displayName).firstOrNull;
|
||||
if (name != null) {
|
||||
_participantEventController.add(
|
||||
ParticipantEvent(displayName: name, type: ParticipantEventType.resumedWithout),
|
||||
);
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
void _updateWaitingState() {
|
||||
final phase = _playbackPhase;
|
||||
final waiting =
|
||||
phase == PlaybackPhase.waitingForPeers ||
|
||||
// Guests waiting on a still-loading host (no digest in that phase).
|
||||
(!isHost && phase == PlaybackPhase.loading && hasCurrentPlayback);
|
||||
if (waiting != _isWaitingForPeers) {
|
||||
_isWaitingForPeers = waiting;
|
||||
}
|
||||
notifyListeners();
|
||||
}
|
||||
|
||||
/// Create a new watch together session as host
|
||||
@@ -230,13 +314,9 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
_displayName = displayName ?? _generateDisplayName();
|
||||
_participants.add(Participant(peerId: _peerService!.myPeerId!, displayName: _displayName, isHost: true));
|
||||
|
||||
_syncManager = WatchTogetherSyncManager(
|
||||
peerService: _peerService!,
|
||||
session: _session!,
|
||||
displayName: _displayName,
|
||||
);
|
||||
_controller = WatchTogetherController(peerService: _peerService!, session: _session!);
|
||||
|
||||
_wireSyncStateChanges();
|
||||
_wireController();
|
||||
_wireReconnectHandler();
|
||||
|
||||
notifyListeners();
|
||||
@@ -274,26 +354,16 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
|
||||
_displayName = displayName ?? _generateDisplayName();
|
||||
|
||||
_syncManager = WatchTogetherSyncManager(
|
||||
peerService: _peerService!,
|
||||
session: _session!,
|
||||
displayName: _displayName,
|
||||
);
|
||||
_controller = WatchTogetherController(peerService: _peerService!, session: _session!);
|
||||
|
||||
_syncManager!.onSessionConfigReceived = (controlMode) {
|
||||
_session = _session!.copyWith(controlMode: controlMode);
|
||||
_syncManager!.updateSession(_session!);
|
||||
notifyListeners();
|
||||
};
|
||||
|
||||
_wireSyncStateChanges();
|
||||
_wireController();
|
||||
_wireReconnectHandler();
|
||||
|
||||
// Add self to participants
|
||||
_participants.add(Participant(peerId: _peerService!.myPeerId!, displayName: _displayName, isHost: false));
|
||||
|
||||
// Announce join to other participants
|
||||
_syncManager!.announceJoin(_displayName);
|
||||
_controller!.announceJoin(_displayName);
|
||||
requestCurrentPlaybackSnapshot();
|
||||
|
||||
notifyListeners();
|
||||
@@ -346,7 +416,7 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
appLogger.d('WatchTogether: Leaving session');
|
||||
|
||||
// Announce leave if connected
|
||||
_syncManager?.announceLeave();
|
||||
_controller?.announceLeave();
|
||||
|
||||
// Clean up subscriptions
|
||||
unawaited(_peerConnectedSubscription?.cancel());
|
||||
@@ -363,8 +433,8 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
_cancelHostReconnectGracePeriod();
|
||||
|
||||
// Clean up services
|
||||
_syncManager?.dispose();
|
||||
_syncManager = null;
|
||||
_controller?.dispose();
|
||||
_controller = null;
|
||||
|
||||
await _peerService?.disconnect();
|
||||
_peerService?.dispose();
|
||||
@@ -373,7 +443,9 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
_session = null;
|
||||
_participants.clear();
|
||||
_isSyncing = false;
|
||||
_isDeferredPlay = false;
|
||||
_isWaitingForPeers = false;
|
||||
_waitingOnPeerIds = const [];
|
||||
_playbackPhase = null;
|
||||
_lastHandledCurrentPlaybackKey = null;
|
||||
_lastActionEventMs.clear();
|
||||
_hostIntentionallyLeft = false;
|
||||
@@ -382,30 +454,47 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
appLogger.d('WatchTogether: Session left');
|
||||
}
|
||||
|
||||
/// Attach a player to the sync manager
|
||||
void attachPlayer(Player player) {
|
||||
if (_syncManager == null) {
|
||||
appLogger.w('WatchTogether: Cannot attach player - no sync manager');
|
||||
/// Attach a player to the sync controller for the given media.
|
||||
///
|
||||
/// [hasFirstFrame] is the screen's first-frame snapshot, [startupHold]
|
||||
/// delays sync readiness past platform startup gates (frame-rate switch),
|
||||
/// and [remoteSeek] routes sync-issued seeks through the screen's seek
|
||||
/// path (Plex transcode restarts).
|
||||
void attachPlayer(
|
||||
Player player, {
|
||||
required String ratingKey,
|
||||
required String serverId,
|
||||
String? mediaTitle,
|
||||
bool hasFirstFrame = false,
|
||||
Future<void>? startupHold,
|
||||
Future<void> Function(Duration target)? remoteSeek,
|
||||
}) {
|
||||
if (_controller == null) {
|
||||
appLogger.w('WatchTogether: Cannot attach player - no sync controller');
|
||||
return;
|
||||
}
|
||||
|
||||
// Initialize sync manager with existing participants (may have joined before player attached)
|
||||
final peerIds = _participants.map((p) => p.peerId).toList();
|
||||
_syncManager!.initializeParticipants(peerIds);
|
||||
|
||||
_syncManager!.attachPlayer(player);
|
||||
appLogger.d('WatchTogether: Player attached to sync manager');
|
||||
_controller!.attachPlayer(
|
||||
player,
|
||||
ratingKey: ratingKey,
|
||||
serverId: serverId,
|
||||
mediaTitle: mediaTitle,
|
||||
hasFirstFrame: hasFirstFrame,
|
||||
startupHold: startupHold,
|
||||
remoteSeek: remoteSeek,
|
||||
);
|
||||
}
|
||||
|
||||
/// Detach the player from the sync manager
|
||||
void detachPlayer() {
|
||||
_syncManager?.detachPlayer();
|
||||
appLogger.d('WatchTogether: Player detached from sync manager');
|
||||
/// Detach the player from the sync controller. [exiting] means the user
|
||||
/// left the video player (ends the media epoch); episode switches detach
|
||||
/// without exiting.
|
||||
void detachPlayer({bool exiting = false}) {
|
||||
_controller?.detachPlayer(exiting: exiting);
|
||||
}
|
||||
|
||||
/// Suppress position sync while the app is backgrounded.
|
||||
/// Suppress sync heartbeats/corrections while the app is backgrounded.
|
||||
void setBackgrounded(bool value) {
|
||||
_syncManager?.setBackgrounded(value);
|
||||
_controller?.setBackgrounded(value);
|
||||
}
|
||||
|
||||
/// Set up listeners for peer service events
|
||||
@@ -432,8 +521,8 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
// Capture display name before removal for notification
|
||||
final disconnectedName = _participants.where((p) => p.peerId == peerId).map((p) => p.displayName).firstOrNull;
|
||||
|
||||
// The sync controller observes peer disconnects itself.
|
||||
_participants.removeWhere((p) => p.peerId == peerId);
|
||||
unawaited(_syncManager?.handlePeerDisconnected(peerId));
|
||||
|
||||
// If host disconnected unexpectedly, start grace period for reconnection.
|
||||
// Skip if the host already sent a deliberate leave message.
|
||||
@@ -525,61 +614,13 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
}
|
||||
break;
|
||||
|
||||
case SyncMessageType.buffering:
|
||||
if (message.peerId != null) {
|
||||
final index = _participants.indexWhere((p) => p.peerId == message.peerId);
|
||||
if (index >= 0) {
|
||||
final newState = message.bufferingState ?? false;
|
||||
if (_participants[index].isBuffering != newState) {
|
||||
_participants[index] = _participants[index].copyWith(isBuffering: newState);
|
||||
if (newState) {
|
||||
_emitActionEvent(message.peerId, ParticipantEventType.buffering);
|
||||
}
|
||||
notifyListeners();
|
||||
}
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
case SyncMessageType.positionSync:
|
||||
if (message.peerId != null && message.position != null) {
|
||||
final index = _participants.indexWhere((p) => p.peerId == message.peerId);
|
||||
if (index >= 0) {
|
||||
_participants[index] = _participants[index].copyWith(lastKnownPosition: message.position!);
|
||||
// Don't notify for position updates - too frequent
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
case SyncMessageType.mediaSwitch:
|
||||
_handleMediaSwitch(message);
|
||||
break;
|
||||
|
||||
case SyncMessageType.hostExitedPlayer:
|
||||
_handleHostExitedPlayer(message);
|
||||
break;
|
||||
|
||||
case SyncMessageType.sessionConfig:
|
||||
_handleSessionConfig(message);
|
||||
break;
|
||||
|
||||
case SyncMessageType.requestSessionConfig:
|
||||
// Handled at sync manager level (host responds with config)
|
||||
break;
|
||||
|
||||
case SyncMessageType.play:
|
||||
_emitActionEvent(message.peerId, ParticipantEventType.resumed);
|
||||
break;
|
||||
|
||||
case SyncMessageType.pause:
|
||||
_emitActionEvent(message.peerId, ParticipantEventType.paused);
|
||||
break;
|
||||
|
||||
case SyncMessageType.seek:
|
||||
_emitActionEvent(message.peerId, ParticipantEventType.seeked);
|
||||
break;
|
||||
|
||||
default:
|
||||
// Playback sync messages (state/status/control/...) are handled by
|
||||
// the session controller.
|
||||
break;
|
||||
}
|
||||
}
|
||||
@@ -600,43 +641,31 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle session config from host (guest only)
|
||||
/// This is handled at provider level so it's processed even before player is attached
|
||||
void _handleSessionConfig(SyncMessage message) {
|
||||
if (isHost) return; // Host doesn't need to process config
|
||||
/// Handle current-media info carried in the host's playback state
|
||||
/// (guest only). Processed even when no player is attached so guests can
|
||||
/// navigate into (or between) playback.
|
||||
void _handleMediaStateReceived(String ratingKey, String serverId, String? mediaTitle) {
|
||||
if (isHost) return;
|
||||
|
||||
if (message.controlMode != null) {
|
||||
appLogger.d('WatchTogether: Received session config, controlMode: ${message.controlMode}');
|
||||
_session = _session!.copyWith(controlMode: message.controlMode!);
|
||||
_syncManager?.updateSession(_session!); // Update sync manager if it exists
|
||||
notifyListeners();
|
||||
}
|
||||
final playbackKey = _buildPlaybackKey(ratingKey, serverIdOrNull(serverId));
|
||||
final shouldDispatch = playbackKey != _lastHandledCurrentPlaybackKey;
|
||||
|
||||
if (message.ratingKey != null && message.serverId != null && message.mediaTitle != null) {
|
||||
final playbackKey = _buildPlaybackKey(message.ratingKey, serverIdOrNull(message.serverId));
|
||||
final shouldDispatch = playbackKey != _lastHandledCurrentPlaybackKey;
|
||||
_updateCurrentPlaybackSnapshot(ratingKey: ratingKey, serverId: ServerId(serverId), mediaTitle: mediaTitle ?? '');
|
||||
notifyListeners();
|
||||
|
||||
_updateCurrentPlaybackSnapshot(
|
||||
ratingKey: message.ratingKey!,
|
||||
serverId: ServerId(message.serverId!),
|
||||
mediaTitle: message.mediaTitle!,
|
||||
if (shouldDispatch) {
|
||||
_dispatchCurrentPlayback(
|
||||
ratingKey: ratingKey,
|
||||
serverId: ServerId(serverId),
|
||||
mediaTitle: mediaTitle ?? '',
|
||||
source: 'playback state',
|
||||
);
|
||||
notifyListeners();
|
||||
|
||||
if (shouldDispatch) {
|
||||
_dispatchCurrentPlayback(
|
||||
ratingKey: message.ratingKey!,
|
||||
serverId: ServerId(message.serverId!),
|
||||
mediaTitle: message.mediaTitle!,
|
||||
source: 'session config',
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Called when user seeks locally (to broadcast to peers)
|
||||
/// Called when user seeks locally (to sync with peers)
|
||||
void onLocalSeek(Duration position) {
|
||||
_syncManager?.onLocalSeek(position);
|
||||
_controller?.onLocalSeek(position);
|
||||
}
|
||||
|
||||
/// Whether the current user can control playback
|
||||
@@ -661,52 +690,12 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
// Update session with new media info
|
||||
_session = _session!.copyWith(mediaRatingKey: ratingKey, mediaServerId: serverId, mediaTitle: mediaTitle);
|
||||
|
||||
// Broadcast media switch to all guests
|
||||
_peerService!.broadcast(
|
||||
SyncMessage.mediaSwitch(
|
||||
ratingKey: ratingKey,
|
||||
serverId: serverId,
|
||||
mediaTitle: mediaTitle,
|
||||
peerId: _peerService!.myPeerId,
|
||||
),
|
||||
);
|
||||
// The controller broadcasts the new media epoch in its playback state.
|
||||
_controller?.setCurrentMedia(ratingKey: ratingKey, serverId: serverId, mediaTitle: mediaTitle);
|
||||
|
||||
notifyListeners();
|
||||
}
|
||||
|
||||
/// Handle media switch message from host (guest only)
|
||||
void _handleMediaSwitch(SyncMessage message) {
|
||||
if (isHost) return; // Host doesn't need to handle their own switch
|
||||
|
||||
if (message.ratingKey == null || message.serverId == null || message.mediaTitle == null) {
|
||||
appLogger.w('WatchTogether: Received incomplete media switch message');
|
||||
return;
|
||||
}
|
||||
|
||||
final playbackKey = _buildPlaybackKey(message.ratingKey, serverIdOrNull(message.serverId));
|
||||
final shouldDispatch = playbackKey != _lastHandledCurrentPlaybackKey;
|
||||
|
||||
_updateCurrentPlaybackSnapshot(
|
||||
ratingKey: message.ratingKey!,
|
||||
serverId: ServerId(message.serverId!),
|
||||
mediaTitle: message.mediaTitle!,
|
||||
);
|
||||
notifyListeners();
|
||||
|
||||
if (!shouldDispatch) {
|
||||
appLogger.d('WatchTogether: Ignoring duplicate media switch for ${message.ratingKey}');
|
||||
return;
|
||||
}
|
||||
|
||||
appLogger.d('WatchTogether: Received media switch: ${message.mediaTitle}');
|
||||
_dispatchCurrentPlayback(
|
||||
ratingKey: message.ratingKey!,
|
||||
serverId: ServerId(message.serverId!),
|
||||
mediaTitle: message.mediaTitle!,
|
||||
source: 'media switch',
|
||||
);
|
||||
}
|
||||
|
||||
/// Notify guests that host is exiting the video player
|
||||
///
|
||||
/// Call this from video player dispose when host exits.
|
||||
@@ -782,7 +771,7 @@ class WatchTogetherProvider with ChangeNotifier {
|
||||
}
|
||||
|
||||
/// Type of participant event
|
||||
enum ParticipantEventType { joined, left, paused, resumed, seeked, buffering }
|
||||
enum ParticipantEventType { joined, left, paused, resumed, seeked, buffering, needsUpdate, resumedWithout }
|
||||
|
||||
/// Event emitted when a participant joins or leaves
|
||||
class ParticipantEvent {
|
||||
|
||||
@@ -0,0 +1,259 @@
|
||||
import 'dart:async';
|
||||
|
||||
import 'package:flutter/services.dart';
|
||||
|
||||
import '../../mpv/mpv.dart';
|
||||
import '../../utils/app_logger.dart';
|
||||
|
||||
enum _ExpectationKind { playing, rate }
|
||||
|
||||
class _Expectation {
|
||||
final _ExpectationKind kind;
|
||||
final bool? playingValue;
|
||||
final double? rateValue;
|
||||
final int deadlineMs;
|
||||
|
||||
_Expectation.playing(bool value, this.deadlineMs)
|
||||
: kind = _ExpectationKind.playing,
|
||||
playingValue = value,
|
||||
rateValue = null;
|
||||
|
||||
_Expectation.rate(double value, this.deadlineMs)
|
||||
: kind = _ExpectationKind.rate,
|
||||
playingValue = null,
|
||||
rateValue = value;
|
||||
}
|
||||
|
||||
/// One player attachment to a Watch Together session.
|
||||
///
|
||||
/// Wraps the screen's [Player] with:
|
||||
/// - **Guarded commands** that survive player teardown races: recoverable
|
||||
/// failures ([StateError], `COMMAND_FAILED`/`NOT_INITIALIZED`
|
||||
/// [PlatformException]s) report `false` and fire [AttachedPlayer.new]'s
|
||||
/// `onLost` once instead of throwing.
|
||||
/// - An **expected-state ledger** separating command acks from user intents
|
||||
/// on the playing/rate streams. Property events arrive *after* the command
|
||||
/// future resolves, so a boolean "remote action in progress" flag misses
|
||||
/// them; the ledger matches observed transitions against outstanding
|
||||
/// expectations instead.
|
||||
/// - Fresh snapshot reads for sync math ([position] uses
|
||||
/// [Player.currentPosition], not the throttled state).
|
||||
///
|
||||
/// The session controller creates one instance per attachment and disposes
|
||||
/// it on detach — instance lifecycle *is* the staleness guard.
|
||||
class AttachedPlayer {
|
||||
AttachedPlayer({required Player player, required this._onLost, this._remoteSeek, int Function()? nowMs})
|
||||
: _player = player,
|
||||
_nowMs = nowMs ?? _systemNowMs {
|
||||
_lastPlaying = player.state.playing;
|
||||
_lastBuffering = player.state.buffering;
|
||||
_lastRate = player.state.rate;
|
||||
|
||||
_subscriptions.add(player.streams.playing.listen(_onPlayingEvent));
|
||||
_subscriptions.add(player.streams.buffering.listen(_onBufferingEvent));
|
||||
_subscriptions.add(player.streams.rate.listen(_onRateEvent));
|
||||
_subscriptions.add(
|
||||
player.streams.playbackRestart.listen((_) {
|
||||
if (!_disposed) _loadedSignalsController.add(null);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
static int _systemNowMs() => DateTime.now().millisecondsSinceEpoch;
|
||||
|
||||
/// How long an issued command may wait for its property event before the
|
||||
/// expectation is considered dead (covers silently-swallowed commands).
|
||||
static const int _expectationTtlMs = 3000;
|
||||
|
||||
final Player _player;
|
||||
final void Function() _onLost;
|
||||
final Future<void> Function(Duration target)? _remoteSeek;
|
||||
final int Function() _nowMs;
|
||||
|
||||
final List<StreamSubscription<dynamic>> _subscriptions = [];
|
||||
final List<_Expectation> _expectations = [];
|
||||
|
||||
final _playingIntentsController = StreamController<bool>.broadcast();
|
||||
final _rateIntentsController = StreamController<double>.broadcast();
|
||||
final _bufferingChangesController = StreamController<bool>.broadcast();
|
||||
final _loadedSignalsController = StreamController<void>.broadcast();
|
||||
|
||||
late bool _lastPlaying;
|
||||
late bool _lastBuffering;
|
||||
late double _lastRate;
|
||||
bool _disposed = false;
|
||||
bool _lostFired = false;
|
||||
|
||||
/// User-initiated play/pause transitions (command acks are filtered out).
|
||||
Stream<bool> get playingIntents => _playingIntentsController.stream;
|
||||
|
||||
/// User-initiated rate changes (command acks are filtered out).
|
||||
Stream<double> get rateIntents => _rateIntentsController.stream;
|
||||
|
||||
/// Raw buffering transitions (`paused-for-cache`).
|
||||
Stream<bool> get bufferingChanges => _bufferingChangesController.stream;
|
||||
|
||||
/// `playback-restart` events: first frame rendered after load and after
|
||||
/// every seek.
|
||||
Stream<void> get loadedSignals => _loadedSignalsController.stream;
|
||||
|
||||
bool get usable => !_disposed && !_player.disposed;
|
||||
|
||||
// Fresh snapshots.
|
||||
Duration get position => _player.currentPosition;
|
||||
bool get playing => _player.state.playing;
|
||||
bool get buffering => _player.state.buffering;
|
||||
bool get completed => _player.state.completed;
|
||||
bool get seekable => _player.state.seekable;
|
||||
Duration get duration => _player.state.duration;
|
||||
double get rate => _player.state.rate;
|
||||
bool get passthroughActive => _player.audioPassthroughActive;
|
||||
|
||||
/// Demuxer cache ahead of the playhead, or null when the backend hasn't
|
||||
/// reported a cache position.
|
||||
Duration? get bufferAhead {
|
||||
final buffer = _player.state.buffer;
|
||||
if (buffer == Duration.zero) return null;
|
||||
final ahead = buffer - position;
|
||||
return ahead.isNegative ? Duration.zero : ahead;
|
||||
}
|
||||
|
||||
/// Start or resume playback. Records a ledger expectation so the resulting
|
||||
/// playing event is consumed as an ack.
|
||||
Future<bool> play() {
|
||||
final expectation = _expect(_Expectation.playing(true, _nowMs() + _expectationTtlMs));
|
||||
return _guarded('play', (player) => player.play(), expectation);
|
||||
}
|
||||
|
||||
Future<bool> pause() {
|
||||
final expectation = _expect(_Expectation.playing(false, _nowMs() + _expectationTtlMs));
|
||||
return _guarded('pause', (player) => player.pause(), expectation);
|
||||
}
|
||||
|
||||
Future<bool> setRate(double rate) {
|
||||
final expectation = _expect(_Expectation.rate(rate, _nowMs() + _expectationTtlMs));
|
||||
return _guarded('setRate', (player) => player.setRate(rate), expectation);
|
||||
}
|
||||
|
||||
/// Seek issued by the sync layer. Routed through the screen's seek
|
||||
/// delegate when provided (Plex transcode restarts need the full path),
|
||||
/// falling back to a plain player seek.
|
||||
Future<bool> seek(Duration target) {
|
||||
return _guarded('seek', (player) async {
|
||||
final delegate = _remoteSeek;
|
||||
if (delegate != null) {
|
||||
try {
|
||||
await delegate(target);
|
||||
return;
|
||||
} catch (e) {
|
||||
appLogger.w('AttachedPlayer: seek delegate failed, falling back to player.seek', error: e);
|
||||
}
|
||||
}
|
||||
await player.seek(target);
|
||||
});
|
||||
}
|
||||
|
||||
_Expectation _expect(_Expectation expectation) {
|
||||
_expectations.add(expectation);
|
||||
return expectation;
|
||||
}
|
||||
|
||||
Future<bool> _guarded(
|
||||
String actionName,
|
||||
Future<void> Function(Player player) command, [
|
||||
_Expectation? expectation,
|
||||
]) async {
|
||||
if (!usable) {
|
||||
_expectations.remove(expectation);
|
||||
_handleLost(actionName, StateError('Player became unavailable'));
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
await command(_player);
|
||||
} on StateError catch (e) {
|
||||
_expectations.remove(expectation);
|
||||
_handleLost(actionName, e);
|
||||
return false;
|
||||
} on PlatformException catch (e) {
|
||||
_expectations.remove(expectation);
|
||||
if (e.code == 'COMMAND_FAILED' || e.code == 'NOT_INITIALIZED') {
|
||||
_handleLost(actionName, e);
|
||||
return false;
|
||||
}
|
||||
rethrow;
|
||||
}
|
||||
|
||||
if (!usable) {
|
||||
_expectations.remove(expectation);
|
||||
if (!_disposed) _handleLost(actionName, StateError('Player became unavailable'));
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
void _handleLost(String actionName, Object error) {
|
||||
if (_disposed || _lostFired) return;
|
||||
_lostFired = true;
|
||||
appLogger.w('AttachedPlayer: $actionName failed because the player became unavailable', error: error);
|
||||
_onLost();
|
||||
}
|
||||
|
||||
void _pruneExpired() {
|
||||
final now = _nowMs();
|
||||
_expectations.removeWhere((e) => now > e.deadlineMs);
|
||||
}
|
||||
|
||||
bool _consumePlayingExpectation(bool value) {
|
||||
_pruneExpired();
|
||||
final index = _expectations.indexWhere((e) => e.kind == _ExpectationKind.playing && e.playingValue == value);
|
||||
if (index < 0) return false;
|
||||
_expectations.removeAt(index);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool _consumeRateExpectation(double value) {
|
||||
_pruneExpired();
|
||||
final index = _expectations.indexWhere(
|
||||
(e) => e.kind == _ExpectationKind.rate && (e.rateValue! - value).abs() < 0.001,
|
||||
);
|
||||
if (index < 0) return false;
|
||||
_expectations.removeAt(index);
|
||||
return true;
|
||||
}
|
||||
|
||||
void _onPlayingEvent(bool value) {
|
||||
if (_disposed || value == _lastPlaying) return;
|
||||
_lastPlaying = value;
|
||||
if (_consumePlayingExpectation(value)) return;
|
||||
_playingIntentsController.add(value);
|
||||
}
|
||||
|
||||
void _onRateEvent(double value) {
|
||||
if (_disposed || value == _lastRate) return;
|
||||
_lastRate = value;
|
||||
if (_consumeRateExpectation(value)) return;
|
||||
_rateIntentsController.add(value);
|
||||
}
|
||||
|
||||
void _onBufferingEvent(bool value) {
|
||||
if (_disposed || value == _lastBuffering) return;
|
||||
_lastBuffering = value;
|
||||
_bufferingChangesController.add(value);
|
||||
}
|
||||
|
||||
Future<void> dispose() async {
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
_expectations.clear();
|
||||
final subscriptions = List<StreamSubscription<dynamic>>.of(_subscriptions);
|
||||
_subscriptions.clear();
|
||||
for (final subscription in subscriptions) {
|
||||
unawaited(subscription.cancel());
|
||||
}
|
||||
await _playingIntentsController.close();
|
||||
await _rateIntentsController.close();
|
||||
await _bufferingChangesController.close();
|
||||
await _loadedSignalsController.close();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
import 'dart:async';
|
||||
|
||||
import '../../utils/app_logger.dart';
|
||||
|
||||
/// NTP-style clock-offset estimation against the session host (guest side).
|
||||
///
|
||||
/// Sends pings through [sendPing] (the controller wraps them into sync
|
||||
/// messages addressed to the host) and consumes pongs via [onPong]. Keeps a
|
||||
/// rolling window of samples and reports the offset of the lowest-RTT sample
|
||||
/// — a single clean exchange beats an average polluted by jittery ones.
|
||||
///
|
||||
/// All time reads go through the injected [nowMs] so tests can virtualize
|
||||
/// time alongside `fakeAsync`.
|
||||
class ClockSync {
|
||||
ClockSync({required this._sendPing, int Function()? nowMs}) : _nowMs = nowMs ?? _systemNowMs;
|
||||
|
||||
static int _systemNowMs() => DateTime.now().millisecondsSinceEpoch;
|
||||
|
||||
static const int _windowSize = 8;
|
||||
static const int _maxAcceptedRttMs = 1000;
|
||||
static const Duration _interval = Duration(seconds: 5);
|
||||
static const Duration _burstSpacing = Duration(milliseconds: 500);
|
||||
static const int _burstCount = 3;
|
||||
static const int _pendingExpiryMs = 10000;
|
||||
|
||||
final void Function(int pingId) _sendPing;
|
||||
final int Function() _nowMs;
|
||||
|
||||
/// In-flight pings: pingId -> local send time. Multiple may be pending.
|
||||
final Map<int, int> _pending = {};
|
||||
|
||||
/// Accepted samples, oldest first.
|
||||
final List<({int offsetMs, int rttMs})> _samples = [];
|
||||
|
||||
Timer? _timer;
|
||||
Timer? _burstTimer;
|
||||
bool _started = false;
|
||||
|
||||
/// How far ahead the host's clock is vs ours, or null before any sample.
|
||||
int? get offsetMs => _best?.offsetMs;
|
||||
|
||||
/// Lowest RTT to the host in the sample window, or null before any sample.
|
||||
int? get minRttMs => _best?.rttMs;
|
||||
|
||||
({int offsetMs, int rttMs})? get _best {
|
||||
if (_samples.isEmpty) return null;
|
||||
var best = _samples.first;
|
||||
for (final sample in _samples.skip(1)) {
|
||||
if (sample.rttMs < best.rttMs) best = sample;
|
||||
}
|
||||
return best;
|
||||
}
|
||||
|
||||
/// Local time translated into the host's clock (identity until a sample
|
||||
/// arrives — callers needing a guarantee should check [offsetMs]).
|
||||
int hostNowMs() => _nowMs() + (offsetMs ?? 0);
|
||||
|
||||
/// Begin measuring: a short convergence burst, then a steady interval.
|
||||
void start() {
|
||||
if (_started) return;
|
||||
_started = true;
|
||||
|
||||
var sent = 0;
|
||||
_ping();
|
||||
sent++;
|
||||
_burstTimer = Timer.periodic(_burstSpacing, (timer) {
|
||||
if (sent >= _burstCount) {
|
||||
timer.cancel();
|
||||
return;
|
||||
}
|
||||
_ping();
|
||||
sent++;
|
||||
});
|
||||
|
||||
_timer = Timer.periodic(_interval, (_) => _ping());
|
||||
}
|
||||
|
||||
void stop() {
|
||||
_started = false;
|
||||
_timer?.cancel();
|
||||
_timer = null;
|
||||
_burstTimer?.cancel();
|
||||
_burstTimer = null;
|
||||
_pending.clear();
|
||||
}
|
||||
|
||||
void _ping() {
|
||||
final now = _nowMs();
|
||||
_pending.removeWhere((_, sentAt) => now - sentAt > _pendingExpiryMs);
|
||||
// The ping id doubles as the send timestamp; nudge to keep ids unique
|
||||
// when two pings land on the same millisecond.
|
||||
var pingId = now;
|
||||
while (_pending.containsKey(pingId)) {
|
||||
pingId++;
|
||||
}
|
||||
_pending[pingId] = now;
|
||||
_sendPing(pingId);
|
||||
}
|
||||
|
||||
/// Feed a pong from the host. [remoteTimestampMs] is the host's clock when
|
||||
/// it created the pong.
|
||||
void onPong(int pingId, int remoteTimestampMs) {
|
||||
final sentAt = _pending.remove(pingId);
|
||||
if (sentAt == null) return; // Not ours or already expired.
|
||||
|
||||
final now = _nowMs();
|
||||
final rtt = now - sentAt;
|
||||
if (rtt < 0 || rtt > _maxAcceptedRttMs) {
|
||||
appLogger.d('ClockSync: discarding sample with RTT=${rtt}ms');
|
||||
return;
|
||||
}
|
||||
|
||||
final offset = remoteTimestampMs - sentAt - (rtt ~/ 2);
|
||||
_samples.add((offsetMs: offset, rttMs: rtt));
|
||||
if (_samples.length > _windowSize) {
|
||||
_samples.removeAt(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,598 @@
|
||||
import 'dart:async';
|
||||
|
||||
import '../../utils/app_logger.dart';
|
||||
import '../models/playback_state.dart';
|
||||
import '../models/sync_message.dart';
|
||||
import '../models/watch_session.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.
|
||||
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].
|
||||
///
|
||||
/// Small drift is corrected invisibly with a brief playback-rate nudge
|
||||
/// (skipped while audio passthrough is active — rate changes tear bitstream
|
||||
/// output down); large drift hard-seeks with a post-seek settle window so we
|
||||
/// never measure mid-seek positions. Local user actions become
|
||||
/// [ControlRequest]s in anyone-mode (with a short optimistic window so the
|
||||
/// next heartbeat doesn't undo them before the host confirms) and snap back
|
||||
/// in host-only mode.
|
||||
class GuestPlaybackReconciler {
|
||||
GuestPlaybackReconciler({
|
||||
required this.myPeerId,
|
||||
required this._sendToHost,
|
||||
required ClockSync clockSync,
|
||||
this._callbacks = const GuestReconcilerCallbacks(),
|
||||
int Function()? nowMs,
|
||||
}) : _clock = clockSync,
|
||||
_nowMs = nowMs ?? _systemNowMs;
|
||||
|
||||
static int _systemNowMs() => DateTime.now().millisecondsSinceEpoch;
|
||||
|
||||
// Tuning constants.
|
||||
static const int tickMs = 500;
|
||||
static const int deadbandMs = 350;
|
||||
static const int nudgeExitMs = 150;
|
||||
static const double nudgeFactor = 0.04;
|
||||
static const int hardSeekThresholdMs = 2000;
|
||||
static const int hardSeekLeadMs = 250;
|
||||
static const int hardSeekCooldownMs = 2000;
|
||||
static const int pausedSeekThresholdMs = 500;
|
||||
static const int settleExtraMs = 250;
|
||||
static const int settleTimeoutMs = 1500;
|
||||
static const int optimisticWindowMs = 2000;
|
||||
static const int nudgeConfirmMs = 500;
|
||||
static const int bufferingStatusRefreshMs = 5000;
|
||||
static const int eofClampMs = 200;
|
||||
static const int eofToleranceMs = 1000;
|
||||
|
||||
final String myPeerId;
|
||||
final void Function(SyncMessage message) _sendToHost;
|
||||
final ClockSync _clock;
|
||||
final GuestReconcilerCallbacks _callbacks;
|
||||
final int Function() _nowMs;
|
||||
|
||||
PlaybackState? _latestState;
|
||||
int _lastSeq = -1;
|
||||
PlaybackPhase? _reportedPhase;
|
||||
List<String> _reportedWaitingOn = const [];
|
||||
ControlMode? _reportedControlMode;
|
||||
|
||||
AttachedPlayer? _player;
|
||||
final List<StreamSubscription<dynamic>> _playerSubscriptions = [];
|
||||
String? _attachedMediaKey;
|
||||
bool _localReady = false;
|
||||
bool _firstFrameSeen = false;
|
||||
bool _startupHoldResolved = true;
|
||||
|
||||
Timer? _tickTimer;
|
||||
bool _backgrounded = false;
|
||||
bool _disposed = false;
|
||||
|
||||
// Correction state.
|
||||
bool _settling = false;
|
||||
Timer? _settleTimer;
|
||||
bool _correcting = false;
|
||||
bool _nudging = false;
|
||||
bool _nudgeDisabled = false;
|
||||
bool _nudgeConfirmed = false;
|
||||
Timer? _nudgeConfirmTimer;
|
||||
int _lastHardSeekMs = -hardSeekCooldownMs;
|
||||
final List<int> _driftSamples = [];
|
||||
|
||||
// Scheduled group start.
|
||||
Timer? _scheduledStartTimer;
|
||||
int? _scheduledStartSeq;
|
||||
|
||||
// Optimistic window after sending a control request.
|
||||
int? _optimisticUntilSeq;
|
||||
int _optimisticDeadlineMs = 0;
|
||||
|
||||
// Status reporting.
|
||||
PeerStatus? _lastSentStatus;
|
||||
Timer? _statusRefreshTimer;
|
||||
|
||||
PlaybackState? get latestState => _latestState;
|
||||
bool get isCorrecting => _correcting;
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Public inputs
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
void attach(
|
||||
AttachedPlayer player, {
|
||||
required String ratingKey,
|
||||
required String serverId,
|
||||
bool hasFirstFrame = false,
|
||||
Future<void>? startupHold,
|
||||
}) {
|
||||
detachPlayer();
|
||||
_player = player;
|
||||
_attachedMediaKey = PlaybackState.mediaKeyFor(ratingKey: ratingKey, serverId: serverId);
|
||||
_firstFrameSeen = hasFirstFrame;
|
||||
_startupHoldResolved = startupHold == null;
|
||||
|
||||
if (startupHold != null) {
|
||||
startupHold.then((_) {
|
||||
if (_disposed || !identical(_player, player)) return;
|
||||
_startupHoldResolved = true;
|
||||
_maybeBecomeReady();
|
||||
});
|
||||
}
|
||||
|
||||
_playerSubscriptions.add(
|
||||
player.loadedSignals.listen((_) {
|
||||
if (_settling) {
|
||||
_settleTimer?.cancel();
|
||||
_settleTimer = Timer(const Duration(milliseconds: settleExtraMs), _endSettle);
|
||||
}
|
||||
if (!_firstFrameSeen) {
|
||||
_firstFrameSeen = true;
|
||||
_maybeBecomeReady();
|
||||
}
|
||||
}),
|
||||
);
|
||||
_playerSubscriptions.add(
|
||||
player.bufferingChanges.listen((_) {
|
||||
_sendStatus();
|
||||
}),
|
||||
);
|
||||
_playerSubscriptions.add(player.playingIntents.listen(_onLocalPlayingIntent));
|
||||
_playerSubscriptions.add(player.rateIntents.listen(_onLocalRateIntent));
|
||||
|
||||
_tickTimer = Timer.periodic(const Duration(milliseconds: tickMs), (_) => _onTick());
|
||||
if (_firstFrameSeen && _startupHoldResolved) {
|
||||
_localReady = true;
|
||||
appLogger.d('WatchTogether: Guest player ready for $_attachedMediaKey');
|
||||
}
|
||||
_sendStatus();
|
||||
if (_localReady) _reconcile();
|
||||
}
|
||||
|
||||
void _maybeBecomeReady() {
|
||||
if (_localReady || !_firstFrameSeen || !_startupHoldResolved) return;
|
||||
_localReady = true;
|
||||
appLogger.d('WatchTogether: Guest player ready for $_attachedMediaKey');
|
||||
_sendStatus();
|
||||
_reconcile();
|
||||
}
|
||||
|
||||
void detachPlayer() {
|
||||
for (final subscription in _playerSubscriptions) {
|
||||
unawaited(subscription.cancel());
|
||||
}
|
||||
_playerSubscriptions.clear();
|
||||
|
||||
// Tell the host we're no longer ready on this media (it re-gates us for
|
||||
// the next epoch start instead of waiting on a stale "ready").
|
||||
if (_player != null && _attachedMediaKey != null) {
|
||||
_lastSentStatus = null;
|
||||
_sendToHost(
|
||||
SyncMessage.status(
|
||||
PeerStatus(mediaKey: _attachedMediaKey!, ready: false, buffering: false, positionMs: 0),
|
||||
peerId: myPeerId,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
_player = null;
|
||||
_attachedMediaKey = null;
|
||||
_localReady = false;
|
||||
_firstFrameSeen = false;
|
||||
_startupHoldResolved = true;
|
||||
_tickTimer?.cancel();
|
||||
_tickTimer = null;
|
||||
_settleTimer?.cancel();
|
||||
_settleTimer = null;
|
||||
_settling = false;
|
||||
_nudging = false;
|
||||
_nudgeConfirmTimer?.cancel();
|
||||
_nudgeConfirmTimer = null;
|
||||
_scheduledStartTimer?.cancel();
|
||||
_scheduledStartTimer = null;
|
||||
_scheduledStartSeq = null;
|
||||
_statusRefreshTimer?.cancel();
|
||||
_statusRefreshTimer = null;
|
||||
_driftSamples.clear();
|
||||
_setCorrecting(false);
|
||||
}
|
||||
|
||||
/// Latest authoritative state from the host (already host-authenticated).
|
||||
void onState(PlaybackState state) {
|
||||
if (state.seq <= _lastSeq) return; // Stale or reordered.
|
||||
_lastSeq = state.seq;
|
||||
final previous = _latestState;
|
||||
_latestState = state;
|
||||
|
||||
if (state.controlMode != _reportedControlMode) {
|
||||
_reportedControlMode = state.controlMode;
|
||||
_callbacks.onControlModeChanged?.call(state.controlMode);
|
||||
}
|
||||
if (state.phase != _reportedPhase) {
|
||||
_reportedPhase = state.phase;
|
||||
_callbacks.onPhaseChanged?.call(state.phase);
|
||||
}
|
||||
if (!_listEquals(state.waitingOn, _reportedWaitingOn)) {
|
||||
_reportedWaitingOn = state.waitingOn;
|
||||
_callbacks.onWaitingOnChanged?.call(state.waitingOn);
|
||||
}
|
||||
if (state.actionHint != null && state.actorPeerId != null && state.actorPeerId != myPeerId) {
|
||||
_callbacks.onRemoteAction?.call(state.actorPeerId!, state.actionHint!);
|
||||
}
|
||||
|
||||
// Close the optimistic window only on an explicit transition (the host
|
||||
// applied our request — or someone else's superseding one). A plain
|
||||
// heartbeat that was already in flight when we sent the request still
|
||||
// carries the pre-request anchor and must not yank us back.
|
||||
if (_optimisticUntilSeq != null && (state.actorPeerId == myPeerId || state.actionHint != null)) {
|
||||
_optimisticUntilSeq = null;
|
||||
}
|
||||
|
||||
// Self-heal: the host thinks it's waiting on us but we're healthy.
|
||||
final player = _player;
|
||||
if (state.waitingOn.contains(myPeerId) && _localReady && player != null && !player.buffering) {
|
||||
_sendStatus(force: true);
|
||||
}
|
||||
|
||||
// The host moved to media we don't have — hand off to the switch flow.
|
||||
if (_attachedMediaKey != null && state.mediaKey != _attachedMediaKey) {
|
||||
_callbacks.onMediaSwitchNeeded?.call(state.ratingKey, state.serverId, state.mediaTitle);
|
||||
return;
|
||||
}
|
||||
if (previous?.mediaKey != state.mediaKey && _attachedMediaKey == null) {
|
||||
// Not in the player yet — let the provider navigate.
|
||||
_callbacks.onMediaSwitchNeeded?.call(state.ratingKey, state.serverId, state.mediaTitle);
|
||||
return;
|
||||
}
|
||||
|
||||
_reconcile();
|
||||
}
|
||||
|
||||
/// User seek on this guest (the screen already executed it locally).
|
||||
void onLocalSeekIntent(Duration position) {
|
||||
if (_latestState == null) return;
|
||||
if (_canControl) {
|
||||
_sendControl(ControlRequest(kind: ControlRequestKind.seek, positionMs: position.inMilliseconds));
|
||||
} else {
|
||||
_reconcile(); // Snap back.
|
||||
}
|
||||
}
|
||||
|
||||
void setBackgrounded(bool value) {
|
||||
if (_backgrounded == value) return;
|
||||
_backgrounded = value;
|
||||
if (!value) _reconcile();
|
||||
}
|
||||
|
||||
/// Host session restarted (fresh join observed) — accept its new counter.
|
||||
void resetSequence() {
|
||||
_lastSeq = -1;
|
||||
}
|
||||
|
||||
void onReconnected() {
|
||||
_sendStatus(force: true);
|
||||
}
|
||||
|
||||
void dispose() {
|
||||
_disposed = true;
|
||||
detachPlayer();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Local intents
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
bool get _canControl => _latestState?.controlMode == ControlMode.anyone;
|
||||
|
||||
void _onLocalPlayingIntent(bool playing) {
|
||||
if (_latestState == null) return;
|
||||
if (_canControl) {
|
||||
_sendControl(
|
||||
ControlRequest(
|
||||
kind: playing ? ControlRequestKind.play : ControlRequestKind.pause,
|
||||
positionMs: _player?.position.inMilliseconds,
|
||||
),
|
||||
);
|
||||
} else {
|
||||
_reconcile(); // Snap back to the room state.
|
||||
}
|
||||
}
|
||||
|
||||
void _onLocalRateIntent(double rate) {
|
||||
if (_latestState == null) return;
|
||||
if (_canControl) {
|
||||
_sendControl(ControlRequest(kind: ControlRequestKind.rate, rate: rate));
|
||||
} else {
|
||||
_reconcile();
|
||||
}
|
||||
}
|
||||
|
||||
void _sendControl(ControlRequest request) {
|
||||
_sendToHost(SyncMessage.control(request, peerId: myPeerId));
|
||||
_optimisticUntilSeq = _lastSeq;
|
||||
_optimisticDeadlineMs = _nowMs() + optimisticWindowMs;
|
||||
}
|
||||
|
||||
bool get _optimisticWindowActive => _optimisticUntilSeq != null && _nowMs() < _optimisticDeadlineMs;
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Reconciliation
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
void _onTick() {
|
||||
_reconcile();
|
||||
// Keep the host's view fresh while we're the one buffering.
|
||||
final player = _player;
|
||||
if (player != null && player.buffering && _statusRefreshTimer == null) {
|
||||
_statusRefreshTimer = Timer(const Duration(milliseconds: bufferingStatusRefreshMs), () {
|
||||
_statusRefreshTimer = null;
|
||||
if (_player?.buffering ?? false) _sendStatus(force: true);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
void _reconcile() {
|
||||
if (_disposed || _backgrounded || _settling) return;
|
||||
final state = _latestState;
|
||||
final player = _player;
|
||||
if (state == null || player == null || !_localReady) return;
|
||||
if (state.mediaKey != _attachedMediaKey) return;
|
||||
if (_optimisticWindowActive) return;
|
||||
|
||||
switch (state.phase) {
|
||||
case PlaybackPhase.loading:
|
||||
// Host is still loading — its anchor is meaningless. Just hold.
|
||||
_exitNudgeIfNeeded(state);
|
||||
_ensurePaused(player);
|
||||
break;
|
||||
|
||||
case PlaybackPhase.waitingForPeers:
|
||||
case PlaybackPhase.paused:
|
||||
_exitNudgeIfNeeded(state);
|
||||
_ensurePaused(player);
|
||||
_alignWhileStopped(player, state);
|
||||
break;
|
||||
|
||||
case PlaybackPhase.playing:
|
||||
_reconcilePlaying(player, state);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
void _reconcilePlaying(AttachedPlayer player, PlaybackState state) {
|
||||
final hostNow = _clock.hostNowMs();
|
||||
|
||||
// Scheduled group start: hold at the anchor, then start on the dot.
|
||||
if (state.anchorHostTimeMs > hostNow) {
|
||||
if (_scheduledStartSeq != state.seq) {
|
||||
_scheduledStartTimer?.cancel();
|
||||
_scheduledStartSeq = state.seq;
|
||||
final delay = state.anchorHostTimeMs - hostNow;
|
||||
appLogger.d('WatchTogether: Group start in ${delay}ms at ${state.anchorPositionMs}ms');
|
||||
_scheduledStartTimer = Timer(Duration(milliseconds: delay), () {
|
||||
_scheduledStartTimer = null;
|
||||
_scheduledStartSeq = null;
|
||||
final currentPlayer = _player;
|
||||
if (currentPlayer == null || _latestState?.seq != state.seq) return;
|
||||
unawaited(currentPlayer.play());
|
||||
});
|
||||
}
|
||||
_exitNudgeIfNeeded(state);
|
||||
_ensurePaused(player);
|
||||
_alignWhileStopped(player, state);
|
||||
return;
|
||||
}
|
||||
if (_scheduledStartSeq != null && _scheduledStartSeq != state.seq) {
|
||||
_scheduledStartTimer?.cancel();
|
||||
_scheduledStartTimer = null;
|
||||
_scheduledStartSeq = null;
|
||||
}
|
||||
|
||||
final durationMs = player.duration.inMilliseconds;
|
||||
var targetMs = state.targetPositionMs(hostNow);
|
||||
if (durationMs > 0 && targetMs > durationMs - eofClampMs) {
|
||||
targetMs = durationMs - eofClampMs;
|
||||
}
|
||||
|
||||
// Both of us rolled into the credits — don't fight EOF.
|
||||
if (player.completed && durationMs > 0 && targetMs >= durationMs - eofToleranceMs) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!player.playing) {
|
||||
if (player.completed) {
|
||||
// Fell off the end while the room plays on — rejoin via seek+play.
|
||||
if (player.seekable && _cooldownElapsed) {
|
||||
_hardSeek(player, targetMs, thenPlay: true);
|
||||
}
|
||||
return;
|
||||
}
|
||||
unawaited(player.play());
|
||||
}
|
||||
|
||||
// Base rate alignment (never while nudging — the nudge owns the rate).
|
||||
if (!_nudging && (player.rate - state.rate).abs() > 0.001) {
|
||||
unawaited(player.setRate(state.rate));
|
||||
}
|
||||
|
||||
if (!player.seekable) return; // Live: play/pause/rate only.
|
||||
|
||||
final drift = _smoothedDrift(player.position.inMilliseconds - targetMs);
|
||||
if (drift == null) return;
|
||||
|
||||
final magnitude = drift.abs();
|
||||
if (magnitude <= (_nudging ? nudgeExitMs : deadbandMs)) {
|
||||
_exitNudgeIfNeeded(state);
|
||||
return;
|
||||
}
|
||||
|
||||
if (magnitude <= deadbandMs) return; // Inside deadband, still nudging.
|
||||
|
||||
if (magnitude <= hardSeekThresholdMs) {
|
||||
_maybeNudge(player, state, drift);
|
||||
return;
|
||||
}
|
||||
|
||||
// Hard correction.
|
||||
_exitNudgeIfNeeded(state);
|
||||
if (!_cooldownElapsed) return;
|
||||
_hardSeek(player, targetMs + hardSeekLeadMs);
|
||||
}
|
||||
|
||||
bool get _cooldownElapsed => _nowMs() - _lastHardSeekMs >= hardSeekCooldownMs;
|
||||
|
||||
void _hardSeek(AttachedPlayer player, int targetMs, {bool thenPlay = false}) {
|
||||
_lastHardSeekMs = _nowMs();
|
||||
_driftSamples.clear();
|
||||
_setCorrecting(true);
|
||||
_beginSettle();
|
||||
appLogger.d('WatchTogether: Hard sync seek to ${targetMs}ms');
|
||||
unawaited(
|
||||
player.seek(Duration(milliseconds: targetMs.clamp(0, 1 << 48))).then((didSeek) async {
|
||||
if (didSeek && thenPlay) await player.play();
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
void _alignWhileStopped(AttachedPlayer player, PlaybackState state) {
|
||||
if (!player.seekable) return;
|
||||
final offBy = (player.position.inMilliseconds - state.anchorPositionMs).abs();
|
||||
if (offBy > pausedSeekThresholdMs && _cooldownElapsed) {
|
||||
_hardSeek(player, state.anchorPositionMs);
|
||||
}
|
||||
}
|
||||
|
||||
void _ensurePaused(AttachedPlayer player) {
|
||||
if (player.playing) {
|
||||
unawaited(player.pause());
|
||||
}
|
||||
}
|
||||
|
||||
void _maybeNudge(AttachedPlayer player, PlaybackState state, int drift) {
|
||||
if (_nudgeDisabled || player.passthroughActive) return; // Tolerate up to the seek band.
|
||||
|
||||
// Ahead of the room → slow down; behind → speed up.
|
||||
final factor = drift > 0 ? (1 - nudgeFactor) : (1 + nudgeFactor);
|
||||
final targetRate = state.rate * factor;
|
||||
if (_nudging && (player.rate - targetRate).abs() < 0.001) return;
|
||||
|
||||
_nudging = true;
|
||||
unawaited(player.setRate(targetRate));
|
||||
|
||||
// Arm the capability check once per (un-confirmed) nudge episode — a
|
||||
// re-issued nudge must not keep pushing the deadline out.
|
||||
if (!_nudgeConfirmed && _nudgeConfirmTimer == null) {
|
||||
_nudgeConfirmTimer = Timer(const Duration(milliseconds: nudgeConfirmMs), () {
|
||||
_nudgeConfirmTimer = null;
|
||||
final currentPlayer = _player;
|
||||
if (currentPlayer == null || !_nudging) return;
|
||||
if ((currentPlayer.rate - targetRate).abs() > 0.005) {
|
||||
appLogger.w('WatchTogether: Rate nudges not taking effect — disabling for this session');
|
||||
_nudgeDisabled = true;
|
||||
_nudging = false;
|
||||
unawaited(currentPlayer.setRate(_latestState?.rate ?? 1.0));
|
||||
} else {
|
||||
_nudgeConfirmed = true;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
void _exitNudgeIfNeeded(PlaybackState state) {
|
||||
if (!_nudging) return;
|
||||
_nudging = false;
|
||||
final player = _player;
|
||||
if (player != null) {
|
||||
unawaited(player.setRate(state.rate));
|
||||
}
|
||||
}
|
||||
|
||||
int? _smoothedDrift(int rawDrift) {
|
||||
_driftSamples.add(rawDrift);
|
||||
if (_driftSamples.length > 3) _driftSamples.removeAt(0);
|
||||
if (_driftSamples.length < 2) return null; // One sample can be a fluke.
|
||||
final sorted = List<int>.of(_driftSamples)..sort();
|
||||
return sorted[sorted.length ~/ 2];
|
||||
}
|
||||
|
||||
void _beginSettle() {
|
||||
_settling = true;
|
||||
_settleTimer?.cancel();
|
||||
_settleTimer = Timer(const Duration(milliseconds: settleTimeoutMs), _endSettle);
|
||||
}
|
||||
|
||||
void _endSettle() {
|
||||
if (!_settling) return;
|
||||
_settling = false;
|
||||
_settleTimer?.cancel();
|
||||
_settleTimer = null;
|
||||
_driftSamples.clear();
|
||||
_setCorrecting(false);
|
||||
}
|
||||
|
||||
void _setCorrecting(bool value) {
|
||||
if (_correcting == value) return;
|
||||
_correcting = value;
|
||||
_callbacks.onCorrectingChanged?.call(value);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Status reporting
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
void _sendStatus({bool force = false}) {
|
||||
final mediaKey = _attachedMediaKey;
|
||||
if (mediaKey == null || _disposed) return;
|
||||
final player = _player;
|
||||
final status = PeerStatus(
|
||||
mediaKey: mediaKey,
|
||||
ready: _localReady,
|
||||
buffering: player?.buffering ?? false,
|
||||
positionMs: player?.position.inMilliseconds ?? 0,
|
||||
rttMs: _clock.minRttMs,
|
||||
);
|
||||
final last = _lastSentStatus;
|
||||
if (!force &&
|
||||
last != null &&
|
||||
last.mediaKey == status.mediaKey &&
|
||||
last.ready == status.ready &&
|
||||
last.buffering == status.buffering) {
|
||||
return;
|
||||
}
|
||||
_lastSentStatus = status;
|
||||
_sendToHost(SyncMessage.status(status, peerId: myPeerId));
|
||||
}
|
||||
|
||||
static bool _listEquals(List<String> a, List<String> b) {
|
||||
if (a.length != b.length) return false;
|
||||
for (var i = 0; i < a.length; i++) {
|
||||
if (a[i] != b[i]) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,802 @@
|
||||
import 'dart:async';
|
||||
import 'dart:math';
|
||||
|
||||
import '../../utils/app_logger.dart';
|
||||
import '../models/playback_state.dart';
|
||||
import '../models/watch_session.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
|
||||
/// streams), peer status reports, control requests, and roster changes; the
|
||||
/// output is a state broadcast through [sendState] plus commands to the
|
||||
/// host's own player (the host delays its own start to the scheduled moment
|
||||
/// just like every guest).
|
||||
///
|
||||
/// Pure Dart and clock-injected so the full scenario matrix runs under
|
||||
/// `fakeAsync`.
|
||||
class HostPlaybackCoordinator {
|
||||
HostPlaybackCoordinator({
|
||||
required this.myPeerId,
|
||||
required this._controlMode,
|
||||
required this._sendState,
|
||||
this._callbacks = const HostCoordinatorCallbacks(),
|
||||
int Function()? nowMs,
|
||||
}) : _nowMs = nowMs ?? _systemNowMs;
|
||||
|
||||
static int _systemNowMs() => DateTime.now().millisecondsSinceEpoch;
|
||||
|
||||
// Tuning constants.
|
||||
static const int stallGraceMs = 500;
|
||||
static const int recoveryHysteresisMs = 400;
|
||||
static const int safetyTimeoutMs = 15000;
|
||||
static const int heartbeatPlayingMs = 2000;
|
||||
static const int heartbeatIdleMs = 5000;
|
||||
static const int startDelayMinMs = 750;
|
||||
static const int startDelayMaxMs = 2000;
|
||||
static const int defaultPeerRttMs = 500;
|
||||
static const int seekDebounceMs = 200;
|
||||
static const int implicitJumpThresholdMs = 1500;
|
||||
static const int selfRecoveryMinBufferAheadMs = 2000;
|
||||
|
||||
final String myPeerId;
|
||||
final void Function(PlaybackState state, {String? toPeerId}) _sendState;
|
||||
final HostCoordinatorCallbacks _callbacks;
|
||||
final int Function() _nowMs;
|
||||
|
||||
ControlMode _controlMode;
|
||||
|
||||
// Media epoch.
|
||||
String? _ratingKey;
|
||||
String? _serverId;
|
||||
String? _mediaTitle;
|
||||
bool get hasActiveEpoch => _ratingKey != null && _serverId != null;
|
||||
String? get _mediaKey =>
|
||||
hasActiveEpoch ? PlaybackState.mediaKeyFor(ratingKey: _ratingKey!, serverId: _serverId!) : null;
|
||||
|
||||
// Player attachment.
|
||||
AttachedPlayer? _player;
|
||||
final List<StreamSubscription<dynamic>> _playerSubscriptions = [];
|
||||
bool _localReady = false;
|
||||
bool _startupHoldResolved = true;
|
||||
bool _localStalled = false;
|
||||
bool _recoveringFromSelfStall = false;
|
||||
|
||||
// Room state.
|
||||
PlaybackPhase _phase = PlaybackPhase.loading;
|
||||
bool _intendedPlaying = false;
|
||||
double _rate = 1.0;
|
||||
bool _firstStartCompleted = false;
|
||||
int _seq = 0;
|
||||
PlaybackState? _lastBroadcast;
|
||||
bool _backgrounded = false;
|
||||
|
||||
// Peer tracking.
|
||||
final Set<String> _knownPeers = {};
|
||||
final Set<String> _incompatiblePeers = {};
|
||||
final Set<String> _excused = {};
|
||||
final Set<String> _stalledPeers = {};
|
||||
final Map<String, PeerStatus> _peerStatuses = {};
|
||||
final Map<String, Timer> _peerStallGraceTimers = {};
|
||||
|
||||
// Pending actions.
|
||||
Timer? _selfStallGraceTimer;
|
||||
Timer? _allReadyCheckTimer;
|
||||
Timer? _safetyTimer;
|
||||
Timer? _heartbeatTimer;
|
||||
Timer? _pendingStartTimer;
|
||||
int? _pendingStartAtMs;
|
||||
int? _pendingStartPositionMs;
|
||||
Timer? _seekDebounceTimer;
|
||||
int? _pendingSeekTargetMs;
|
||||
String? _pendingActor;
|
||||
bool _disposed = false;
|
||||
|
||||
PlaybackPhase get phase => _phase;
|
||||
Set<String> get incompatiblePeers => Set.unmodifiable(_incompatiblePeers);
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Public inputs
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
/// Host switched (or initially picked) media — a new epoch. Safe to call
|
||||
/// repeatedly with the same media; only an actual change broadcasts.
|
||||
void setLocalMedia({required String ratingKey, required String serverId, String? mediaTitle}) {
|
||||
final newKey = PlaybackState.mediaKeyFor(ratingKey: ratingKey, serverId: serverId);
|
||||
if (newKey == _mediaKey) {
|
||||
if (mediaTitle != null && mediaTitle != _mediaTitle) _mediaTitle = mediaTitle;
|
||||
return;
|
||||
}
|
||||
|
||||
_ratingKey = ratingKey;
|
||||
_serverId = serverId;
|
||||
_mediaTitle = mediaTitle;
|
||||
_localReady = false;
|
||||
_localStalled = false;
|
||||
_recoveringFromSelfStall = false;
|
||||
_firstStartCompleted = false;
|
||||
_intendedPlaying = true; // Opening media implies the room wants to play.
|
||||
_excused.clear();
|
||||
_stalledPeers.clear();
|
||||
_cancelPendingStart();
|
||||
_cancelSafety();
|
||||
_cancelStallTimers();
|
||||
_setPhase(PlaybackPhase.loading);
|
||||
_broadcast(hint: PlaybackActionHint.mediaSwitch, actor: myPeerId);
|
||||
appLogger.d('WatchTogether: Host epoch -> $newKey');
|
||||
}
|
||||
|
||||
/// Attach the host's player for the given media. [hasFirstFrame] is the
|
||||
/// screen's first-frame snapshot (covers attaching to an already-rendering
|
||||
/// player); [startupHold] delays readiness past platform startup gates
|
||||
/// (e.g. the Android frame-rate switch).
|
||||
void attach(
|
||||
AttachedPlayer player, {
|
||||
required String ratingKey,
|
||||
required String serverId,
|
||||
String? mediaTitle,
|
||||
bool hasFirstFrame = false,
|
||||
Future<void>? startupHold,
|
||||
}) {
|
||||
detachPlayer();
|
||||
final sameEpoch =
|
||||
hasActiveEpoch && PlaybackState.mediaKeyFor(ratingKey: ratingKey, serverId: serverId) == _mediaKey;
|
||||
setLocalMedia(ratingKey: ratingKey, serverId: serverId, mediaTitle: mediaTitle);
|
||||
|
||||
_player = player;
|
||||
_rate = player.rate;
|
||||
|
||||
// Same-media re-attach with a reloading player (quality/version switch):
|
||||
// group-wait at the last known spot until we render again, then the
|
||||
// normal all-ready resolution resumes the room.
|
||||
if (sameEpoch && !hasFirstFrame && _phase == PlaybackPhase.playing) {
|
||||
_intendedPlaying = true;
|
||||
_cancelPendingStart();
|
||||
_setPhase(PlaybackPhase.waitingForPeers);
|
||||
_broadcast(anchorPositionOverrideMs: _lastBroadcast?.anchorPositionMs);
|
||||
_armSafetyIfGated();
|
||||
}
|
||||
|
||||
_startupHoldResolved = startupHold == null;
|
||||
if (startupHold != null) {
|
||||
startupHold.then((_) {
|
||||
if (_disposed || !identical(_player, player)) return;
|
||||
_startupHoldResolved = true;
|
||||
_maybeLocalLoaded();
|
||||
});
|
||||
}
|
||||
|
||||
_playerSubscriptions.add(player.loadedSignals.listen((_) => _onLoadedSignal()));
|
||||
_playerSubscriptions.add(player.bufferingChanges.listen(_onSelfBuffering));
|
||||
_playerSubscriptions.add(player.playingIntents.listen(_onLocalPlayingIntent));
|
||||
_playerSubscriptions.add(player.rateIntents.listen(_onLocalRateIntent));
|
||||
|
||||
if (hasFirstFrame) {
|
||||
_localReady = true;
|
||||
_maybeLocalLoaded();
|
||||
}
|
||||
_restartHeartbeat();
|
||||
}
|
||||
|
||||
/// Detach the player (episode switch keeps the session; [exiting] ends the
|
||||
/// epoch because the host left the video player).
|
||||
void detachPlayer({bool exiting = false}) {
|
||||
for (final subscription in _playerSubscriptions) {
|
||||
unawaited(subscription.cancel());
|
||||
}
|
||||
_playerSubscriptions.clear();
|
||||
_player = null;
|
||||
_localReady = false;
|
||||
_localStalled = false;
|
||||
_recoveringFromSelfStall = false;
|
||||
_startupHoldResolved = true;
|
||||
_cancelPendingStart();
|
||||
_cancelStallTimers();
|
||||
_heartbeatTimer?.cancel();
|
||||
_heartbeatTimer = null;
|
||||
if (exiting) {
|
||||
_ratingKey = null;
|
||||
_serverId = null;
|
||||
_mediaTitle = null;
|
||||
_cancelSafety();
|
||||
_setPhase(PlaybackPhase.loading);
|
||||
}
|
||||
}
|
||||
|
||||
void setBackgrounded(bool value) {
|
||||
if (_backgrounded == value) return;
|
||||
_backgrounded = value;
|
||||
if (!value && hasActiveEpoch && _player != null) {
|
||||
_onHeartbeat();
|
||||
}
|
||||
}
|
||||
|
||||
void updateControlMode(ControlMode mode) {
|
||||
if (_controlMode == mode) return;
|
||||
_controlMode = mode;
|
||||
if (hasActiveEpoch) _broadcast();
|
||||
}
|
||||
|
||||
void onPeerJoined(String peerId, {required bool compatible}) {
|
||||
if (peerId == myPeerId) return;
|
||||
if (!compatible) {
|
||||
_incompatiblePeers.add(peerId);
|
||||
_knownPeers.remove(peerId);
|
||||
return;
|
||||
}
|
||||
_incompatiblePeers.remove(peerId);
|
||||
_knownPeers.add(peerId);
|
||||
if (hasActiveEpoch) {
|
||||
_broadcast(toPeerId: peerId);
|
||||
}
|
||||
}
|
||||
|
||||
void onPeerLeft(String peerId) {
|
||||
_knownPeers.remove(peerId);
|
||||
_incompatiblePeers.remove(peerId);
|
||||
_excused.remove(peerId);
|
||||
_stalledPeers.remove(peerId);
|
||||
_peerStatuses.remove(peerId);
|
||||
_peerStallGraceTimers.remove(peerId)?.cancel();
|
||||
_scheduleAllReadyCheck(0);
|
||||
}
|
||||
|
||||
void onPeerStatus(String peerId, PeerStatus status) {
|
||||
if (peerId == myPeerId || _incompatiblePeers.contains(peerId)) return;
|
||||
_knownPeers.add(peerId);
|
||||
final previous = _peerStatuses[peerId];
|
||||
_peerStatuses[peerId] = status;
|
||||
|
||||
final onCurrentEpoch = status.mediaKey == _mediaKey;
|
||||
|
||||
// A previously-excused peer that is healthy again rejoins the gate set.
|
||||
if (onCurrentEpoch && status.ready && !status.buffering) {
|
||||
_excused.remove(peerId);
|
||||
}
|
||||
|
||||
if (!onCurrentEpoch) {
|
||||
_peerStallGraceTimers.remove(peerId)?.cancel();
|
||||
_stalledPeers.remove(peerId);
|
||||
_scheduleAllReadyCheck(0);
|
||||
return;
|
||||
}
|
||||
|
||||
// Stall detection: a ready peer that reports buffering while the room
|
||||
// plays gets a short grace window before pausing everyone.
|
||||
if (status.ready && status.buffering) {
|
||||
if (_phase == PlaybackPhase.playing && !_stalledPeers.contains(peerId)) {
|
||||
_peerStallGraceTimers[peerId] ??= Timer(const Duration(milliseconds: stallGraceMs), () {
|
||||
_peerStallGraceTimers.remove(peerId);
|
||||
final latest = _peerStatuses[peerId];
|
||||
if (latest == null || !latest.buffering || latest.mediaKey != _mediaKey) return;
|
||||
if (_phase != PlaybackPhase.playing) return;
|
||||
_stalledPeers.add(peerId);
|
||||
_enterWaiting();
|
||||
});
|
||||
} else if (_phase == PlaybackPhase.waitingForPeers && !_stalledPeers.contains(peerId)) {
|
||||
// Already waiting on someone else — fold this stall in immediately.
|
||||
_stalledPeers.add(peerId);
|
||||
_scheduleAllReadyCheck(0);
|
||||
}
|
||||
} else {
|
||||
_peerStallGraceTimers.remove(peerId)?.cancel();
|
||||
final wasStalled = _stalledPeers.remove(peerId);
|
||||
final becameReady = status.ready && (previous == null || !previous.ready || previous.mediaKey != _mediaKey);
|
||||
if (wasStalled) {
|
||||
_scheduleAllReadyCheck(recoveryHysteresisMs);
|
||||
} else if (becameReady) {
|
||||
_scheduleAllReadyCheck(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void onControlRequest(String peerId, ControlRequest request) {
|
||||
if (!hasActiveEpoch) return;
|
||||
switch (request.kind) {
|
||||
case ControlRequestKind.play:
|
||||
_requestPlay(actor: peerId);
|
||||
break;
|
||||
case ControlRequestKind.pause:
|
||||
_requestPause(actor: peerId);
|
||||
break;
|
||||
case ControlRequestKind.seek:
|
||||
if (request.positionMs != null) {
|
||||
_applyRemoteSeek(request.positionMs!, actor: peerId);
|
||||
}
|
||||
break;
|
||||
case ControlRequestKind.rate:
|
||||
if (request.rate != null) {
|
||||
_applyRemoteRate(request.rate!, actor: peerId);
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/// User seek on the host (the screen already executed it on the player).
|
||||
void onLocalSeekIntent(Duration position) {
|
||||
if (!hasActiveEpoch) return;
|
||||
_pendingSeekTargetMs = position.inMilliseconds;
|
||||
_seekDebounceTimer?.cancel();
|
||||
_seekDebounceTimer = Timer(const Duration(milliseconds: seekDebounceMs), () {
|
||||
final target = _pendingSeekTargetMs;
|
||||
_pendingSeekTargetMs = null;
|
||||
if (target == null || !hasActiveEpoch) return;
|
||||
_afterHostSeek(target, actor: myPeerId);
|
||||
});
|
||||
}
|
||||
|
||||
void onStateRequested(String peerId) {
|
||||
if (!hasActiveEpoch) return;
|
||||
_broadcast(toPeerId: peerId);
|
||||
}
|
||||
|
||||
void onReconnected() {
|
||||
if (hasActiveEpoch) _broadcast();
|
||||
}
|
||||
|
||||
void dispose() {
|
||||
_disposed = true;
|
||||
detachPlayer(exiting: true);
|
||||
_allReadyCheckTimer?.cancel();
|
||||
_seekDebounceTimer?.cancel();
|
||||
_peerStatuses.clear();
|
||||
_knownPeers.clear();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Local player signals
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
void _onLoadedSignal() {
|
||||
if (_localReady) return;
|
||||
_localReady = true;
|
||||
_maybeLocalLoaded();
|
||||
}
|
||||
|
||||
void _maybeLocalLoaded() {
|
||||
if (!_localReady || !_startupHoldResolved || !hasActiveEpoch) return;
|
||||
final player = _player;
|
||||
if (player == null) return;
|
||||
|
||||
appLogger.d('WatchTogether: Host player ready for $_mediaKey');
|
||||
|
||||
if (_phase == PlaybackPhase.loading) {
|
||||
// The sync layer owns the start — undo anything that slipped into play.
|
||||
if (player.playing) {
|
||||
unawaited(player.pause());
|
||||
}
|
||||
_setPhase(PlaybackPhase.waitingForPeers);
|
||||
_broadcast();
|
||||
_armSafetyIfGated();
|
||||
}
|
||||
_scheduleAllReadyCheck(0);
|
||||
|
||||
// A play latched while we were loading (paused room) resumes now.
|
||||
if (_phase == PlaybackPhase.paused && _intendedPlaying) {
|
||||
_requestPlay(actor: _pendingActor ?? myPeerId);
|
||||
}
|
||||
}
|
||||
|
||||
void _onSelfBuffering(bool buffering) {
|
||||
if (!_localReady) return; // Pre-ready buffering is the loading flow.
|
||||
|
||||
if (buffering) {
|
||||
_recoveringFromSelfStall = false;
|
||||
if (_phase != PlaybackPhase.playing || _localStalled) return;
|
||||
_selfStallGraceTimer?.cancel();
|
||||
_selfStallGraceTimer = Timer(const Duration(milliseconds: stallGraceMs), () {
|
||||
final player = _player;
|
||||
if (player == null || !player.buffering || _phase != PlaybackPhase.playing) return;
|
||||
_localStalled = true;
|
||||
// Unlike a remote stall we leave the host player unpaused so mpv can
|
||||
// refill its cache and recover on its own; its clock is frozen anyway.
|
||||
_enterWaiting();
|
||||
});
|
||||
} else {
|
||||
_selfStallGraceTimer?.cancel();
|
||||
_selfStallGraceTimer = null;
|
||||
if (_localStalled) {
|
||||
_localStalled = false;
|
||||
_recoveringFromSelfStall = true;
|
||||
_scheduleAllReadyCheck(recoveryHysteresisMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void _onLocalPlayingIntent(bool playing) {
|
||||
if (!hasActiveEpoch) return;
|
||||
if (playing) {
|
||||
_requestPlay(actor: myPeerId);
|
||||
} else {
|
||||
_requestPause(actor: myPeerId);
|
||||
}
|
||||
}
|
||||
|
||||
void _onLocalRateIntent(double rate) {
|
||||
if (!hasActiveEpoch) return;
|
||||
_rate = rate;
|
||||
_broadcast(hint: PlaybackActionHint.rate, actor: myPeerId);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Play / pause / seek / rate policy
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
void _requestPlay({required String actor}) {
|
||||
if (_phase == PlaybackPhase.playing) return;
|
||||
_intendedPlaying = true;
|
||||
_pendingActor = actor;
|
||||
if (actor != myPeerId) _callbacks.onRemoteAction?.call(actor, PlaybackActionHint.play);
|
||||
|
||||
final player = _player;
|
||||
if (!_localReady) {
|
||||
// Still loading: latch the intent, undo any local unpause, and stay in
|
||||
// the loading phase — its anchor would be meaningless to guests.
|
||||
if (player != null && player.playing) {
|
||||
unawaited(player.pause());
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
final gating = _gatingPeers();
|
||||
if (gating.isEmpty) {
|
||||
_scheduleStart(actor: actor);
|
||||
} else {
|
||||
// Want to play but can't yet — hold (and undo a local unpause).
|
||||
if (player != null && player.playing) {
|
||||
unawaited(player.pause());
|
||||
}
|
||||
if (_phase != PlaybackPhase.waitingForPeers) {
|
||||
_setPhase(PlaybackPhase.waitingForPeers);
|
||||
_broadcast(actor: actor);
|
||||
_armSafetyIfGated();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void _requestPause({required String actor}) {
|
||||
_intendedPlaying = false;
|
||||
_pendingActor = null;
|
||||
_cancelPendingStart();
|
||||
_cancelSafety();
|
||||
if (actor != myPeerId) _callbacks.onRemoteAction?.call(actor, PlaybackActionHint.pause);
|
||||
|
||||
final player = _player;
|
||||
if (player != null && player.playing) {
|
||||
unawaited(player.pause());
|
||||
}
|
||||
// While loading, only latch the intent — the all-ready resolution after
|
||||
// local readiness lands on paused because _intendedPlaying is false.
|
||||
if (_phase == PlaybackPhase.loading) return;
|
||||
_setPhase(PlaybackPhase.paused);
|
||||
_broadcast(hint: PlaybackActionHint.pause, actor: actor);
|
||||
}
|
||||
|
||||
void _applyRemoteSeek(int targetMs, {required String actor}) {
|
||||
final player = _player;
|
||||
if (player == null) return;
|
||||
_callbacks.onRemoteAction?.call(actor, PlaybackActionHint.seek);
|
||||
unawaited(
|
||||
player.seek(Duration(milliseconds: targetMs)).then((didSeek) {
|
||||
if (didSeek) _afterHostSeek(targetMs, actor: actor);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
void _afterHostSeek(int targetMs, {required String actor}) {
|
||||
// Re-anchor at the seek target. If a scheduled start is pending, move
|
||||
// its position too so the start fires from the new spot.
|
||||
if (_pendingStartAtMs != null) {
|
||||
_pendingStartPositionMs = targetMs;
|
||||
}
|
||||
_broadcast(hint: PlaybackActionHint.seek, actor: actor, anchorPositionOverrideMs: targetMs);
|
||||
}
|
||||
|
||||
void _applyRemoteRate(double rate, {required String actor}) {
|
||||
final player = _player;
|
||||
if (player == null) return;
|
||||
_callbacks.onRemoteAction?.call(actor, PlaybackActionHint.rate);
|
||||
unawaited(
|
||||
player.setRate(rate).then((didSet) {
|
||||
if (!didSet) return;
|
||||
_rate = rate;
|
||||
_broadcast(hint: PlaybackActionHint.rate, actor: actor);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Readiness / group-wait machinery
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
/// Peers (including self) the room cannot play without right now.
|
||||
Set<String> _gatingPeers() {
|
||||
final gating = <String>{};
|
||||
final mediaKey = _mediaKey;
|
||||
if (mediaKey == null) return gating;
|
||||
|
||||
for (final peerId in _knownPeers) {
|
||||
if (_excused.contains(peerId)) continue;
|
||||
final status = _peerStatuses[peerId];
|
||||
if (status == null || status.mediaKey != mediaKey) {
|
||||
// Never reported for this epoch: gate only the initial start —
|
||||
// mid-session they're late joiners who catch up on their own.
|
||||
if (!_firstStartCompleted) gating.add(peerId);
|
||||
continue;
|
||||
}
|
||||
if (!status.ready) {
|
||||
if (!_firstStartCompleted) gating.add(peerId);
|
||||
continue;
|
||||
}
|
||||
if (_stalledPeers.contains(peerId)) gating.add(peerId);
|
||||
}
|
||||
if (!_localReady || _localStalled) gating.add(myPeerId);
|
||||
return gating;
|
||||
}
|
||||
|
||||
void _enterWaiting() {
|
||||
if (_phase == PlaybackPhase.waitingForPeers) return;
|
||||
final player = _player;
|
||||
// Anchor where the room stops. Pause our player unless the stall is our
|
||||
// own (mpv recovers paused-for-cache by itself).
|
||||
if (player != null && player.playing && !_localStalled) {
|
||||
unawaited(player.pause());
|
||||
}
|
||||
_intendedPlaying = true; // A stall interrupts playback we intend to resume.
|
||||
_setPhase(PlaybackPhase.waitingForPeers);
|
||||
_broadcast();
|
||||
_armSafetyIfGated();
|
||||
}
|
||||
|
||||
void _scheduleAllReadyCheck(int delayMs) {
|
||||
_allReadyCheckTimer?.cancel();
|
||||
_allReadyCheckTimer = null;
|
||||
if (delayMs <= 0) {
|
||||
_checkAllReady();
|
||||
} else {
|
||||
_allReadyCheckTimer = Timer(Duration(milliseconds: delayMs), _checkAllReady);
|
||||
}
|
||||
}
|
||||
|
||||
void _checkAllReady() {
|
||||
if (_disposed || _phase != PlaybackPhase.waitingForPeers) return;
|
||||
final gating = _gatingPeers();
|
||||
if (gating.isNotEmpty) {
|
||||
_broadcastIfWaitingOnChanged(gating);
|
||||
return;
|
||||
}
|
||||
|
||||
// After our own stall, require some cache headroom before resuming so we
|
||||
// don't immediately drag the room back into a stall.
|
||||
final player = _player;
|
||||
if (_recoveringFromSelfStall && player != null) {
|
||||
if (player.buffering) return; // A new stall event will re-drive us.
|
||||
final ahead = player.bufferAhead;
|
||||
if (ahead != null && ahead.inMilliseconds < selfRecoveryMinBufferAheadMs) {
|
||||
_scheduleAllReadyCheck(500);
|
||||
return;
|
||||
}
|
||||
}
|
||||
_recoveringFromSelfStall = false;
|
||||
_resolveAllReady();
|
||||
}
|
||||
|
||||
void _resolveAllReady() {
|
||||
_cancelSafety();
|
||||
if (_intendedPlaying) {
|
||||
_scheduleStart(actor: _pendingActor ?? myPeerId);
|
||||
} else {
|
||||
_setPhase(PlaybackPhase.paused);
|
||||
_broadcast();
|
||||
}
|
||||
_pendingActor = null;
|
||||
}
|
||||
|
||||
void _scheduleStart({required String actor}) {
|
||||
final player = _player;
|
||||
if (player == null || !_localReady) return;
|
||||
_cancelPendingStart();
|
||||
|
||||
final otherPeers = _knownPeers.where((p) => !_excused.contains(p)).toList();
|
||||
int delayMs;
|
||||
if (otherPeers.isEmpty) {
|
||||
delayMs = 0;
|
||||
} else {
|
||||
var maxRtt = 0;
|
||||
for (final peerId in otherPeers) {
|
||||
maxRtt = max(maxRtt, _peerStatuses[peerId]?.rttMs ?? defaultPeerRttMs);
|
||||
}
|
||||
delayMs = max(startDelayMinMs, min((maxRtt * 1.5).round(), startDelayMaxMs));
|
||||
}
|
||||
|
||||
final startAt = _nowMs() + delayMs;
|
||||
final startPositionMs = player.position.inMilliseconds;
|
||||
_pendingStartAtMs = startAt;
|
||||
_pendingStartPositionMs = startPositionMs;
|
||||
_firstStartCompleted = true;
|
||||
_setPhase(PlaybackPhase.playing);
|
||||
_broadcast(hint: PlaybackActionHint.play, actor: actor);
|
||||
|
||||
void fireStart() {
|
||||
_pendingStartTimer = null;
|
||||
_pendingStartAtMs = null;
|
||||
final startPos = _pendingStartPositionMs;
|
||||
_pendingStartPositionMs = null;
|
||||
final currentPlayer = _player;
|
||||
if (currentPlayer == null || _phase != PlaybackPhase.playing) return;
|
||||
if (startPos != null && (currentPlayer.position.inMilliseconds - startPos).abs() > 250) {
|
||||
unawaited(currentPlayer.seek(Duration(milliseconds: startPos)).then((_) => currentPlayer.play()));
|
||||
} else {
|
||||
unawaited(currentPlayer.play());
|
||||
}
|
||||
}
|
||||
|
||||
if (delayMs <= 0 && player.playing) {
|
||||
// Solo resume of an already-playing player: nothing to do.
|
||||
_pendingStartTimer = null;
|
||||
_pendingStartAtMs = null;
|
||||
_pendingStartPositionMs = null;
|
||||
} else {
|
||||
// The host waits for the group moment like everyone else — undo a
|
||||
// user-initiated unpause until the scheduled start fires.
|
||||
if (player.playing) {
|
||||
unawaited(player.pause());
|
||||
}
|
||||
_pendingStartTimer = Timer(Duration(milliseconds: delayMs), fireStart);
|
||||
}
|
||||
}
|
||||
|
||||
void _armSafetyIfGated() {
|
||||
_cancelSafety();
|
||||
if (_gatingPeers().difference({myPeerId}).isEmpty) return;
|
||||
_safetyTimer = Timer(const Duration(milliseconds: safetyTimeoutMs), () {
|
||||
if (_phase != PlaybackPhase.waitingForPeers) return;
|
||||
final gating = _gatingPeers()..remove(myPeerId);
|
||||
if (gating.isEmpty) return;
|
||||
_excused.addAll(gating);
|
||||
_stalledPeers.removeAll(gating);
|
||||
appLogger.w('WatchTogether: Resuming without ${gating.join(', ')} after ${safetyTimeoutMs ~/ 1000}s');
|
||||
_callbacks.onResumedWithout?.call(gating.toList()..sort());
|
||||
_scheduleAllReadyCheck(0);
|
||||
});
|
||||
}
|
||||
|
||||
void _cancelPendingStart() {
|
||||
_pendingStartTimer?.cancel();
|
||||
_pendingStartTimer = null;
|
||||
_pendingStartAtMs = null;
|
||||
_pendingStartPositionMs = null;
|
||||
}
|
||||
|
||||
void _cancelSafety() {
|
||||
_safetyTimer?.cancel();
|
||||
_safetyTimer = null;
|
||||
}
|
||||
|
||||
void _cancelStallTimers() {
|
||||
_selfStallGraceTimer?.cancel();
|
||||
_selfStallGraceTimer = null;
|
||||
for (final timer in _peerStallGraceTimers.values) {
|
||||
timer.cancel();
|
||||
}
|
||||
_peerStallGraceTimers.clear();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Heartbeat & broadcasting
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
void _restartHeartbeat() {
|
||||
_heartbeatTimer?.cancel();
|
||||
if (_player == null) return;
|
||||
final interval = _phase == PlaybackPhase.playing ? heartbeatPlayingMs : heartbeatIdleMs;
|
||||
_heartbeatTimer = Timer.periodic(Duration(milliseconds: interval), (_) => _onHeartbeat());
|
||||
}
|
||||
|
||||
void _onHeartbeat() {
|
||||
if (_backgrounded || _disposed || !hasActiveEpoch) return;
|
||||
final player = _player;
|
||||
if (player == null) return;
|
||||
|
||||
// Implicit-jump detection: a position far from where the last broadcast
|
||||
// predicts, with no seek intent in flight, means something seeked the
|
||||
// player behind our back (OS remote, EOF jump) — re-anchor with a seek
|
||||
// hint so guests snap instead of nudging.
|
||||
PlaybackActionHint? hint;
|
||||
final last = _lastBroadcast;
|
||||
if (last != null && _pendingStartAtMs == null && _pendingSeekTargetMs == null && !player.buffering) {
|
||||
final expected = last.targetPositionMs(_nowMs());
|
||||
if ((player.position.inMilliseconds - expected).abs() > implicitJumpThresholdMs) {
|
||||
hint = PlaybackActionHint.seek;
|
||||
}
|
||||
}
|
||||
_broadcast(hint: hint, actor: hint != null ? myPeerId : null);
|
||||
}
|
||||
|
||||
void _broadcastIfWaitingOnChanged(Set<String> gating) {
|
||||
final last = _lastBroadcast;
|
||||
if (last == null) return;
|
||||
final current = gating.toList()..sort();
|
||||
if (current.length == last.waitingOn.length && last.waitingOn.toSet().containsAll(current)) return;
|
||||
_broadcast();
|
||||
}
|
||||
|
||||
void _setPhase(PlaybackPhase phase) {
|
||||
if (_phase == phase) return;
|
||||
_phase = phase;
|
||||
_callbacks.onPhaseChanged?.call(phase);
|
||||
_restartHeartbeat();
|
||||
}
|
||||
|
||||
void _broadcast({PlaybackActionHint? hint, String? actor, String? toPeerId, int? anchorPositionOverrideMs}) {
|
||||
if (_disposed || !hasActiveEpoch) return;
|
||||
|
||||
final player = _player;
|
||||
int anchorPositionMs;
|
||||
int anchorHostTimeMs;
|
||||
if (_pendingStartAtMs != null && _phase == PlaybackPhase.playing) {
|
||||
anchorPositionMs = _pendingStartPositionMs ?? player?.position.inMilliseconds ?? 0;
|
||||
anchorHostTimeMs = _pendingStartAtMs!;
|
||||
} else {
|
||||
anchorPositionMs = anchorPositionOverrideMs ?? player?.position.inMilliseconds ?? 0;
|
||||
anchorHostTimeMs = _nowMs();
|
||||
}
|
||||
|
||||
final waitingOn = _phase == PlaybackPhase.waitingForPeers ? (_gatingPeers().toList()..sort()) : const <String>[];
|
||||
|
||||
final state = PlaybackState(
|
||||
seq: ++_seq,
|
||||
ratingKey: _ratingKey!,
|
||||
serverId: _serverId!,
|
||||
mediaTitle: _mediaTitle,
|
||||
phase: _phase,
|
||||
anchorPositionMs: anchorPositionMs,
|
||||
anchorHostTimeMs: anchorHostTimeMs,
|
||||
rate: _rate,
|
||||
controlMode: _controlMode,
|
||||
waitingOn: waitingOn,
|
||||
actorPeerId: actor,
|
||||
actionHint: hint,
|
||||
);
|
||||
|
||||
if (toPeerId == null) {
|
||||
final previousWaiting = _lastBroadcast?.waitingOn ?? const [];
|
||||
_lastBroadcast = state;
|
||||
if (!_listEquals(previousWaiting, waitingOn)) {
|
||||
_callbacks.onWaitingOnChanged?.call(waitingOn);
|
||||
}
|
||||
}
|
||||
_sendState(state, toPeerId: toPeerId);
|
||||
}
|
||||
|
||||
static bool _listEquals(List<String> a, List<String> b) {
|
||||
if (a.length != b.length) return false;
|
||||
for (var i = 0; i < a.length; i++) {
|
||||
if (a[i] != b[i]) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,374 @@
|
||||
import 'dart:async';
|
||||
|
||||
import '../../mpv/mpv.dart';
|
||||
import '../../utils/app_logger.dart';
|
||||
import '../models/playback_state.dart';
|
||||
import '../models/sync_message.dart';
|
||||
import '../models/watch_session.dart';
|
||||
import 'attached_player.dart';
|
||||
import 'clock_sync.dart';
|
||||
import 'guest_playback_reconciler.dart';
|
||||
import 'host_playback_coordinator.dart';
|
||||
import 'watch_together_peer_service.dart';
|
||||
|
||||
/// Session-scoped playback-sync controller.
|
||||
///
|
||||
/// Lives for the whole Watch Together session (created at create/join, not
|
||||
/// at player attach), so no sync message is ever dropped during episode
|
||||
/// switches or other attach gaps — the player attachment is just an output
|
||||
/// binding the role engine reconciles against.
|
||||
///
|
||||
/// Routes the v2 protocol between the relay and the role engine:
|
||||
/// host → [HostPlaybackCoordinator] (single writer of [PlaybackState]),
|
||||
/// guest → [GuestPlaybackReconciler] (+ [ClockSync] against the host).
|
||||
class WatchTogetherController {
|
||||
WatchTogetherController({
|
||||
required WatchTogetherPeerService peerService,
|
||||
required WatchSession session,
|
||||
int Function()? nowMs,
|
||||
}) : _peerService = peerService,
|
||||
_session = session,
|
||||
_nowMs = nowMs ?? _systemNowMs {
|
||||
if (session.isHost) {
|
||||
_coordinator = HostPlaybackCoordinator(
|
||||
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),
|
||||
),
|
||||
nowMs: _nowMs,
|
||||
);
|
||||
} else {
|
||||
_clockSync = ClockSync(sendPing: _sendClockPing, nowMs: _nowMs);
|
||||
_reconciler = GuestPlaybackReconciler(
|
||||
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),
|
||||
),
|
||||
nowMs: _nowMs,
|
||||
);
|
||||
_clockSync!.start();
|
||||
}
|
||||
|
||||
_subscriptions.add(peerService.onMessageReceived.listen(_enqueueMessage));
|
||||
_subscriptions.add(peerService.onPeerDisconnected.listen(_handlePeerDisconnected));
|
||||
}
|
||||
|
||||
static int _systemNowMs() => DateTime.now().millisecondsSinceEpoch;
|
||||
|
||||
final WatchTogetherPeerService _peerService;
|
||||
final int Function() _nowMs;
|
||||
WatchSession _session;
|
||||
|
||||
HostPlaybackCoordinator? _coordinator;
|
||||
GuestPlaybackReconciler? _reconciler;
|
||||
ClockSync? _clockSync;
|
||||
|
||||
AttachedPlayer? _attachedPlayer;
|
||||
final List<StreamSubscription<dynamic>> _subscriptions = [];
|
||||
Future<void> _messageQueue = Future.value();
|
||||
bool _disposed = false;
|
||||
|
||||
/// Protocol versions learned from join messages (absent ⇒ v1).
|
||||
final Map<String, int> _peerVersions = {};
|
||||
final Set<String> _updateToastShown = {};
|
||||
|
||||
// Provider-facing callbacks.
|
||||
void Function(PlaybackPhase phase)? onPhaseChanged;
|
||||
void Function(List<String> peerIds)? onWaitingOnChanged;
|
||||
void Function(bool correcting)? onCorrectingChanged;
|
||||
void Function(ControlMode mode)? onControlModeReceived;
|
||||
void Function(String ratingKey, String serverId, String? mediaTitle)? onMediaStateReceived;
|
||||
void Function(String peerId, PlaybackActionHint hint)? onRemoteAction;
|
||||
void Function(String peerId)? onPeerNeedsUpdate;
|
||||
void Function(List<String> peerIds)? onResumedWithout;
|
||||
|
||||
bool get hasPlayer => _attachedPlayer != null;
|
||||
|
||||
PlaybackPhase? get phase => _session.isHost ? _coordinator?.phase : _reconciler?.latestState?.phase;
|
||||
|
||||
/// Update the session (e.g. when the control mode changes).
|
||||
void updateSession(WatchSession session) {
|
||||
_session = session;
|
||||
_coordinator?.updateControlMode(session.controlMode);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Player attachment
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
/// Attach the local player for [ratingKey]/[serverId].
|
||||
///
|
||||
/// [hasFirstFrame] is the screen's first-frame snapshot; [startupHold]
|
||||
/// delays readiness until platform startup gates (frame-rate switch)
|
||||
/// release; [remoteSeek] routes sync seeks through the screen's seek path
|
||||
/// (Plex transcode restarts).
|
||||
void attachPlayer(
|
||||
Player player, {
|
||||
required String ratingKey,
|
||||
required String serverId,
|
||||
String? mediaTitle,
|
||||
bool hasFirstFrame = false,
|
||||
Future<void>? startupHold,
|
||||
Future<void> Function(Duration target)? remoteSeek,
|
||||
}) {
|
||||
detachPlayer();
|
||||
|
||||
final attached = AttachedPlayer(
|
||||
player: player,
|
||||
onLost: () {
|
||||
appLogger.w('WatchTogether: Player attachment lost, detaching from sync');
|
||||
detachPlayer();
|
||||
},
|
||||
remoteSeek: remoteSeek,
|
||||
nowMs: _nowMs,
|
||||
);
|
||||
_attachedPlayer = attached;
|
||||
|
||||
if (_session.isHost) {
|
||||
_coordinator!.attach(
|
||||
attached,
|
||||
ratingKey: ratingKey,
|
||||
serverId: serverId,
|
||||
mediaTitle: mediaTitle,
|
||||
hasFirstFrame: hasFirstFrame,
|
||||
startupHold: startupHold,
|
||||
);
|
||||
} else {
|
||||
_reconciler!.attach(
|
||||
attached,
|
||||
ratingKey: ratingKey,
|
||||
serverId: serverId,
|
||||
hasFirstFrame: hasFirstFrame,
|
||||
startupHold: startupHold,
|
||||
);
|
||||
}
|
||||
appLogger.d('WatchTogether: Player attached (host: ${_session.isHost})');
|
||||
}
|
||||
|
||||
/// Detach the player. [exiting] means the user left the video player (the
|
||||
/// epoch ends); an episode switch keeps the session and epoch flow.
|
||||
void detachPlayer({bool exiting = false}) {
|
||||
final attached = _attachedPlayer;
|
||||
if (attached == null) return;
|
||||
_attachedPlayer = null;
|
||||
_coordinator?.detachPlayer(exiting: exiting);
|
||||
_reconciler?.detachPlayer();
|
||||
unawaited(attached.dispose());
|
||||
appLogger.d('WatchTogether: Player detached (exiting: $exiting)');
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Provider inputs
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
/// Host switched media (also called right after attach with the same key,
|
||||
/// which is a no-op).
|
||||
void setCurrentMedia({required String ratingKey, required String serverId, String? mediaTitle}) {
|
||||
_coordinator?.setLocalMedia(ratingKey: ratingKey, serverId: serverId, mediaTitle: mediaTitle);
|
||||
}
|
||||
|
||||
/// User seek executed locally (screen hook).
|
||||
void onLocalSeek(Duration position) {
|
||||
if (_session.isHost) {
|
||||
_coordinator?.onLocalSeekIntent(position);
|
||||
} else {
|
||||
_reconciler?.onLocalSeekIntent(position);
|
||||
}
|
||||
}
|
||||
|
||||
void setBackgrounded(bool value) {
|
||||
_coordinator?.setBackgrounded(value);
|
||||
_reconciler?.setBackgrounded(value);
|
||||
}
|
||||
|
||||
void announceJoin(String displayName) {
|
||||
final peerId = _peerService.myPeerId;
|
||||
if (peerId == null) return;
|
||||
_peerService.broadcast(SyncMessage.join(peerId: peerId, displayName: displayName, isHost: _session.isHost));
|
||||
}
|
||||
|
||||
void announceLeave() {
|
||||
final peerId = _peerService.myPeerId;
|
||||
if (peerId == null) return;
|
||||
_peerService.broadcast(SyncMessage.leave(peerId: peerId));
|
||||
}
|
||||
|
||||
/// Ask the host to (re-)send its current state.
|
||||
void requestState() {
|
||||
if (_session.isHost) return;
|
||||
final request = SyncMessage.requestState(peerId: _peerService.myPeerId);
|
||||
final hostPeerId = _session.hostPeerId;
|
||||
if (hostPeerId != null) {
|
||||
_peerService.sendTo(hostPeerId, request);
|
||||
} else {
|
||||
_peerService.broadcast(request);
|
||||
}
|
||||
}
|
||||
|
||||
/// Relay reconnect completed: re-establish mutual state.
|
||||
void onReconnected() {
|
||||
if (_session.isHost) {
|
||||
_coordinator?.onReconnected();
|
||||
} else {
|
||||
_reconciler?.onReconnected();
|
||||
requestState();
|
||||
}
|
||||
}
|
||||
|
||||
void dispose() {
|
||||
_disposed = true;
|
||||
detachPlayer(exiting: true);
|
||||
for (final subscription in _subscriptions) {
|
||||
unawaited(subscription.cancel());
|
||||
}
|
||||
_subscriptions.clear();
|
||||
_clockSync?.stop();
|
||||
_coordinator?.dispose();
|
||||
_reconciler?.dispose();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
// Transport plumbing
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
void _sendState(PlaybackState state, {String? toPeerId}) {
|
||||
final message = SyncMessage.state(state, peerId: _peerService.myPeerId);
|
||||
if (toPeerId != null) {
|
||||
_peerService.sendTo(toPeerId, message);
|
||||
} else {
|
||||
_peerService.broadcast(message);
|
||||
}
|
||||
}
|
||||
|
||||
void _sendToHost(SyncMessage message) {
|
||||
final hostPeerId = _session.hostPeerId;
|
||||
if (hostPeerId != null) {
|
||||
_peerService.sendTo(hostPeerId, message);
|
||||
} else {
|
||||
_peerService.broadcast(message);
|
||||
}
|
||||
}
|
||||
|
||||
void _sendClockPing(int pingId) {
|
||||
_sendToHost(SyncMessage.ping(pingId, peerId: _peerService.myPeerId));
|
||||
}
|
||||
|
||||
void _enqueueMessage(SyncMessage message) {
|
||||
_messageQueue = _messageQueue.then((_) => _handleMessage(message)).catchError((
|
||||
Object error,
|
||||
StackTrace stackTrace,
|
||||
) {
|
||||
appLogger.e('WatchTogether: Failed to handle ${message.type.name} message', error: error, stackTrace: stackTrace);
|
||||
});
|
||||
}
|
||||
|
||||
Future<void> _handleMessage(SyncMessage message) async {
|
||||
if (_disposed) return;
|
||||
final senderId = message.peerId;
|
||||
if (senderId == null || senderId == _peerService.myPeerId) return;
|
||||
|
||||
switch (message.type) {
|
||||
case SyncMessageType.state:
|
||||
// Only the host may author room state.
|
||||
if (_session.isHost || senderId != _session.hostPeerId) return;
|
||||
final state = message.state;
|
||||
if (state != null) _reconciler?.onState(state);
|
||||
break;
|
||||
|
||||
case SyncMessageType.status:
|
||||
final status = message.status;
|
||||
if (_session.isHost && status != null) {
|
||||
_coordinator?.onPeerStatus(senderId, status);
|
||||
}
|
||||
break;
|
||||
|
||||
case SyncMessageType.control:
|
||||
if (!_session.isHost) return;
|
||||
// In host-only mode nobody else gets a say.
|
||||
if (_session.controlMode == ControlMode.hostOnly) return;
|
||||
if (_isIncompatible(senderId)) return;
|
||||
final control = message.control;
|
||||
if (control != null) _coordinator?.onControlRequest(senderId, control);
|
||||
break;
|
||||
|
||||
case SyncMessageType.requestState:
|
||||
if (_session.isHost) _coordinator?.onStateRequested(senderId);
|
||||
break;
|
||||
|
||||
case SyncMessageType.ping:
|
||||
if (message.pingId != null) {
|
||||
// The pong timestamp is "host clock now" for the guest's offset
|
||||
// math — it must come from the same clock as the state anchors.
|
||||
_peerService.sendTo(
|
||||
senderId,
|
||||
SyncMessage(
|
||||
type: SyncMessageType.pong,
|
||||
timestamp: _nowMs(),
|
||||
pingId: message.pingId,
|
||||
peerId: _peerService.myPeerId,
|
||||
),
|
||||
);
|
||||
}
|
||||
break;
|
||||
|
||||
case SyncMessageType.pong:
|
||||
if (message.pingId != null && !_session.isHost) {
|
||||
_clockSync?.onPong(message.pingId!, message.timestamp);
|
||||
}
|
||||
break;
|
||||
|
||||
case SyncMessageType.join:
|
||||
_handleJoin(senderId, message);
|
||||
break;
|
||||
|
||||
case SyncMessageType.leave:
|
||||
_peerVersions.remove(senderId);
|
||||
_coordinator?.onPeerLeft(senderId);
|
||||
break;
|
||||
|
||||
case SyncMessageType.hostExitedPlayer:
|
||||
// Handled at the provider level.
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
bool _isIncompatible(String peerId) => (_peerVersions[peerId] ?? 1) != SyncMessage.protocolVersion;
|
||||
|
||||
void _handleJoin(String senderId, SyncMessage message) {
|
||||
final version = message.version ?? 1;
|
||||
final firstSighting = !_peerVersions.containsKey(senderId);
|
||||
_peerVersions[senderId] = version;
|
||||
final compatible = version == SyncMessage.protocolVersion;
|
||||
|
||||
if (!compatible && _updateToastShown.add(senderId)) {
|
||||
appLogger.w('WatchTogether: Peer $senderId speaks protocol v$version (ours: ${SyncMessage.protocolVersion})');
|
||||
onPeerNeedsUpdate?.call(senderId);
|
||||
}
|
||||
|
||||
if (_session.isHost) {
|
||||
_coordinator?.onPeerJoined(senderId, compatible: compatible);
|
||||
} else if (senderId == _session.hostPeerId && firstSighting) {
|
||||
// A fresh host join can mean a restarted host app with a reset
|
||||
// sequence counter — accept its numbering from scratch.
|
||||
_reconciler?.resetSequence();
|
||||
}
|
||||
}
|
||||
|
||||
void _handlePeerDisconnected(String peerId) {
|
||||
_peerVersions.remove(peerId);
|
||||
_updateToastShown.remove(peerId);
|
||||
_coordinator?.onPeerLeft(peerId);
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,10 +1,11 @@
|
||||
// Models
|
||||
export 'models/watch_session.dart';
|
||||
export 'models/sync_message.dart';
|
||||
export 'models/playback_state.dart';
|
||||
|
||||
// Services
|
||||
export 'services/watch_together_peer_service.dart';
|
||||
export 'services/watch_together_sync_manager.dart';
|
||||
export 'services/watch_together_controller.dart';
|
||||
|
||||
// Providers
|
||||
export 'providers/watch_together_provider.dart';
|
||||
|
||||
@@ -338,6 +338,8 @@ class _ParticipantNotificationOverlayState extends State<ParticipantNotification
|
||||
ParticipantEventType.resumed => t.watchTogether.participantResumed(name: n.event.displayName),
|
||||
ParticipantEventType.seeked => t.watchTogether.participantSeeked(name: n.event.displayName),
|
||||
ParticipantEventType.buffering => t.watchTogether.participantBuffering(name: n.event.displayName),
|
||||
ParticipantEventType.needsUpdate => t.watchTogether.participantNeedsUpdate(name: n.event.displayName),
|
||||
ParticipantEventType.resumedWithout => t.watchTogether.resumingWithout(name: n.event.displayName),
|
||||
};
|
||||
return Container(
|
||||
key: ValueKey(n.id),
|
||||
@@ -380,13 +382,20 @@ class SyncingIndicator extends StatelessWidget {
|
||||
class WaitingForParticipantsIndicator extends StatelessWidget {
|
||||
const WaitingForParticipantsIndicator({super.key});
|
||||
|
||||
static String _label(List<String> names) {
|
||||
if (names.isEmpty) return t.watchTogether.waitingForParticipants;
|
||||
final shown = names.length <= 2 ? names.join(', ') : '${names.take(2).join(', ')} +${names.length - 2}';
|
||||
return t.watchTogether.waitingForName(name: shown);
|
||||
}
|
||||
|
||||
@override
|
||||
Widget build(BuildContext context) {
|
||||
return Selector<WatchTogetherProvider, bool>(
|
||||
selector: (_, provider) => provider.isDeferredPlay,
|
||||
builder: (context, isDeferredPlay, child) {
|
||||
if (!isDeferredPlay) return const SizedBox.shrink();
|
||||
return _StatusPill(tvIcon: Symbols.hourglass_empty_rounded, label: t.watchTogether.waitingForParticipants);
|
||||
return Selector<WatchTogetherProvider, (bool, List<String>)>(
|
||||
selector: (_, provider) => (provider.isWaitingForPeers, provider.waitingOnNames),
|
||||
builder: (context, value, child) {
|
||||
final (isWaiting, names) = value;
|
||||
if (!isWaiting) return const SizedBox.shrink();
|
||||
return _StatusPill(tvIcon: Symbols.hourglass_empty_rounded, label: _label(names));
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user