Reactive/continuation

README.md

@@ -79,13 +79,12 @@ Once you have the `Ergometer` instance for the erg you want to connect to, you c
 
 ```dart
 StreamSubscription<Ergometer> ergConnectionStream = myErg.connectAndDiscover().listen((event) {
-    if(event == ErgometerConnectionState.connected) {
-      //do stuff here once the erg is connected
-    } else if (event == ErgometerConnectionState.disconnected) {
-      //handle disconnection here
-    }
-  });
-}
+  if(event == ErgometerConnectionState.connected) {
+    //do stuff here once the erg is connected
+  } else if (event == ErgometerConnectionState.disconnected) {
+    //handle disconnection here
+  }
+});
 ```
 
 When you are done, disconnect from your erg by cancelling the stream:

lib/models/ergblemanager.dart

@@ -1,9 +1,17 @@
 import 'package:c2bluetooth/constants.dart' as Identifiers;
+import 'package:flutter/foundation.dart';
 import 'package:flutter_reactive_ble/flutter_reactive_ble.dart';
 import 'ergometer.dart';
 
 class ErgBleManager {
-  final _manager = FlutterReactiveBle();
+  final FlutterReactiveBle _manager;
+
+  ErgBleManager() : _manager = FlutterReactiveBle();
+
+  /// Allow [ErgBleManager] to be tested using a Mocked bluetooth client
+  @visibleForTesting
+  ErgBleManager.withDependency({FlutterReactiveBle? bleClient})
+      : _manager = bleClient ?? FlutterReactiveBle();
 
   /// Begin scanning for Ergs.
   ///
@@ -12,7 +20,7 @@ class ErgBleManager {
   Stream<Ergometer> startErgScan() {
     return _manager.scanForDevices(withServices: [
       Uuid.parse(Identifiers.C2_ROWING_BASE_UUID)
-    ]).map((scanResult) => Ergometer(scanResult));
+    ]).map((scanResult) => Ergometer(scanResult, bleClient: _manager));
   }
 
   /// Clean up/destroy/deallocate resources so that they are availalble again

lib/models/ergometer.dart

@@ -2,19 +2,20 @@ import 'dart:typed_data';
 
 import 'package:c2bluetooth/c2bluetooth.dart';
 import 'package:flutter_reactive_ble/flutter_reactive_ble.dart';
-import '../internal/commands.dart';
-import '../internal/datatypes.dart';
+import 'package:c2bluetooth/internal/commands.dart';
+import 'package:c2bluetooth/internal/datatypes.dart';
 import 'package:csafe_fitness/csafe_fitness.dart';
-import '../helpers.dart';
+import 'package:c2bluetooth/helpers.dart';
 import 'workout.dart';
 import 'package:c2bluetooth/constants.dart' as Identifiers;
 import 'package:rxdart/rxdart.dart';
 
 enum ErgometerConnectionState { connecting, connected, disconnected }
 
 class Ergometer {
-  final _flutterReactiveBle = FlutterReactiveBle();
+  final FlutterReactiveBle _flutterReactiveBle;
   DiscoveredDevice _peripheral;
+  Stream<ConnectionStateUpdate>? _connection;
   Csafe? _csafeClient;
 
   /// Get the name of this erg. i.e. "PM5" + serial number
@@ -25,7 +26,8 @@ class Ergometer {
   /// This is intended only for internal use by [ErgBleManager.startErgScan].
   /// Consider this method a private API that is subject to unannounced breaking
   /// changes. There are likely much better methods to use for whatever you are trying to do.
-  Ergometer(this._peripheral);
+  Ergometer(this._peripheral, {required FlutterReactiveBle bleClient})
+      : _flutterReactiveBle = bleClient;
 
   /// Connect to this erg and discover the services and characteristics that it offers
   /// this returns a stream of [ErgometerConnectionState] events to enable monitoring the erg's connection state and disconnecting.
@@ -36,35 +38,44 @@ class Ergometer {
     //this may cause problems if the device goes out of range between scenning and trying to connect. maybe use connectToAdvertisingDevice instead to mitigate this and prevent a hang on android
 
     //if no services are specified in the `servicesWithCharacteristicsToDiscover` parameter, then full service discovery will be performed
-    return _flutterReactiveBle.connectToDevice(id: _peripheral.id).asyncMap((connectionStateUpdate) {
-      switch (connectionStateUpdate.connectionState) {
-        case DeviceConnectionState.connecting:
-          return ErgometerConnectionState.connecting;
-        case DeviceConnectionState.connected:
-          return ErgometerConnectionState.connected;
-        case DeviceConnectionState.disconnecting:
-          return ErgometerConnectionState.disconnected;
-        case DeviceConnectionState.disconnected:
-          return ErgometerConnectionState.disconnected;
-        default:
-          return ErgometerConnectionState.disconnected;
-      }
-    });
+    _connection = _flutterReactiveBle.connectToDevice(id: _peripheral.id);
+    return getMonitorConnectionState;
+  }
+
+  /// Deprecation notice: disconnect does not exists on FlutterReactiveBle library
+  @Deprecated("Destroy the Ergometer object to disconnect")
+  void disconnectOrCancel() {
+    throw NoSuchMethodError;
+  }
 
+  /// Subscribe to a stream of data from the erg
+  ///  (ex: general.distance, stroke.drive_length, ...)
+  Stream<dynamic> monitorForData(Set<String> datakey) {
+    throw UnimplementedError('$datakey not implemented');
   }
 
   /// Returns a stream of [WorkoutSummary] objects upon completion of any workout that would normally be saved to the Erg's memory. This includes any pre-programmed piece and any "just row" pieces longer than 1 minute.
   @Deprecated("This API is being deprecated in an upcoming version")
   Stream<WorkoutSummary> monitorForWorkoutSummary() {
-
-    var workoutSummaryCharacteristic1 = QualifiedCharacteristic(serviceId: Uuid.parse(Identifiers.C2_ROWING_PRIMARY_SERVICE_UUID), characteristicId: Uuid.parse(Identifiers.C2_ROWING_END_OF_WORKOUT_SUMMARY_CHARACTERISTIC_UUID), deviceId: _peripheral.id);
-
-    var workoutSummaryCharacteristic2 = QualifiedCharacteristic(serviceId: Uuid.parse(Identifiers.C2_ROWING_PRIMARY_SERVICE_UUID), characteristicId: Uuid.parse(Identifiers.C2_ROWING_END_OF_WORKOUT_SUMMARY_CHARACTERISTIC2_UUID), deviceId: _peripheral.id);
-
-    Stream<Uint8List> ws1 = _flutterReactiveBle.subscribeToCharacteristic(workoutSummaryCharacteristic1).asyncMap((datapoint) => Uint8List.fromList(datapoint));
-
-
-    Stream<Uint8List> ws2 = _flutterReactiveBle.subscribeToCharacteristic(workoutSummaryCharacteristic2).asyncMap((datapoint) => Uint8List.fromList(datapoint));
+    var workoutSummaryCharacteristic1 = QualifiedCharacteristic(
+        serviceId: Uuid.parse(Identifiers.C2_ROWING_PRIMARY_SERVICE_UUID),
+        characteristicId: Uuid.parse(
+            Identifiers.C2_ROWING_END_OF_WORKOUT_SUMMARY_CHARACTERISTIC_UUID),
+        deviceId: _peripheral.id);
+
+    var workoutSummaryCharacteristic2 = QualifiedCharacteristic(
+        serviceId: Uuid.parse(Identifiers.C2_ROWING_PRIMARY_SERVICE_UUID),
+        characteristicId: Uuid.parse(
+            Identifiers.C2_ROWING_END_OF_WORKOUT_SUMMARY_CHARACTERISTIC2_UUID),
+        deviceId: _peripheral.id);
+
+    Stream<Uint8List> ws1 = _flutterReactiveBle
+        .subscribeToCharacteristic(workoutSummaryCharacteristic1)
+        .asyncMap((datapoint) => Uint8List.fromList(datapoint));
+
+    Stream<Uint8List> ws2 = _flutterReactiveBle
+        .subscribeToCharacteristic(workoutSummaryCharacteristic2)
+        .asyncMap((datapoint) => Uint8List.fromList(datapoint));
 
     return Rx.zip2(ws1, ws2, (Uint8List ws1Result, Uint8List ws2Result) {
       List<int> combinedList = ws1Result.toList();
@@ -73,13 +84,42 @@ class Ergometer {
     });
   }
 
+  // Ensure compatibility
+  @Deprecated("Use getMonitorConnectionState getter")
+  Stream<ErgometerConnectionState> monitorConnectionState() {
+    return getMonitorConnectionState;
+  }
+
+  /// Expose a stream of events to enable monitoring the erg's connection state
+  /// This acts as a wrapper around the state provided by the internal bluetooth library to aid with swapping it out later.
+  Stream<ErgometerConnectionState> get getMonitorConnectionState =>
+      _connection!.asyncMap((connectionStateUpdate) {
+        switch (connectionStateUpdate.connectionState) {
+          case DeviceConnectionState.connecting:
+            return ErgometerConnectionState.connecting;
+          case DeviceConnectionState.connected:
+            return ErgometerConnectionState.connected;
+          case DeviceConnectionState.disconnecting:
+            return ErgometerConnectionState.disconnected;
+          default:
+            return ErgometerConnectionState.disconnected;
+        }
+      });
+
   /// An internal read function for accessing the PM's CSAFE API over bluetooth.
   ///
   /// Intended for passing to the csafe_fitness library to allow it to read response data  from the erg
   Stream<Uint8List> _readCsafe() {
-    var csafeRxCharacteristic = QualifiedCharacteristic(serviceId: Uuid.parse(Identifiers.C2_ROWING_CONTROL_SERVICE_UUID), characteristicId: Uuid.parse(Identifiers.C2_ROWING_PM_TRANSMIT_CHARACTERISTIC_UUID), deviceId: _peripheral.id);
-
-    return _flutterReactiveBle.subscribeToCharacteristic(csafeRxCharacteristic).asyncMap((datapoint) => Uint8List.fromList(datapoint)).asyncMap((datapoint) {
+    var csafeRxCharacteristic = QualifiedCharacteristic(
+        serviceId: Uuid.parse(Identifiers.C2_ROWING_CONTROL_SERVICE_UUID),
+        characteristicId:
+            Uuid.parse(Identifiers.C2_ROWING_PM_TRANSMIT_CHARACTERISTIC_UUID),
+        deviceId: _peripheral.id);
+
+    return _flutterReactiveBle
+        .subscribeToCharacteristic(csafeRxCharacteristic)
+        .asyncMap((datapoint) => Uint8List.fromList(datapoint))
+        .asyncMap((datapoint) {
       print("reading data: $datapoint");
       return datapoint;
     });
@@ -89,7 +129,11 @@ class Ergometer {
   ///
   /// Intended for passing to the csafe_fitness library to allow it to write commands to the erg
   void _writeCsafe(Uint8List value) {
-    var csafeTxCharacteristic = QualifiedCharacteristic(serviceId: Uuid.parse(Identifiers.C2_ROWING_CONTROL_SERVICE_UUID), characteristicId: Uuid.parse(Identifiers.C2_ROWING_PM_RECEIVE_CHARACTERISTIC_UUID), deviceId: _peripheral.id);
+    var csafeTxCharacteristic = QualifiedCharacteristic(
+        serviceId: Uuid.parse(Identifiers.C2_ROWING_CONTROL_SERVICE_UUID),
+        characteristicId:
+            Uuid.parse(Identifiers.C2_ROWING_PM_RECEIVE_CHARACTERISTIC_UUID),
+        deviceId: _peripheral.id);
 
     // return _peripheral.writeCharacteristic(
     //     Identifiers.C2_ROWING_CONTROL_SERVICE_UUID,
@@ -98,10 +142,12 @@ class Ergometer {
     //     true);
     // //.asyncMap((datapoint) => datapoint.read());
 
-    _flutterReactiveBle.writeCharacteristicWithResponse(csafeTxCharacteristic, value: value);
+    _flutterReactiveBle.writeCharacteristicWithResponse(csafeTxCharacteristic,
+        value: value);
   }
 
-  @Deprecated("This is a temporary function for development/experimentation and will be gone very soon")
+  @Deprecated(
+      "This is a temporary function for development/experimentation and will be gone very soon")
   void configure2kWorkout() async {
     //Workout workout
     await _csafeClient!.sendCommands([

pubspec.yaml

@@ -19,6 +19,7 @@ dependencies:
 dev_dependencies:
   flutter_test:
     sdk: flutter
+  mocktail: ^1.0.4
 
 # For information on the generic Dart part of this file, see the
 # following page: https://dart.dev/tools/pub/pubspec

test/ergblemanager_test.dart

@@ -1,17 +1,61 @@
+import 'dart:typed_data';
+
+import 'package:c2bluetooth/c2bluetooth.dart';
+import 'package:c2bluetooth/constants.dart' as Identifiers;
+import 'package:flutter_reactive_ble/flutter_reactive_ble.dart';
 import 'package:flutter_test/flutter_test.dart';
+import 'package:mocktail/mocktail.dart';
 
-// import '../lib/models/ergblemanager.dart';
+class MockFlutterReactiveBle extends Mock implements FlutterReactiveBle {}
 
 void main() {
-  test('can obtain stream of ergometers present', () {
-    // final bytes = Uint8List.fromList([0, 0, 0, 128]);
-    // expect(CsafeIntExtension.fromBytes(bytes), 128);
-    // expect(CsafeIntExtension.fromBytes(bytes, Endian.little), 2147483648);
-  });
+  setUp(() {});
+  test('translate the stream of discovered devices as ergometers', () {
+    /// The whole purpose of the startErgScan method is to translate
+    /// FlutterReactiveBle stream of DiscoveredDevice into Ergometer objects.
+    ///
+    /// - non-PM5 devices are already filtered-out by FlutterReactiveBle
+    /// - during subscribing we return a fake status data
+
+    /// declare ErgBleManager with a mocked Reactive Ble
+    final mockReactive = MockFlutterReactiveBle();
+    final ble = ErgBleManager.withDependency(bleClient: mockReactive);
+
+    /// create a fake stream of Discovered devices matching C2_ROWING_BASE_UUID service
+    final fakePM_1 = DiscoveredDevice(
+        id: 'xxxx',
+        name: 'PM5_1',
+        serviceUuids: [Uuid.parse(Identifiers.C2_ROWING_BASE_UUID)],
+        serviceData: {},
+        manufacturerData: Uint8List.fromList([1, 0, 0]),
+        rssi: 10);
+    final fakePM_2 = DiscoveredDevice(
+        id: 'yyyy',
+        name: 'PM5_2',
+        serviceUuids: [Uuid.parse(Identifiers.C2_ROWING_BASE_UUID)],
+        serviceData: {},
+        manufacturerData: Uint8List.fromList([2, 0, 0]),
+        rssi: 10);
+    final fakeScan = Stream.fromIterable([fakePM_1, fakePM_2]);
+
+    /// Adding mock answer from the [FlutterReactiveBle]
+    when(() => mockReactive.scanForDevices(
+            withServices: any(
+                named: "withServices",
+                that: predicate<List<Uuid>>((services) => services
+                    .contains(Uuid.parse(Identifiers.C2_ROWING_BASE_UUID))))))
+        .thenAnswer((_) => fakeScan);
+    when(() => mockReactive.statusStream)
+        .thenAnswer((_) => Stream.value(BleStatus.ready));
 
-  test('does not recognize non-concept2 devices', () {
-    // final bytes = Uint8List.fromList([0, 0, 0, 128]);
-    // expect(CsafeIntExtension.fromBytes(bytes), 128);
-    // expect(CsafeIntExtension.fromBytes(bytes, Endian.little), 2147483648);
+    /// Ensure DiscoveredDevice events are translated as Ergometer events
+    /// we expect only them in matching order
+    expect(
+        ble.startErgScan(),
+        emitsInOrder([
+          predicate<Ergometer>((e) => e.name == fakePM_1.name),
+          predicate<Ergometer>((e) => e.name == fakePM_2.name),
+          emitsDone,
+        ]));
   });
 }