diff --git a/open_wearable/lib/models/connectors/commands/subscribe_command.dart b/open_wearable/lib/models/connectors/commands/subscribe_command.dart index 79bc32a27..3e6862021 100644 --- a/open_wearable/lib/models/connectors/commands/subscribe_command.dart +++ b/open_wearable/lib/models/connectors/commands/subscribe_command.dart @@ -1,5 +1,6 @@ import 'package:open_earable_flutter/open_earable_flutter.dart'; +import '../../shared_sensor_configuration.dart'; import 'command.dart'; import 'ipc_internal_param_names.dart'; import 'param_readers.dart'; @@ -59,10 +60,12 @@ class SubscribeCommand extends RuntimeCommand { args: args, ).sensorStream; case 'sensor_configuration': - return _requireCapability( - wearable: wearable, - streamName: streamName, - ).sensorConfigurationStream; + return sharedSensorConfiguration( + _requireCapability( + wearable: wearable, + streamName: streamName, + ), + ); case 'button_events': return _requireCapability( wearable: wearable, diff --git a/open_wearable/lib/models/shared_sensor_configuration.dart b/open_wearable/lib/models/shared_sensor_configuration.dart new file mode 100644 index 000000000..50906e3fc --- /dev/null +++ b/open_wearable/lib/models/shared_sensor_configuration.dart @@ -0,0 +1,74 @@ +import 'dart:async'; +import 'package:open_earable_flutter/open_earable_flutter.dart'; + +typedef _Report = Map; +final _reports = Expando<_SharedReports>(); + +/// One hardware subscription per manager, shared by the UI and connectors. +/// New listeners receive the latest report; the last cancellation releases BLE. +Stream> + sharedSensorConfiguration(SensorConfigurationManager manager) => + (_reports[manager] ??= _SharedReports(manager)).stream; + +class _SharedReports { + _SharedReports(this.manager); + final SensorConfigurationManager manager; + final _listeners = >{}; + StreamSubscription<_Report>? _source; + Future? _cancelling; + bool _starting = false; + _Report? _latest; + + late final stream = Stream<_Report>.multi( + (listener) { + _listeners.add(listener); + if (_latest != null) listener.add(_latest!); + listener.onCancel = () { + _listeners.remove(listener); + if (_listeners.isEmpty) { + _latest = null; + if (_source != null) { + _cancelling = _source!.cancel(); + _source = null; + } + } + }; + _start(); + }, + isBroadcast: true, + ); + + Future _start() async { + if (_source != null || _starting) return; + _starting = true; + try { + // Finish cancellation before a new getter can replace the BLE listener. + if (_cancelling != null) await _cancelling; + if (_listeners.isEmpty) return; + _source = manager.sensorConfigurationStream.listen( + (report) { + _latest = Map.unmodifiable(report); + for (final listener in _listeners.toList()) { + listener.addSync(_latest!); + } + }, + onError: (Object error, StackTrace stack) { + for (final listener in _listeners.toList()) { + listener.addErrorSync(error, stack); + } + }, + onDone: () { + _source = null; + _latest = null; + final listeners = _listeners.toList(); + _listeners.clear(); + for (final listener in listeners) { + listener.closeSync(); + } + }, + ); + } finally { + _starting = false; + } + } +} diff --git a/open_wearable/lib/view_models/sensor_configuration_provider.dart b/open_wearable/lib/view_models/sensor_configuration_provider.dart index cdaf3f226..8c9224269 100644 --- a/open_wearable/lib/view_models/sensor_configuration_provider.dart +++ b/open_wearable/lib/view_models/sensor_configuration_provider.dart @@ -4,6 +4,7 @@ import 'package:flutter/foundation.dart'; import 'package:open_earable_flutter/open_earable_flutter.dart' hide logger; import '../models/logger.dart'; +import '../models/shared_sensor_configuration.dart'; /// Summary of a profile/configuration restore attempt. class SensorConfigurationRestoreResult { @@ -54,7 +55,7 @@ class SensorConfigurationProvider with ChangeNotifier { required SensorConfigurationManager sensorConfigurationManager, }) : _sensorConfigurationManager = sensorConfigurationManager { _sensorConfigurationSubscription = - _sensorConfigurationManager.sensorConfigurationStream.listen((event) { + sharedSensorConfiguration(_sensorConfigurationManager).listen((event) { final hadReceivedConfigurationReport = _hasReceivedConfigurationReport; final previousReportedConfigurations = Map.of( diff --git a/open_wearable/test/models/shared_sensor_configuration_test.dart b/open_wearable/test/models/shared_sensor_configuration_test.dart new file mode 100644 index 000000000..86ff8cee8 --- /dev/null +++ b/open_wearable/test/models/shared_sensor_configuration_test.dart @@ -0,0 +1,66 @@ +import 'dart:async'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:open_earable_flutter/open_earable_flutter.dart'; +import 'package:open_wearable/models/shared_sensor_configuration.dart'; + +typedef Report = Map; + +class _Manager implements SensorConfigurationManager { + int reads = 0; + late StreamController source; + @override + List get sensorConfigurations => []; + @override + Stream get sensorConfigurationStream { + reads++; + source = StreamController(sync: true); + return source.stream; + } +} + +void main() { + test('UI and connectors share reports, replay, and independent cancellation', + () async { + final manager = _Manager(); + final ui = []; + final remote = []; + final a = sharedSensorConfiguration(manager).listen(ui.add); + manager.source.add({}); + final b = sharedSensorConfiguration(manager).listen(remote.add); + await Future.delayed(Duration.zero); + expect(manager.reads, 1); + expect(ui, hasLength(1)); + expect(remote, hasLength(1)); + manager.source.add({}); + expect(ui, hasLength(2)); + expect(remote, hasLength(2)); + await b.cancel(); + expect(manager.source.hasListener, isTrue); + manager.source.add({}); + expect(ui, hasLength(3)); + await a.cancel(); + expect(manager.source.hasListener, isFalse); + final c = sharedSensorConfiguration(manager).listen(remote.add); + await Future.delayed(Duration.zero); + expect(manager.reads, 2); + expect(remote, hasLength(2)); // no stale replay after all listeners left + manager.source.add({}); + expect(remote, hasLength(3)); + await c.cancel(); + await manager.source.close(); + }); + test('hardware errors reach both consumers and completion closes both', + () async { + final manager = _Manager(); + final errors = []; + var done = 0; + sharedSensorConfiguration(manager) + .listen((_) {}, onError: errors.add, onDone: () => done++); + sharedSensorConfiguration(manager) + .listen((_) {}, onError: errors.add, onDone: () => done++); + manager.source.addError(StateError('BLE disconnected')); + expect(errors, hasLength(2)); + await manager.source.close(); + expect(done, 2); + }); +}