Stream subscription fix

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 2.3.4
+
+* fixed a bug where the stream of sensor values would not be properly closed when the device is disconnected, which could lead to memory leaks and other issues
+* fixed a bug where ble subscriptions whould be cancelled for all devices when a single connection changes
+
 ## 2.3.3
 
 * renamed TauRing to OpenRing

lib/src/managers/ble_manager.dart

@@ -41,6 +41,23 @@ class BleManager extends BleGattManager {
     return _connectedDevicesIds.contains(deviceId);
   }
 
+  void _closeAndRemoveStreamsForDevice(String deviceId) {
+    final prefix = "$deviceId||";
+    final keys = _streamControllers.keys
+        .where((key) => key.startsWith(prefix))
+        .toList();
+
+    for (final key in keys) {
+      final controllers = _streamControllers.remove(key);
+      if (controllers == null) {
+        continue;
+      }
+      for (final controller in controllers) {
+        controller.close();
+      }
+    }
+  }
+
   void _init() {
     _scanStreamController = StreamController<DiscoveredDevice>.broadcast();
 
@@ -56,6 +73,7 @@ class BleManager extends BleGattManager {
         _connectCallbacks.remove(deviceId);
       } else {
         _connectedDevicesIds.remove(deviceId);
+        _closeAndRemoveStreamsForDevice(deviceId);
         _disconnectCallbacks[deviceId]?.call();
         _disconnectCallbacks.remove(deviceId);
       }
@@ -186,11 +204,9 @@ class BleManager extends BleGattManager {
     DiscoveredDevice device,
     VoidCallback onDisconnect,
   ) {
-    for (var list in _streamControllers.values) {
-      for (var e in list) {
-        e.close();
-      }
-    }
+    // Multi-device setup: only reset stale streams for the device that is
+    // being connected, not globally for all devices.
+    _closeAndRemoveStreamsForDevice(device.id);
 
     Completer<(bool, List<BleService>)> completer =
         Completer<(bool, List<BleService>)>();

lib/src/managers/open_earable_sensor_manager.dart

@@ -70,7 +70,7 @@ class OpenEarableSensorHandler extends SensorHandler<OpenEarableSensorConfig> {
     StreamController<Map<String, dynamic>> streamController =
         StreamController();
     int lastTimestamp = 0;
-    _bleManager
+    final subscription = _bleManager
         .subscribe(
       deviceId: deviceId,
       serviceId: sensorServiceUuid,
@@ -131,9 +131,13 @@ class OpenEarableSensorHandler extends SensorHandler<OpenEarableSensorConfig> {
           }
         }
       },
-      onError: (error) {},
+      onError: (error) {
+        streamController.addError(error);
+      },
     );
 
+    streamController.onCancel = subscription.cancel;
+
     return streamController.stream;
   }
 

lib/src/managers/v2_sensor_handler.dart

@@ -37,7 +37,7 @@ class V2SensorHandler extends SensorHandler<V2SensorConfig> {
 
     StreamController<Map<String, dynamic>> streamController =
         StreamController();
-    _bleManager
+    final subscription = _bleManager
         .subscribe(
       deviceId: _discoveredDevice.id,
       serviceId: sensorServiceUuid,
@@ -54,9 +54,12 @@ class V2SensorHandler extends SensorHandler<V2SensorConfig> {
       },
       onError: (error) {
         logger.e("Error while subscribing to sensor data: $error");
+        streamController.addError(error);
       },
     );
 
+    streamController.onCancel = subscription.cancel;
+
     return streamController.stream;
   }
 

pubspec.yaml

@@ -1,6 +1,6 @@
 name: open_earable_flutter
 description: This package provides functionality for interacting with OpenEarable devices. Control LED colors, control audio, and access raw sensor data.
-version: 2.3.3
+version: 2.3.4
 repository: https://github.com/OpenEarable/open_earable_flutter/tree/main
 
 platforms: