Share sensor configuration reports with network clients

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<SensorConfigurationManager>(
-          wearable: wearable,
-          streamName: streamName,
-        ).sensorConfigurationStream;
+        return sharedSensorConfiguration(
+          _requireCapability<SensorConfigurationManager>(
+            wearable: wearable,
+            streamName: streamName,
+          ),
+        );
       case 'button_events':
         return _requireCapability<ButtonManager>(
           wearable: wearable,

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<SensorConfiguration, SensorConfigurationValue>;
+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<Map<SensorConfiguration, SensorConfigurationValue>>
+    sharedSensorConfiguration(SensorConfigurationManager manager) =>
+        (_reports[manager] ??= _SharedReports(manager)).stream;
+
+class _SharedReports {
+  _SharedReports(this.manager);
+  final SensorConfigurationManager manager;
+  final _listeners = <MultiStreamController<_Report>>{};
+  StreamSubscription<_Report>? _source;
+  Future<void>? _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<void> _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;
+    }
+  }
+}

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<String, SensorConfigurationValue>.of(

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<SensorConfiguration, SensorConfigurationValue>;
+
+class _Manager implements SensorConfigurationManager {
+  int reads = 0;
+  late StreamController<Report> source;
+  @override
+  List<SensorConfiguration> get sensorConfigurations => [];
+  @override
+  Stream<Report> get sensorConfigurationStream {
+    reads++;
+    source = StreamController<Report>(sync: true);
+    return source.stream;
+  }
+}
+
+void main() {
+  test('UI and connectors share reports, replay, and independent cancellation',
+      () async {
+    final manager = _Manager();
+    final ui = <Report>[];
+    final remote = <Report>[];
+    final a = sharedSensorConfiguration(manager).listen(ui.add);
+    manager.source.add({});
+    final b = sharedSensorConfiguration(manager).listen(remote.add);
+    await Future<void>.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<void>.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 = <Object>[];
+    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);
+  });
+}