Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion open_wearable/lib/models/connector_settings.dart
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ class ConnectorSettings {
/// Stops the running server and resets the runtime status.
static Future<void> dispose() async {
_stopNetworkStatusRefresh();
await _webSocketServer?.stop();
await _webSocketServer?.dispose();
_webSocketServer = null;
_setRuntimeStatus(const ConnectorRuntimeStatus.disabled());
}
Expand Down
37 changes: 24 additions & 13 deletions open_wearable/lib/models/connectors/websocket_ipc_server.dart
Original file line number Diff line number Diff line change
Expand Up @@ -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(
<String, dynamic>{
'event': 'connected',
'wearable': _serializeWearableSummary(wearable),
},
);
});
for (final command in createDefaultIpcCommands(this)) {
addCommand(command);
}
Expand Down Expand Up @@ -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<void> 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);
Expand Down Expand Up @@ -464,22 +480,17 @@ class WebSocketIpcServer implements CommandRuntime {
);
});

_connectSubscription ??= wearableManager.connectStream.listen((wearable) {
_registerConnectedWearable(wearable);
_broadcastEvent(
<String, dynamic>{
'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);
}
});
}

Expand Down
52 changes: 52 additions & 0 deletions open_wearable/test/models/connectors/websocket_inventory_test.dart
Original file line number Diff line number Diff line change
@@ -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<Wearable>.broadcast(sync: true);
@override
Stream<Wearable> 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<void> 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();
});
}
Loading