diff --git a/lib/services/companion_remote/lan_discovery_service.dart b/lib/services/companion_remote/lan_discovery_service.dart index 75c7965a..fe9841ba 100644 --- a/lib/services/companion_remote/lan_discovery_service.dart +++ b/lib/services/companion_remote/lan_discovery_service.dart @@ -161,7 +161,7 @@ class LanDiscoveryService { Future stopBroadcasting() async { _broadcastTimer?.cancel(); _broadcastTimer = null; - _broadcastSockets?.close(); + await _broadcastSockets?.close(); _broadcastSockets = null; appLogger.d('LanDiscovery: Broadcasting stopped'); } @@ -222,14 +222,10 @@ class LanDiscoveryService { appLogger.d('LanDiscovery: Listening on port $discoveryPort'); - _listenSubscription = _listenSocket!.listen((RawSocketEvent event) { - if (event == RawSocketEvent.read) { - final datagram = _listenSocket?.receive(); - if (datagram != null) { - _handleDatagram(datagram, contexts); - } - } - }); + _listenSubscription = _listenSocket!.listenDatagrams( + (datagram) => _handleDatagram(datagram, contexts), + debugLabel: 'LanDiscovery listener', + ); } catch (e) { appLogger.e('LanDiscovery: Failed to bind listener', error: e); } diff --git a/lib/services/jellyfin_lan_discovery_service.dart b/lib/services/jellyfin_lan_discovery_service.dart index 09410a10..7ff2644b 100644 --- a/lib/services/jellyfin_lan_discovery_service.dart +++ b/lib/services/jellyfin_lan_discovery_service.dart @@ -25,23 +25,14 @@ class JellyfinLanDiscoveryService { InternetAddress? broadcastAddress, }) async { UdpBroadcastSocketSet? socketSet; - final subscriptions = >[]; final discovered = {}; try { socketSet = await UdpBroadcastSockets.bind(); - for (final socket in socketSet.sockets) { - subscriptions.add( - socket.listen((event) { - if (event != RawSocketEvent.read) return; - Datagram? datagram; - while ((datagram = socket.receive()) != null) { - final server = parseDiscoveryResponse(datagram!.data); - if (server == null) continue; - discovered.putIfAbsent(server.id, () => server); - } - }), - ); - } + socketSet.listen((datagram) { + final server = parseDiscoveryResponse(datagram.data); + if (server == null) return; + discovered.putIfAbsent(server.id, () => server); + }, debugLabel: 'Jellyfin LAN discovery'); final data = utf8.encode(discoveryMessage); final target = broadcastAddress ?? UdpBroadcastSockets.limitedBroadcastAddress; @@ -52,10 +43,7 @@ class JellyfinLanDiscoveryService { } catch (e, st) { appLogger.w('Jellyfin LAN discovery failed', error: e, stackTrace: st); } finally { - for (final subscription in subscriptions) { - await subscription.cancel(); - } - socketSet?.close(); + await socketSet?.close(); } return sortDiscoveredServers(discovered.values); diff --git a/lib/utils/udp_broadcast_sockets.dart b/lib/utils/udp_broadcast_sockets.dart index 0e951217..17dbc28a 100644 --- a/lib/utils/udp_broadcast_sockets.dart +++ b/lib/utils/udp_broadcast_sockets.dart @@ -5,13 +5,20 @@ import 'app_logger.dart'; class UdpBroadcastSocketSet { final List _sockets; + final List> _subscriptions = []; - const UdpBroadcastSocketSet._(this._sockets); + UdpBroadcastSocketSet._(this._sockets); bool get isEmpty => _sockets.isEmpty; Iterable get sockets => _sockets; + void listen(void Function(Datagram datagram) onDatagram, {required String debugLabel}) { + for (final socket in _sockets) { + _subscriptions.add(socket.listenDatagrams(onDatagram, debugLabel: debugLabel)); + } + } + void send(List data, InternetAddress address, int port) { for (final socket in _sockets) { try { @@ -22,13 +29,42 @@ class UdpBroadcastSocketSet { } } - void close() { + Future close() async { + final subscriptions = List>.of(_subscriptions); + _subscriptions.clear(); + for (final subscription in subscriptions) { + try { + await subscription.cancel(); + } catch (e, st) { + appLogger.w('UDP datagram subscription cancel failed', error: e, stackTrace: st); + } + } for (final socket in _sockets) { socket.close(); } } } +extension DatagramSocketListen on RawDatagramSocket { + StreamSubscription listenDatagrams( + void Function(Datagram datagram) onDatagram, { + required String debugLabel, + }) { + return listen( + (event) { + if (event != RawSocketEvent.read) return; + Datagram? datagram; + while ((datagram = receive()) != null) { + onDatagram(datagram!); + } + }, + onError: (Object e, StackTrace st) { + appLogger.w('$debugLabel datagram socket error', error: e, stackTrace: st); + }, + ); + } +} + class UdpBroadcastSockets { UdpBroadcastSockets._(); diff --git a/test/services/jellyfin_lan_discovery_service_test.dart b/test/services/jellyfin_lan_discovery_service_test.dart index 008376f1..2eecafe3 100644 --- a/test/services/jellyfin_lan_discovery_service_test.dart +++ b/test/services/jellyfin_lan_discovery_service_test.dart @@ -1,7 +1,9 @@ import 'dart:convert'; +import 'dart:io'; import 'package:flutter_test/flutter_test.dart'; import 'package:plezy/services/jellyfin_lan_discovery_service.dart'; +import 'package:plezy/utils/udp_broadcast_sockets.dart'; void main() { group('JellyfinLanDiscoveryService', () { @@ -41,5 +43,43 @@ void main() { expect(sorted.map((server) => server.id), ['srv-1', 'srv-2', 'srv-3']); }); + + test('listenDatagrams receives queued loopback datagrams', () async { + final receiver = await RawDatagramSocket.bind(InternetAddress.loopbackIPv4, 0); + final sender = await RawDatagramSocket.bind(InternetAddress.loopbackIPv4, 0); + final received = []; + final subscription = receiver.listenDatagrams( + (datagram) => received.add(utf8.decode(datagram.data)), + debugLabel: 'JellyfinLanDiscoveryService test', + ); + + try { + sender.send(utf8.encode('one'), InternetAddress.loopbackIPv4, receiver.port); + sender.send(utf8.encode('two'), InternetAddress.loopbackIPv4, receiver.port); + + await _waitFor(() => received.length >= 2); + + expect(received, containsAll(['one', 'two'])); + } finally { + await subscription.cancel(); + receiver.close(); + sender.close(); + } + }); + + test('UdpBroadcastSocketSet close cancels owned datagram listeners', () async { + final socketSet = await UdpBroadcastSockets.bind(); + socketSet.listen((_) {}, debugLabel: 'JellyfinLanDiscoveryService test'); + + await expectLater(socketSet.close(), completes); + }); }); } + +Future _waitFor(bool Function() condition) async { + for (var i = 0; i < 50; i++) { + if (condition()) return; + await Future.delayed(const Duration(milliseconds: 20)); + } + fail('Timed out waiting for condition'); +}