// `controller` can't be a private-named parameter, so an initializing formal // isn't possible for `_controller` (same reason as playback_engine.dart). // ignore_for_file: prefer_initializing_formals import 'dart:async'; import 'dart:io'; import '../playback/playback_engine.dart'; import 'command_dispatch.dart'; import 'messages.dart'; /// Turns the local device into a controllable player: runs an in-process /// WebSocket server, authenticates remotes against the shared server password, /// streams playback snapshots to them, and applies the commands they send to /// the local [PlaybackController]. /// /// Lifecycle is owned by `remote_providers.dart`, which starts the host while /// this device is playable and advertises its [port] over mDNS /// (`discovery.dart`). The server binds an ephemeral port on all IPv4 /// interfaces; discovery carries the port to remotes. class RemoteHost { RemoteHost({ required PlaybackController controller, required this.serverKey, required this.password, required this.deviceName, }) : _controller = controller; final PlaybackController _controller; /// Only remotes reporting this same [serverKey] (i.e. signed into the same /// Subsonic server) are allowed to connect — song ids must resolve on both. final String serverKey; /// Shared secret for the HMAC handshake — never sent, only proven. final String password; /// Friendly name shown to remotes ("Living Room iPad"). final String deviceName; HttpServer? _server; void Function()? _removeListener; final Set _clients = {}; final StreamController _clientCount = StreamController.broadcast(); /// Monotonic snapshot sequence so a remote can drop stale/out-of-order state. int _seq = 0; /// The last state actually put on the wire, for cheap change detection. PlaybackState? _sentState; Timer? _throttle; PlaybackState? _pending; /// Bound port once [start] has run, else null. int? get port => _server?.port; /// Emits the number of connected remotes (drives the "N remotes" UI). Stream get clientCountStream => _clientCount.stream; int get clientCount => _clients.length; /// Bind and begin serving. Returns the chosen ephemeral port. Idempotent — /// a second call returns the already-bound port. Future start() async { final existing = _server; if (existing != null) return existing.port; final server = await HttpServer.bind(InternetAddress.anyIPv4, 0); _server = server; server.listen((req) async { if (!WebSocketTransformer.isUpgradeRequest(req)) { req.response ..statusCode = HttpStatus.forbidden ..close(); return; } try { final ws = await WebSocketTransformer.upgrade(req); _handleSocket(ws); } catch (_) { // Failed upgrade — nothing to clean up. } }); // fireImmediately:false — the per-client welcome snapshot covers the // initial state; we only want deltas here. _removeListener = _controller.addListener(_onStateChanged, fireImmediately: false); return server.port; } /// Stop serving and drop every remote. Safe to call when not started. Future stop() async { _removeListener?.call(); _removeListener = null; _throttle?.cancel(); _throttle = null; _pending = null; _sentState = null; for (final c in _clients.toList()) { await c.close().catchError((_) {}); } _clients.clear(); if (!_clientCount.isClosed) _clientCount.add(0); await _server?.close(force: true); _server = null; } Future dispose() async { await stop(); await _clientCount.close(); } // ---- Handshake + per-client loop -------------------------------------- void _handleSocket(WebSocket socket) { // Keepalive: ping idle clients so a controller that vanished without a // close frame (crashed, walked out of Wi-Fi range) is detected and pruned // instead of lingering as a zombie in `_clients`. socket.pingInterval = const Duration(seconds: 15); final nonce = RemoteAuth.newNonce(); var authed = false; // Drop a peer that connects but never authenticates. final authTimeout = Timer(const Duration(seconds: 10), () { if (!authed) socket.close().catchError((_) {}); }); _send( socket, ChallengeMessage( version: kRemoteProtocolVersion, serverKey: serverKey, nonce: nonce, device: deviceName, ), ); socket.listen( (data) async { RemoteMessage msg; try { msg = RemoteMessage.decode(data as String); } catch (_) { return; // ignore a malformed frame } if (!authed) { final denyReason = _rejectReason(msg, nonce); if (denyReason != null) { _send(socket, DenyMessage(reason: denyReason)); await socket.close().catchError((_) {}); return; } authed = true; authTimeout.cancel(); _send(socket, WelcomeMessage(device: deviceName)); _clients.add(socket); if (!_clientCount.isClosed) _clientCount.add(_clients.length); // Immediate full snapshot so the remote paints without waiting for // the next state change. _send(socket, SnapshotMessage(_snapshotOf(_controller.currentState))); return; } if (msg is CommandMessage) { try { await applyRemoteCommand(_controller, msg.command); } catch (_) { // A command that fails locally (e.g. offline host) surfaces via the // next snapshot's error field — no need to tear down the client. } } }, onDone: () { authTimeout.cancel(); _removeClient(socket); }, onError: (_) { authTimeout.cancel(); _removeClient(socket); }, cancelOnError: true, ); } /// Null if the auth message is acceptable, else a human-readable deny reason. String? _rejectReason(RemoteMessage msg, String nonce) { if (msg is! AuthMessage) return 'expected auth'; if (msg.version != kRemoteProtocolVersion) return 'protocol version'; if (msg.serverKey != serverKey) return 'different server'; if (!RemoteAuth.verify( password: password, nonce: nonce, proof: msg.proof, )) { return 'authentication failed'; } return null; } void _removeClient(WebSocket socket) { if (_clients.remove(socket) && !_clientCount.isClosed) { _clientCount.add(_clients.length); } } // ---- Broadcast -------------------------------------------------------- /// Send discrete changes (play/pause, track, queue, shuffle/loop, error) /// immediately; coalesce position-only ticks to at most one per second (the /// remote interpolates position between snapshots). void _onStateChanged(PlaybackState s) { if (_clients.isEmpty) { _sentState = s; return; } final prev = _sentState; if (prev == null || _discreteChanged(prev, s)) { _flush(s); } else { _pending = s; _throttle ??= Timer(const Duration(seconds: 1), () { _throttle = null; final p = _pending; _pending = null; if (p != null) _flush(p); }); } } void _flush(PlaybackState s) { _throttle?.cancel(); _throttle = null; _pending = null; _sentState = s; final frame = SnapshotMessage(_snapshotOf(s)).encode(); for (final c in _clients.toList()) { try { c.add(frame); } catch (_) { _removeClient(c); } } } static bool _discreteChanged(PlaybackState a, PlaybackState b) => a.playing != b.playing || a.currentIndex != b.currentIndex || a.shuffle != b.shuffle || a.loop != b.loop || a.error != b.error || // copyWith reuses the same queue list unless it actually changed, so an // identity check cheaply distinguishes a queue edit from a position tick. !identical(a.queue, b.queue); StateSnapshot _snapshotOf(PlaybackState s) => StateSnapshot( queue: s.queue, currentIndex: s.currentIndex, playing: s.playing, positionMs: s.position.inMilliseconds, durationMs: s.effectiveDuration.inMilliseconds, shuffle: s.shuffle, loop: s.loop.name, seq: ++_seq, ); void _send(WebSocket socket, RemoteMessage msg) { try { socket.add(msg.encode()); } catch (_) { // Socket already gone; the listen callbacks will clean it up. } } }