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

fix: watch together server races, reconnect readiness, session end

Browse Source
This commit is contained in:
edde746
2026-04-16 15:24:44 +02:00
parent 1d3ae3e1c5
commit 34815f22ee
6 changed files with 117 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/screens/video_player_screen.dart
+1 -2
View File
@@ -3557,6 +3557,7 @@ class VideoPlayerScreenState extends State<VideoPlayerScreen> with WidgetsBindin
// Watch Together overlays (isolated from video surface repaints)
RepaintBoundary(
child: Stack(
fit: StackFit.expand,
children: [
// Watch Together: reconnecting to host overlay
Selector<WatchTogetherProvider, bool>(
@@ -3597,8 +3598,6 @@ class VideoPlayerScreenState extends State<VideoPlayerScreen> with WidgetsBindin
);
},
),
// Watch Together: session indicator and menu
WatchTogetherOverlay(onLeaveSession: _handleBackButton),
// Watch Together: participant join/leave/buffering notifications
const ParticipantNotificationOverlay(),
// Watch Together: waiting for participants to load
lib/watch_together/providers/watch_together_provider.dart
+26 -9
View File
@@ -175,10 +175,11 @@ class WatchTogetherProvider with ChangeNotifier {
}
}
/// Wire up reconnection handler to re-announce join after reconnect
/// Wire up reconnection handler to re-announce join and readiness after reconnect
void _wireReconnectHandler() {
_peerService!.onReconnected = () {
_syncManager?.announceJoin(_displayName);
_syncManager?.reannounceReadyIfNeeded();
};
}
@@ -308,17 +309,33 @@ class WatchTogetherProvider with ChangeNotifier {
///
/// Returns `true` if the user became the host.
Future<bool> enterRoom(String sessionId, {ControlMode controlMode = ControlMode.anyone, String? displayName}) async {
// Probe the relay with a lightweight peer service to check room occupancy,
// then do a single createSession or joinSession. This avoids the crash-prone
// join→teardown→create cycle on the provider.
final customRelayUrl = SettingsService.instanceOrNull?.getCustomRelayUrl();
final probe = WatchTogetherPeerService(customBaseUrl: customRelayUrl);
bool shouldBeHost;
try {
await joinSession(sessionId, displayName: displayName);
return false; // joined as guest
await probe.joinSession(sessionId);
shouldBeHost = probe.connectedPeers.isEmpty;
} on PeerError catch (e) {
// Room doesn't exist — create it as host
if (e.serverCode == 'room_not_found') {
await createSession(controlMode: controlMode, displayName: displayName, sessionId: sessionId);
return true;
shouldBeHost = true;
} else {
await probe.disconnect();
probe.dispose();
rethrow;
}
rethrow;
}
await probe.disconnect();
probe.dispose();
if (shouldBeHost) {
await createSession(controlMode: controlMode, displayName: displayName, sessionId: sessionId);
} else {
await joinSession(sessionId, displayName: displayName);
}
return shouldBeHost;
}
/// Leave the current session
@@ -490,11 +507,11 @@ class WatchTogetherProvider with ChangeNotifier {
);
}
// If the host deliberately left, end the session immediately
// instead of waiting for the 15s reconnect grace period.
// If the host deliberately left, end the session for everyone.
if (!isHost && message.peerId == _session?.hostPeerId) {
_hostIntentionallyLeft = true;
_handleHostExitedPlayer(message);
leaveSession();
}
notifyListeners();
lib/watch_together/services/watch_together_sync_manager.dart
+28 -2
View File
@@ -125,10 +125,19 @@ class WatchTogetherSyncManager {
// If host, start broadcasting position periodically
if (_session.isHost) {
_startPositionSync();
// Note: sessionConfig is sent after video loads (with correct position) in buffering handler
}
// Note: playerReady will be announced when video loads (first buffering: false)
// If the video is already loaded (buffering stream already fired before we
// subscribed), announce ready now so peers aren't stuck waiting.
if (!player.state.buffering && !_hasAnnouncedReady) {
_hasAnnouncedReady = true;
_peerReady[_peerService.myPeerId!] = true;
_peerService.broadcast(SyncMessage.playerReady(peerId: _peerService.myPeerId!, ready: true));
appLogger.d('WatchTogether: Video already loaded on attach, announcing ready');
if (_session.isHost) {
_sendSessionConfig();
}
}
// If guest, request current session config from host in case we missed
// a mediaSwitch broadcast (e.g., host switched episodes while we were
@@ -144,6 +153,8 @@ class WatchTogetherSyncManager {
/// Initialize participant tracking from existing session participants
/// Call this before attachPlayer() to ensure we know about participants who joined before
void initializeParticipants(List<String> peerIds) {
// Clear stale entries (e.g. host's own peerId left over from a previous detachPlayer)
_peerReady.clear();
for (final peerId in peerIds) {
if (peerId != _peerService.myPeerId) {
if (_session.isHost) {
@@ -232,6 +243,7 @@ class WatchTogetherSyncManager {
return;
}
if (isPlaying && !_firstPlayCompleted) _firstPlayCompleted = true;
if (!isPlaying) _setDeferredPlay(false);
_broadcastPlayPause(isPlaying);
}),
@@ -944,6 +956,20 @@ class WatchTogetherSyncManager {
}
}
/// Re-announce player readiness after reconnect.
///
/// During reconnect the host resets our _peerReady entry to false via
/// _handlePeerJoin, but our _hasAnnouncedReady flag is still true (never
/// reset because the player stays attached). Re-broadcast so the host
/// doesn't stay stuck in the deferred-play gate.
void reannounceReadyIfNeeded() {
if (_hasAnnouncedReady && _peerService.myPeerId != null) {
_peerReady[_peerService.myPeerId!] = true;
_peerService.broadcast(SyncMessage.playerReady(peerId: _peerService.myPeerId!, ready: true));
appLogger.d('WatchTogether: Re-announced player ready after reconnect');
}
}
/// Send join announcement to all peers
void announceJoin(String displayName) {
_peerService.broadcast(
lib/watch_together/widgets/watch_together_overlay.dart
+16 -25
View File
@@ -14,31 +14,24 @@ import '../models/watch_session.dart';
import '../providers/watch_together_provider.dart';
/// Overlay shown on the video player when in a watch together session
class WatchTogetherOverlay extends StatelessWidget {
/// Callback when the user wants to leave the session
/// Session indicator badge for embedding in the video controls header.
/// Shows participant count, host badge, and opens the session menu on tap.
class WatchTogetherSessionIndicator extends StatelessWidget {
final VoidCallback? onLeaveSession;
const WatchTogetherOverlay({super.key, this.onLeaveSession});
const WatchTogetherSessionIndicator({super.key, this.onLeaveSession});
@override
Widget build(BuildContext context) {
return Consumer<WatchTogetherProvider>(
builder: (context, provider, child) {
if (!provider.isInSession) {
return const SizedBox.shrink();
}
return Positioned(
top: 16,
right: 16,
child: _SessionIndicator(
participantCount: provider.participantCount,
isHost: provider.isHost,
isSyncing: provider.isSyncing,
controlMode: provider.controlMode,
sessionId: provider.sessionId,
onTap: () => _showSessionMenu(context, provider),
),
return _SessionIndicator(
participantCount: provider.participantCount,
isHost: provider.isHost,
isSyncing: provider.isSyncing,
controlMode: provider.controlMode,
sessionId: provider.sessionId,
onTap: () => _showSessionMenu(context, provider),
);
},
);
@@ -71,8 +64,6 @@ class _SessionIndicator extends StatelessWidget {
@override
Widget build(BuildContext context) {
final theme = Theme.of(context);
return Material(
color: Colors.black54,
borderRadius: const BorderRadius.all(Radius.circular(20)),
@@ -94,7 +85,7 @@ class _SessionIndicator extends StatelessWidget {
: const CircularProgressIndicator(strokeWidth: 2, color: Colors.white),
)
else
Icon(Symbols.group, size: 18, color: isHost ? theme.colorScheme.primary : Colors.white),
Icon(Symbols.group, size: 18, color: isHost ? Colors.amber : Colors.white),
const SizedBox(width: 6),
@@ -109,13 +100,13 @@ class _SessionIndicator extends StatelessWidget {
const SizedBox(width: 6),
Container(
padding: const EdgeInsets.symmetric(horizontal: 6, vertical: 2),
decoration: BoxDecoration(
color: theme.colorScheme.primary,
borderRadius: const BorderRadius.all(Radius.circular(4)),
decoration: const BoxDecoration(
color: Colors.amber,
borderRadius: BorderRadius.all(Radius.circular(4)),
),
child: Text(
t.watchTogether.hostBadge,
style: const TextStyle(color: Colors.white, fontSize: 10, fontWeight: FontWeight.bold),
style: const TextStyle(color: Colors.black, fontSize: 10, fontWeight: FontWeight.bold),
),
),
],
lib/widgets/video_controls/widgets/video_controls_header.dart
+13
View File
@@ -1,8 +1,11 @@
import 'package:flutter/material.dart';
import 'package:provider/provider.dart';
import 'package:plezy/utils/formatters.dart';
import '../../../models/plex_metadata.dart';
import '../../../i18n/strings.g.dart';
import '../../../watch_together/widgets/watch_together_overlay.dart';
import '../../../watch_together/providers/watch_together_provider.dart';
import '../../app_bar_back_button.dart';
/// Header layout style for video controls
@@ -47,6 +50,16 @@ class VideoControlsHeader extends StatelessWidget {
),
const SizedBox(width: 16),
Expanded(child: style == VideoHeaderStyle.singleLine ? _buildSingleLineTitle() : _buildMultiLineTitle()),
Selector<WatchTogetherProvider, bool>(
selector: (_, p) => p.isInSession,
builder: (context, inSession, child) {
if (!inSession) return const SizedBox.shrink();
return const Padding(
padding: EdgeInsets.only(right: 8),
child: WatchTogetherSessionIndicator(),
);
},
),
?trailing,
],
);
server/main.go
+33 -29
View File
@@ -202,10 +202,11 @@ type serverMsg struct {
type Client struct {
conn *websocket.Conn
send chan []byte
done chan struct{}
}
func newClient(conn *websocket.Conn) *Client {
c := &Client{conn: conn, send: make(chan []byte, 64)}
c := &Client{conn: conn, send: make(chan []byte, 64), done: make(chan struct{})}
go c.writePump()
return c
}
@@ -215,13 +216,12 @@ func (c *Client) writePump() {
defer ticker.Stop()
for {
select {
case data, ok := <-c.send:
if !ok {
c.conn.WriteMessage(websocket.CloseMessage, nil)
return
}
case data := <-c.send:
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
c.conn.WriteMessage(websocket.TextMessage, data)
case <-c.done:
c.conn.WriteMessage(websocket.CloseMessage, nil)
return
case <-ticker.C:
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
@@ -231,20 +231,24 @@ func (c *Client) writePump() {
}
}
func (c *Client) trySend(data []byte) {
select {
case c.send <- data:
case <-c.done:
default:
}
}
func (c *Client) sendJSON(msg serverMsg) {
data, err := json.Marshal(msg)
if err != nil {
return
}
select {
case c.send <- data:
default:
// Drop if buffer full (slow client)
}
c.trySend(data)
}
func (c *Client) close() {
close(c.send)
close(c.done)
}
// --- Room ---
@@ -281,10 +285,7 @@ func (r *Room) broadcastExcept(senderID string, msg serverMsg) {
r.mu.RUnlock()
for _, client := range targets {
select {
case client.send <- data:
default:
}
client.trySend(data)
}
}
@@ -299,12 +300,8 @@ func (r *Room) sendTo(targetID string, msg serverMsg) bool {
if !ok {
return false
}
select {
case client.send <- data:
return true
default:
return false
}
client.trySend(data)
return true
}
// --- Log store ---
@@ -556,20 +553,27 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
var currentPeerID string
var isHost bool
// Cleanup on disconnect
// Cleanup on disconnect — only if our Client is still the one in the room.
// A reconnecting peer reuses the same peerId, so the map entry may have
// been overwritten by a newer Client before this defer runs.
defer func() {
if currentRoom != nil && currentPeerID != "" {
currentRoom.mu.Lock()
delete(currentRoom.Peers, currentPeerID)
stale := currentRoom.Peers[currentPeerID] != client
if !stale {
delete(currentRoom.Peers, currentPeerID)
}
currentRoom.mu.Unlock()
currentRoom.broadcastExcept(currentPeerID, serverMsg{
Type: "peerLeft",
PeerID: currentPeerID,
})
if !stale {
currentRoom.broadcastExcept(currentPeerID, serverMsg{
Type: "peerLeft",
PeerID: currentPeerID,
})
}
if isHost {
s.conns.releaseRoom(ip)
}
log.Printf("peer %s left room %s", currentPeerID, currentRoom.SessionID)
log.Printf("peer %s left room %s (stale=%v)", currentPeerID, currentRoom.SessionID, stale)
}
}()