From 8069ff52d58005429891f66c8b75b643f773beed Mon Sep 17 00:00:00 2001 From: Rohit Sangwan Date: Thu, 8 May 2025 16:22:18 +0530 Subject: [PATCH] Implement streams to get updates (#155) * Implement streams to get updates * Resolve comments * Update Readme * Apply suggestions from code review Co-authored-by: Foti Dim * Update Changelog * Update Readme --------- Co-authored-by: Foti Dim --- CHANGELOG.md | 2 + README.md | 40 +++++++--- .../xcshareddata/xcschemes/Runner.xcscheme | 1 + example/pubspec.lock | 2 +- lib/src/models/ble_connection_update.dart | 11 --- lib/src/models/model_exports.dart | 1 - lib/src/universal_ble.dart | 69 +++++++++------- lib/src/universal_ble_platform_interface.dart | 78 ++++++++++++++----- lib/src/universal_ble_stream_controller.dart | 37 +++++++++ .../universal_ble_stream_controller_test.dart | 32 ++++++++ 10 files changed, 200 insertions(+), 73 deletions(-) delete mode 100644 lib/src/models/ble_connection_update.dart create mode 100644 lib/src/universal_ble_stream_controller.dart create mode 100644 test/universal_ble_stream_controller_test.dart diff --git a/CHANGELOG.md b/CHANGELOG.md index 57765ad..430f5b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,7 @@ ## 0.18.0 +* Fix `connect` api bug to detect proper connection state * Improve docs for isPaired +* Add Streams: `scanStream`, `availabilityStream`, `connectionStream`, `characteristicValueStream`, `pairingStateStream` ## 0.17.0 * Fix Windows crash when calling pair APIs with an unknown deviceId diff --git a/README.md b/README.md index b279204..f950980 100644 --- a/README.md +++ b/README.md @@ -58,10 +58,13 @@ import 'package:universal_ble/universal_ble.dart'; ### Scanning ```dart -// Set a scan result handler -UniversalBle.onScanResult = (bleDevice) { +// Get scan updates from stream +UniversalBle.scanStream.listen((bleDevice) { // e.g. Use BleDevice ID to connect -} +}); + +// Or set a handler +UniversalBle.onScanResult = (bleDevice) {} // Perform a scan UniversalBle.startScan(); @@ -88,12 +91,15 @@ if (state == AvailabilityState.poweredOn) { UniversalBle.startScan(); } -// Or listen to bluetooth availability changes -UniversalBle.onAvailabilityChange = (state) { +// Listen to bluetooth availability changes using stream +UniversalBle.availabilityStream.listen((state) { if (state == AvailabilityState.poweredOn) { UniversalBle.startScan(); } -}; +}); + +// Or set a handler +UniversalBle.onAvailabilityChange = (state) {}; ``` See the [Bluetooth Availability](#bluetooth-availability) section for more. @@ -159,7 +165,12 @@ UniversalBle.connect(deviceId); // Disconnect from a device UniversalBle.disconnect(deviceId); -// Get connection/disconnection updates +// Get connection/disconnection updates using stream +UniversalBle.connectionStream(deviceId).listen((bool isConnected) { + debugPrint('OnConnectionChange $deviceId, $isConnected'); +}); + +// Or set a handler to get updates of all devices UniversalBle.onConnectionChange = (String deviceId, bool isConnected, String? error) { debugPrint('OnConnectionChange $deviceId, $isConnected Error: $error'); } @@ -192,7 +203,12 @@ UniversalBle.writeValue(deviceId, serviceId, characteristicId, value); // Subscribe to a characteristic UniversalBle.setNotifiable(deviceId, serviceId, characteristicId, BleInputProperty.notification); -// Get characteristic updates in `onValueChange` +// Get characteristic updates using stream +UniversalBle.characteristicValueStream(deviceId, characteristicId).listen((Uint8List value) { + debugPrint('OnValueChange $deviceId, $characteristicId, ${hex.encode(value)}'); +}); + +// Or set a handler to get updates of all characteristics UniversalBle.onValueChange = (String deviceId, String characteristicId, Uint8List value) { debugPrint('onValueChange $deviceId, $characteristicId, ${hex.encode(value)}'); } @@ -246,9 +262,13 @@ To discover encrypted characteristics, make sure your device is not paired and u #### Pairing state changes ```dart -UniversalBle.onPairingStateChange = (String deviceId, bool isPaired) { +// Get pairing state updates using stream +UniversalBle.pairingStateStream(deviceId).listen((bool isPaired) { // Handle pairing state change -} +}); + +// Or set a handler to get pairing state updates of all devices +UniversalBle.onPairingStateChange = (String deviceId, bool isPaired) {} ``` #### Unpair diff --git a/example/macos/Runner.xcodeproj/xcshareddata/xcschemes/Runner.xcscheme b/example/macos/Runner.xcodeproj/xcshareddata/xcschemes/Runner.xcscheme index 793b0cf..b3b52f0 100644 --- a/example/macos/Runner.xcodeproj/xcshareddata/xcschemes/Runner.xcscheme +++ b/example/macos/Runner.xcodeproj/xcshareddata/xcschemes/Runner.xcscheme @@ -59,6 +59,7 @@ ignoresPersistentStateOnLaunch = "NO" debugDocumentVersioning = "YES" debugServiceExtension = "internal" + enableGPUValidationMode = "1" allowLocationSimulation = "YES"> diff --git a/example/pubspec.lock b/example/pubspec.lock index 3ef07d5..680ce1f 100644 --- a/example/pubspec.lock +++ b/example/pubspec.lock @@ -394,7 +394,7 @@ packages: path: ".." relative: true source: path - version: "0.17.0" + version: "0.18.0" vector_math: dependency: transitive description: diff --git a/lib/src/models/ble_connection_update.dart b/lib/src/models/ble_connection_update.dart deleted file mode 100644 index f9727a7..0000000 --- a/lib/src/models/ble_connection_update.dart +++ /dev/null @@ -1,11 +0,0 @@ -class BleConnectionUpdate { - final String deviceId; - final bool isConnected; - final String? error; - - BleConnectionUpdate({ - required this.deviceId, - required this.isConnected, - this.error, - }); -} diff --git a/lib/src/models/model_exports.dart b/lib/src/models/model_exports.dart index 07e5354..2a2d267 100644 --- a/lib/src/models/model_exports.dart +++ b/lib/src/models/model_exports.dart @@ -1,4 +1,3 @@ -export 'package:universal_ble/src/models/ble_connection_update.dart'; export 'package:universal_ble/src/models/manufacturer_data.dart'; export 'package:universal_ble/src/models/platform_config.dart'; export 'package:universal_ble/src/models/queue_type.dart'; diff --git a/lib/src/universal_ble.dart b/lib/src/universal_ble.dart index 874c6ba..8d7690a 100644 --- a/lib/src/universal_ble.dart +++ b/lib/src/universal_ble.dart @@ -35,6 +35,26 @@ class UniversalBle { UniversalLogger.logInfo('Queue ${queueType.name}'); } + /// Scan Stream + static Stream get scanStream => _platform.scanStream; + + /// Bluetooth availability state stream + static Stream get availabilityStream => + _platform.availabilityStream; + + /// Connection stream of a device + static Stream connectionStream(String deviceId) => + _platform.connectionStream(deviceId); + + /// Characteristic value stream + static Stream characteristicValueStream( + String deviceId, String characteristicId) => + _platform.characteristicValueStream(deviceId, characteristicId); + + /// Pairing state stream + static Stream pairingStateStream(String deviceId) => + _platform.pairingStateStream(deviceId); + /// Get Bluetooth availability state. /// To be notified of updates, set [onAvailabilityChange] listener. static Future getBluetoothAvailabilityState() async { @@ -70,10 +90,6 @@ class UniversalBle { ); } - /// Connection stream of a device - static Stream connectionStream(String deviceId) => - _platform.connectionStream(deviceId); - /// Connect to a device. /// It is advised to stop scanning before connecting. /// It throws error if device connection fails. @@ -85,40 +101,35 @@ class UniversalBle { }) async { connectionTimeout ??= const Duration(seconds: 60); StreamSubscription? connectionSubscription; + Completer completer = Completer(); + + void handleError(dynamic error) { + if (completer.isCompleted) return; + connectionSubscription?.cancel(); + completer.completeError(ConnectionException(error)); + } try { - Completer completer = Completer(); - - connectionSubscription = connectionStream(deviceId).listen( - (BleConnectionUpdate event) { - connectionSubscription?.cancel(); - if (!completer.isCompleted) { - String? error = event.error; - if (error != null) { - completer.completeError(ConnectionException(error)); - } else { - completer.complete(event.isConnected); + connectionSubscription = _platform + .bleConnectionUpdateStreamController.stream + .where((e) => e.deviceId == deviceId) + .listen( + (e) { + if (e.error != null) { + handleError(e.error); + } else { + if (!completer.isCompleted) { + completer.complete(e.isConnected); } } }, - onError: (error) { - if (!completer.isCompleted) { - connectionSubscription?.cancel(); - completer.completeError(ConnectionException(error)); - } - }, + onError: handleError, + cancelOnError: true, ); _platform .connect(deviceId, connectionTimeout: connectionTimeout) - .catchError( - (error) { - if (!completer.isCompleted) { - connectionSubscription?.cancel(); - completer.completeError(ConnectionException(error)); - } - }, - ); + .catchError(handleError); if (!await completer.future.timeout(connectionTimeout)) { throw ConnectionException("Failed to connect"); diff --git a/lib/src/universal_ble_platform_interface.dart b/lib/src/universal_ble_platform_interface.dart index 818352d..91017ab 100644 --- a/lib/src/universal_ble_platform_interface.dart +++ b/lib/src/universal_ble_platform_interface.dart @@ -1,5 +1,6 @@ import 'dart:async'; import 'dart:typed_data'; +import 'package:universal_ble/src/universal_ble_stream_controller.dart'; import 'package:universal_ble/universal_ble.dart'; abstract class UniversalBlePlatform { @@ -10,7 +11,23 @@ abstract class UniversalBlePlatform { OnAvailabilityChange? onAvailabilityChange; OnPairingStateChange? onPairingStateChange; final Map _pairStateMap = {}; - StreamController? _connectionStreamController; + + final _scanStreamController = UniversalBleStreamController(); + + final bleConnectionUpdateStreamController = UniversalBleStreamController< + ({String deviceId, bool isConnected, String? error})>(); + + final _valueStreamController = UniversalBleStreamController< + ({String deviceId, String characteristicId, Uint8List value})>(); + + final _pairStateStreamController = + UniversalBleStreamController<({String deviceId, bool isPaired})>(); + + /// Send latest availability state upon subscribing + late final _availabilityStreamController = + UniversalBleStreamController( + initialEvent: getBluetoothAvailabilityState, + ); Future getBluetoothAvailabilityState(); @@ -64,19 +81,39 @@ abstract class UniversalBlePlatform { bool receivesAdvertisements(String deviceId) => true; - Stream connectionStream(String deviceId) { - _setupConnectionStreamIfRequired(); - return _connectionStreamController!.stream; - } + /// Streams + Stream get scanStream => _scanStreamController.stream; + Stream get availabilityStream => + _availabilityStreamController.stream; + + Stream connectionStream(String deviceId) => + bleConnectionUpdateStreamController.stream + .where((e) => e.deviceId == deviceId) + .map((e) => e.isConnected); + + Stream characteristicValueStream( + String deviceId, String characteristicId) => + _valueStreamController.stream.where((e) { + return e.deviceId == deviceId && e.characteristicId == characteristicId; + }).map((e) => e.value); + + Stream pairingStateStream(String deviceId) => + _pairStateStreamController.stream + .where((e) => e.deviceId == deviceId) + .map((e) => e.isPaired); + + /// Update Handlers void updateScanResult(BleDevice bleDevice) { + _scanStreamController.add(bleDevice); + try { onScanResult?.call(bleDevice); } catch (_) {} } void updateConnection(String deviceId, bool isConnected, [String? error]) { - _connectionStreamController?.add(BleConnectionUpdate( + bleConnectionUpdateStreamController.add(( deviceId: deviceId, isConnected: isConnected, error: error, @@ -88,7 +125,16 @@ abstract class UniversalBlePlatform { } void updateCharacteristicValue( - String deviceId, String characteristicId, Uint8List value) { + String deviceId, + String characteristicId, + Uint8List value, + ) { + _valueStreamController.add(( + deviceId: deviceId, + characteristicId: characteristicId, + value: value, + )); + try { onValueChange?.call( deviceId, BleUuidParser.string(characteristicId), value); @@ -96,6 +142,8 @@ abstract class UniversalBlePlatform { } void updateAvailability(AvailabilityState state) { + _availabilityStreamController.add(state); + try { onAvailabilityChange?.call(state); } catch (_) {} @@ -105,24 +153,12 @@ abstract class UniversalBlePlatform { if (_pairStateMap[deviceId] == isPaired) return; _pairStateMap[deviceId] = isPaired; + _pairStateStreamController.add((deviceId: deviceId, isPaired: isPaired)); + try { onPairingStateChange?.call(deviceId, isPaired); } catch (_) {} } - - /// Creates an auto disposable streamController - void _setupConnectionStreamIfRequired() { - if (_connectionStreamController != null) return; - - _connectionStreamController = StreamController.broadcast(); - - // Auto dispose if no more subscribers - _connectionStreamController?.onCancel = () { - // logInfo('Disposing Connection Stream'); - _connectionStreamController?.close(); - _connectionStreamController = null; - }; - } } // Callback types diff --git a/lib/src/universal_ble_stream_controller.dart b/lib/src/universal_ble_stream_controller.dart new file mode 100644 index 0000000..861865f --- /dev/null +++ b/lib/src/universal_ble_stream_controller.dart @@ -0,0 +1,37 @@ +import 'dart:async'; + +/// Auto disposable StreamController +class UniversalBleStreamController { + Future Function()? initialEvent; + UniversalBleStreamController({this.initialEvent}); + + StreamController? _streamController; + + Stream get stream { + _setupStreamIfRequired(); + return _streamController!.stream; + } + + bool get isClosed => _streamController?.isClosed ?? true; + + void add(T data) => _streamController?.add(data); + + void close() { + _streamController?.close(); + _streamController = null; + } + + void _setupStreamIfRequired() { + if (_streamController != null) return; + + _streamController = StreamController.broadcast( + onListen: () async { + try { + T? event = await initialEvent?.call(); + if (event != null) add(event); + } catch (_) {} + }, + onCancel: close, + ); + } +} diff --git a/test/universal_ble_stream_controller_test.dart b/test/universal_ble_stream_controller_test.dart new file mode 100644 index 0000000..c667338 --- /dev/null +++ b/test/universal_ble_stream_controller_test.dart @@ -0,0 +1,32 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:universal_ble/src/universal_ble_stream_controller.dart'; + +void main() { + group("Test UniversalBleStreamController", () { + test('Auto Dispose Stream', () { + var streamController = UniversalBleStreamController(); + expect(true, streamController.isClosed); + + // Stream should auto initialize on first subscription + var subscription = streamController.stream.listen((data) {}); + expect(false, streamController.isClosed); + + streamController.add(1); + + // Should auto close on cancelling last subscription + subscription.cancel(); + expect(true, streamController.isClosed); + }); + + test('Get InitialEvent on listen', () async { + var streamController = UniversalBleStreamController( + initialEvent: () async => 1, + ); + expect(true, streamController.isClosed); + + var firstValue = await streamController.stream.first; + expect(true, streamController.isClosed); + expect(1, firstValue); + }); + }); +}