feat: Add async characteristic subscription that awaits BLE setup

packages/flutter_reactive_ble/lib/src/connected_device_operation.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:flutter_reactive_ble/flutter_reactive_ble.dart';
 import 'package:meta/meta.dart';
 import 'package:reactive_ble_platform_interface/reactive_ble_platform_interface.dart';
@@ -22,6 +24,11 @@ abstract class ConnectedDeviceOperation {
     Future<void> isDisconnected,
   );
 
+  Future<Stream<List<int>>> subscribeToCharacteristicAsync(
+    CharacteristicInstance characteristic,
+    Future<void> isDisconnected,
+  );
+
   Future<int> requestMtu(String deviceId, int mtu);
 
   Future<List<DiscoveredService>> discoverServices(String deviceId);
@@ -101,6 +108,102 @@ class ConnectedDeviceOperationImpl implements ConnectedDeviceOperation {
     return autosubscribingRepeater.stream;
   }
 
+  @override
+  Future<Stream<List<int>>> subscribeToCharacteristicAsync(
+    CharacteristicInstance characteristic,
+    Future<void> isDisconnected,
+  ) async {
+    final specificCharacteristicValueStream = characteristicValueStream
+        .where((update) => update.characteristic == characteristic)
+        .map((update) => update.result.dematerialize());
+
+    StreamController<List<int>>? controller;
+    StreamSubscription<List<int>>? valueSubscription;
+    final pendingActions = <void Function(StreamController<List<int>>)>[];
+
+    void startListeningToValues(Stream<List<int>> source) {
+      valueSubscription = source.listen(
+        (data) {
+          if (controller != null && controller.hasListener) {
+            controller.add(data);
+          } else {
+            pendingActions.add((c) => c.add(data));
+          }
+        },
+        onError: (Object error, StackTrace stackTrace) {
+          if (controller != null && controller.hasListener) {
+            controller.addError(error, stackTrace);
+          } else {
+            pendingActions.add((c) => c.addError(error, stackTrace));
+          }
+        },
+        onDone: () {
+          if (controller != null && controller.hasListener) {
+            controller.close();
+          } else {
+            pendingActions.add((c) => c.close());
+          }
+        },
+      );
+    }
+
+    startListeningToValues(specificCharacteristicValueStream);
+
+    try {
+      await _blePlatform
+          .subscribeToNotifications(characteristic)
+          .firstWhere((_) => true, orElse: () {});
+    } catch (_) {
+      await valueSubscription?.cancel();
+      rethrow;
+    }
+
+    var isSubscribed = true;
+
+    controller = StreamController<List<int>>.broadcast(
+      onListen: () {
+        if (!isSubscribed) {
+          isSubscribed = true;
+          startListeningToValues(
+            _blePlatform
+                .subscribeToNotifications(characteristic)
+                .asyncExpand((_) => specificCharacteristicValueStream),
+          );
+          return;
+        }
+        if (pendingActions.isNotEmpty) {
+          final actions = List.of(pendingActions);
+          pendingActions.clear();
+          for (final action in actions) {
+            action(controller!);
+          }
+        }
+      },
+      onCancel: () async {
+        isSubscribed = false;
+        pendingActions.clear();
+        await valueSubscription?.cancel();
+        valueSubscription = null;
+        await _blePlatform
+            .stopSubscribingToNotifications(characteristic)
+            .catchError((Object e) =>
+                // ignore: avoid_print
+                print("Error unsubscribing from notifications: $e"));
+      },
+    );
+
+    unawaited(
+      isDisconnected.then<void>((_) async {
+        pendingActions.clear();
+        await valueSubscription?.cancel();
+        valueSubscription = null;
+        await controller?.close();
+      }),
+    );
+
+    return controller.stream;
+  }
+
   @override
   Future<int> requestMtu(String deviceId, int mtu) async =>
       _blePlatform.requestMtuSize(deviceId, mtu);

packages/flutter_reactive_ble/lib/src/reactive_ble.dart

@@ -412,6 +412,19 @@ class FlutterReactiveBle {
     yield* (await resolveSingle(characteristic)).subscribe();
   }
 
+  /// Subscribes to updates from the characteristic specified, waiting for the
+  /// BLE subscription to be completed before returning.
+  ///
+  /// This stream terminates automatically when the device is disconnected.
+  ///
+  /// This method assumes there is a single characteristic with the ids specified in [characteristic]. If there are
+  /// multiple characteristics with the same id on a device, use [resolve] to find them all. Or use
+  /// [getDiscoveredServices] to select the [Service]s and [Characteristic]s you're interested in.
+  Future<Stream<List<int>>> subscribeToCharacteristicAsync(QualifiedCharacteristic characteristic) async {
+    await initialize();
+    return (await resolveSingle(characteristic)).subscribeAsync();
+  }
+
   Future<Iterable<Characteristic>> resolve(QualifiedCharacteristic characteristic) async {
     final services = await getDiscoveredServices(characteristic.deviceId);
     return services
@@ -559,6 +572,29 @@ class Characteristic {
         );
   }
 
+  /// Subscribes to updates from the characteristic specified, waiting for the
+  /// BLE subscription to be completed before returning.
+  ///
+  /// This stream terminates automatically when the device is disconnected.
+  Future<Stream<List<int>>> subscribeAsync() async {
+    _assertValidity();
+
+    final isDisconnected = _lib.connectedDeviceStream
+        .where((update) =>
+            update.deviceId == service.deviceId &&
+            (update.connectionState == DeviceConnectionState.disconnecting ||
+                update.connectionState == DeviceConnectionState.disconnected))
+        .cast<void>()
+        .firstWhere((_) => true, orElse: () {});
+
+    await _lib.initialize();
+
+    return _lib._connectedDeviceOperator.subscribeToCharacteristicAsync(
+      _ids,
+      isDisconnected,
+    );
+  }
+
   // A Characteristic becomes invalid when its device gets disconnected. After reconnecting to the device,
   // services and characteristics have to be rediscovered
   bool _valid = true;

packages/flutter_reactive_ble/test/connected_device_operation_test.dart

@@ -341,6 +341,130 @@ void main() {
                 ]));
           });
         });
+
+        group('Subscribe to characteristic async', () {
+          test('It waits for subscription to complete before returning stream',
+              () async {
+            final subscriptionCompleter = Completer<void>();
+            when(_blePlatform.subscribeToNotifications(charDevice))
+                .thenAnswer((_) => subscriptionCompleter.future.asStream());
+            when(_blePlatform.charValueUpdateStream)
+                .thenAnswer((_) => Stream.fromIterable([
+                      valueUpdate1,
+                      valueUpdateOtherDevice,
+                      valueUpdate2,
+                      valueUpdateSameDeviceOtherChar,
+                    ]));
+
+            var completed = false;
+            final streamFuture = _sut
+                .subscribeToCharacteristicAsync(
+                    charDevice, terminateCompleter.future)
+                .then((stream) {
+              completed = true;
+              return stream;
+            });
+
+            await Future<void>.delayed(Duration.zero);
+            expect(completed, isFalse);
+
+            subscriptionCompleter.complete();
+            final stream = await streamFuture;
+            expect(completed, isTrue);
+
+            expect(
+              stream,
+              emitsInOrder(<List<int>>[
+                [1],
+                [2],
+              ]),
+            );
+          });
+
+          test(
+              'It buffers notifications emitted before the returned stream is listened to',
+              () async {
+            final charValueController =
+                StreamController<CharacteristicValue>.broadcast();
+            addTearDown(charValueController.close);
+
+            when(_blePlatform.charValueUpdateStream)
+                .thenAnswer((_) => charValueController.stream);
+
+            final stream = await _sut.subscribeToCharacteristicAsync(
+                charDevice, terminateCompleter.future);
+
+            // Emit a notification before any listener is attached to `stream`
+            charValueController.add(valueUpdate1);
+            await Future<void>.delayed(Duration.zero);
+
+            final received = <List<int>>[];
+            final subscription = stream.listen(received.add);
+            addTearDown(subscription.cancel);
+
+            charValueController.add(valueUpdate2);
+            await Future<void>.delayed(Duration.zero);
+
+            expect(received, <List<int>>[
+              [1],
+              [2],
+            ]);
+          });
+
+          test(
+              'It stops subscribing to notifications when stream subscription is cancelled',
+              () async {
+            final charValueController =
+                StreamController<CharacteristicValue>.broadcast();
+            addTearDown(charValueController.close);
+
+            when(_blePlatform.charValueUpdateStream)
+                .thenAnswer((_) => charValueController.stream);
+
+            final stream = await _sut.subscribeToCharacteristicAsync(
+                charDevice, terminateCompleter.future);
+
+            final subscription = stream.listen((_) {});
+            await subscription.cancel();
+
+            verify(_blePlatform.stopSubscribingToNotifications(charDevice))
+                .called(1);
+          });
+
+          test('It closes stream when device disconnects', () async {
+            final charValueController =
+                StreamController<CharacteristicValue>.broadcast();
+            addTearDown(charValueController.close);
+
+            when(_blePlatform.charValueUpdateStream)
+                .thenAnswer((_) => charValueController.stream);
+
+            final stream = await _sut.subscribeToCharacteristicAsync(
+                charDevice, terminateCompleter.future);
+
+            terminateCompleter.complete(const ConnectionStateUpdate(
+              deviceId: '123',
+              connectionState: DeviceConnectionState.disconnected,
+              failure: null,
+            ));
+
+            await expectLater(stream, emitsDone);
+          });
+
+          test('It propagates error if subscribeToNotifications fails',
+              () async {
+            when(_blePlatform.subscribeToNotifications(charDevice))
+                .thenAnswer((_) => Stream<void>.error(Exception('BLE failure')));
+            when(_blePlatform.charValueUpdateStream)
+                .thenAnswer((_) => const Stream.empty());
+
+            await expectLater(
+              _sut.subscribeToCharacteristicAsync(
+                  charDevice, terminateCompleter.future),
+              throwsException,
+            );
+          });
+        });
       });
 
       group('Negotiate mtusize', () {

packages/flutter_reactive_ble/test/reactive_ble_test.dart

@@ -589,6 +589,87 @@ void main() {
       });
     });
 
+    group('Subscribe to characteristic async', () {
+      const update = ConnectionStateUpdate(
+        deviceId: '123',
+        connectionState: DeviceConnectionState.disconnected,
+        failure: null,
+      );
+
+      late QualifiedCharacteristic char;
+      late Stream<List<int>> valueStream;
+      late Stream<List<int>?> resultStream;
+
+      setUp(() async {
+        char = _createChar();
+        when(_deviceConnector.deviceConnectionStateUpdateStream).thenAnswer((_) => Stream.fromIterable([update]));
+        when(_deviceOperation.getDiscoverServices(char.deviceId)).thenAnswer((_) async => [
+              DiscoveredService(
+                serviceId: char.serviceId,
+                serviceInstanceId: "11",
+                characteristicIds: [char.characteristicId],
+                includedServices: [],
+                characteristics: [
+                  DiscoveredCharacteristic(
+                    characteristicId: Uuid.parse("1234"),
+                    characteristicInstanceId: "101",
+                    serviceId: char.serviceId,
+                    isReadable: true,
+                    isWritableWithResponse: true,
+                    isWritableWithoutResponse: true,
+                    isNotifiable: true,
+                    isIndicatable: true,
+                  ),
+                  DiscoveredCharacteristic(
+                    characteristicId: char.characteristicId,
+                    characteristicInstanceId: "101",
+                    serviceId: char.serviceId,
+                    isReadable: true,
+                    isWritableWithResponse: true,
+                    isWritableWithoutResponse: true,
+                    isNotifiable: true,
+                    isIndicatable: true,
+                  ),
+                ],
+              )
+            ]);
+
+        valueStream = Stream.fromIterable([
+          [1],
+          [2]
+        ]);
+
+        when(_deviceOperation.subscribeToCharacteristicAsync(any, any)).thenAnswer((_) async => valueStream);
+
+        resultStream = await _sut.subscribeToCharacteristicAsync(char);
+      });
+
+      test("It subscribes to the correct characteristic", () {
+        verify(_deviceOperation.subscribeToCharacteristicAsync(
+          CharacteristicInstance(
+            characteristicId: char.characteristicId,
+            characteristicInstanceId: "101",
+            serviceId: char.serviceId,
+            serviceInstanceId: "11",
+            deviceId: char.deviceId,
+          ),
+          any,
+        )).called(1);
+      });
+
+      test('It emits correct value', () {
+        expect(
+          resultStream,
+          emitsInOrder(
+            <List<int>>[
+              [1],
+              [2]
+            ],
+          ),
+        );
+      });
+    });
+
     group('Discover Services', () {
       const deviceId = '123';