diff --git a/open_wearable/lib/models/connector_settings.dart b/open_wearable/lib/models/connector_settings.dart index e1a45f9e0..ba225d10d 100644 --- a/open_wearable/lib/models/connector_settings.dart +++ b/open_wearable/lib/models/connector_settings.dart @@ -150,7 +150,7 @@ class ConnectorSettings { /// Stops the running server and resets the runtime status. static Future dispose() async { _stopNetworkStatusRefresh(); - await _webSocketServer?.stop(); + await _webSocketServer?.dispose(); _webSocketServer = null; _setRuntimeStatus(const ConnectorRuntimeStatus.disabled()); } diff --git a/open_wearable/lib/models/connectors/websocket_ipc_server.dart b/open_wearable/lib/models/connectors/websocket_ipc_server.dart index 9c04b9e67..c8a87a618 100644 --- a/open_wearable/lib/models/connectors/websocket_ipc_server.dart +++ b/open_wearable/lib/models/connectors/websocket_ipc_server.dart @@ -54,6 +54,16 @@ class WebSocketIpcServer implements CommandRuntime { _wearableConnector = wearableConnector, _audioPlaybackService = audioPlaybackService ?? WebsocketAudioPlaybackService() { + // Keep the inventory current even while the network endpoint is disabled. + _connectSubscription = this.wearableManager.connectStream.listen((wearable) { + _registerConnectedWearable(wearable); + _broadcastEvent( + { + 'event': 'connected', + 'wearable': _serializeWearableSummary(wearable), + }, + ); + }); for (final command in createDefaultIpcCommands(this)) { addCommand(command); } @@ -176,17 +186,23 @@ class WebSocketIpcServer implements CommandRuntime { await _scanSubscription?.cancel(); await _connectingSubscription?.cancel(); - await _connectSubscription?.cancel(); _scanSubscription = null; _connectingSubscription = null; - _connectSubscription = null; _discoveredDevicesById.clear(); - _connectedWearablesById.clear(); _advertisedHost = null; logger.i('[connector.websocket] stopped'); } + /// Release app-lifetime device tracking when this server is discarded. + Future dispose() async { + await stop(); + await _connectSubscription?.cancel(); + _connectSubscription = null; + _connectedWearablesById.clear(); + await _scanEventsController.close(); + } + /// Removes a disconnected client session from the active set. void _onClientClosed(_ClientSession client) { _clients.remove(client); @@ -464,22 +480,17 @@ class WebSocketIpcServer implements CommandRuntime { ); }); - _connectSubscription ??= wearableManager.connectStream.listen((wearable) { - _registerConnectedWearable(wearable); - _broadcastEvent( - { - 'event': 'connected', - 'wearable': _serializeWearableSummary(wearable), - }, - ); - }); + } /// Tracks a connected wearable and removes it when it disconnects. void _registerConnectedWearable(Wearable wearable) { + if (identical(_connectedWearablesById[wearable.deviceId], wearable)) return; _connectedWearablesById[wearable.deviceId] = wearable; wearable.addDisconnectListener(() { - _connectedWearablesById.remove(wearable.deviceId); + if (identical(_connectedWearablesById[wearable.deviceId], wearable)) { + _connectedWearablesById.remove(wearable.deviceId); + } }); } diff --git a/open_wearable/test/models/connectors/websocket_inventory_test.dart b/open_wearable/test/models/connectors/websocket_inventory_test.dart new file mode 100644 index 000000000..da1c177dd --- /dev/null +++ b/open_wearable/test/models/connectors/websocket_inventory_test.dart @@ -0,0 +1,52 @@ +import 'dart:async'; +import 'package:logger/logger.dart'; +import 'package:open_wearable/models/logger.dart'; +import 'package:open_wearable/models/connectors/websocket_audio_playback_service.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:open_earable_flutter/open_earable_flutter.dart'; +import 'package:open_wearable/models/connectors/websocket_ipc_server.dart'; + +class _Manager implements WearableManager { + final connections = StreamController.broadcast(sync: true); + @override + Stream get connectStream => connections.stream; + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + +class _Wearable extends Wearable { + _Wearable(this.deviceId, WearableDisconnectNotifier notifier) + : super(name: 'OpenEarable', disconnectNotifier: notifier); + @override + final String deviceId; + @override + Future disconnect() async {} +} + +class _Audio implements WebsocketAudioPlaybackService { + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + +void main() { + initLogger(Logger(level: Level.off)); + test('tracks connections and disconnects while disabled and across stops', + () async { + final manager = _Manager(); + final server = WebSocketIpcServer(wearableManager: manager, audioPlaybackService: _Audio()); + final oldNotifier = WearableDisconnectNotifier(); + final newNotifier = WearableDisconnectNotifier(); + manager.connections.add(_Wearable('left', oldNotifier)); + expect((await server.listConnected()).single['device_id'], 'left'); + await server.stop(); + expect(await server.listConnected(), hasLength(1)); + manager.connections.add(_Wearable('left', newNotifier)); + oldNotifier.notifyListeners(); + expect(await server.listConnected(), hasLength(1)); + newNotifier.notifyListeners(); + expect(await server.listConnected(), isEmpty); + await server.dispose(); + expect(manager.connections.hasListener, isFalse); + await manager.connections.close(); + }); +}