From 46268691a6147f8c1c9d5d93c9ebd9b58c5c5c8f Mon Sep 17 00:00:00 2001 From: TobiasRoeddiger Date: Fri, 2 Oct 2026 11:45:16 +0200 Subject: [PATCH 1/2] fix(sensors): share configuration reports between UI and connector --- .../commands/subscribe_command.dart | 11 ++-- .../models/shared_sensor_configuration.dart | 66 +++++++++++++++++++ .../sensor_configuration_provider.dart | 3 +- .../shared_sensor_configuration_test.dart | 66 +++++++++++++++++++ 4 files changed, 141 insertions(+), 5 deletions(-) create mode 100644 open_wearable/lib/models/shared_sensor_configuration.dart create mode 100644 open_wearable/test/models/shared_sensor_configuration_test.dart 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..089b3305f --- /dev/null +++ b/open_wearable/lib/models/shared_sensor_configuration.dart @@ -0,0 +1,66 @@ +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<_Report> 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); + }); +} From ce4ec2294c41007a04c713a69aac9a7a151ae854 Mon Sep 17 00:00:00 2001 From: TobiasRoeddiger Date: Fri, 2 Oct 2026 11:53:22 +0200 Subject: [PATCH 2/2] style: keep shared stream public signature explicit --- .../models/shared_sensor_configuration.dart | 76 ++++++++++--------- 1 file changed, 42 insertions(+), 34 deletions(-) diff --git a/open_wearable/lib/models/shared_sensor_configuration.dart b/open_wearable/lib/models/shared_sensor_configuration.dart index 089b3305f..50906e3fc 100644 --- a/open_wearable/lib/models/shared_sensor_configuration.dart +++ b/open_wearable/lib/models/shared_sensor_configuration.dart @@ -6,8 +6,9 @@ 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<_Report> sharedSensorConfiguration(SensorConfigurationManager manager) => - (_reports[manager] ??= _SharedReports(manager)).stream; +Stream> + sharedSensorConfiguration(SensorConfigurationManager manager) => + (_reports[manager] ??= _SharedReports(manager)).stream; class _SharedReports { _SharedReports(this.manager); @@ -18,21 +19,24 @@ class _SharedReports { 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; + 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); + }; + _start(); + }, + isBroadcast: true, + ); Future _start() async { if (_source != null || _starting) return; @@ -41,24 +45,28 @@ class _SharedReports { // 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(); - } - }); + _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; }