From 2fe21962d29d5f649e2f8e7e71aaf07f02fee59c Mon Sep 17 00:00:00 2001 From: Rohit Sangwan Date: Fri, 7 Jun 2024 20:55:46 +0530 Subject: [PATCH] Add queue per device (#48) * Add queue per device * Improve API and documentation * Update lib/src/universal_ble.dart * Update lib/src/ble_command_queue.dart --------- Co-authored-by: Foti Dim Co-authored-by: Foti Dim --- CHANGELOG.md | 4 + README.md | 30 ++- example/lib/home/home.dart | 18 +- lib/src/ble_command_queue.dart | 106 ++++------- lib/src/models/model_exports.dart | 3 + lib/src/models/queue_type.dart | 5 + lib/src/queue.dart | 81 ++++++++ lib/src/universal_ble.dart | 176 +++++++++--------- lib/src/universal_ble_platform_interface.dart | 2 + lib/universal_ble.dart | 1 - pubspec.yaml | 2 +- 11 files changed, 262 insertions(+), 166 deletions(-) create mode 100644 lib/src/models/queue_type.dart create mode 100644 lib/src/queue.dart diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c90a71..5b3c1df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,7 @@ +## 0.9.12 +* Add .perDevice queue +* Improve code level documentation + ## 0.9.11 * Add device name prefix filtering diff --git a/README.md b/README.md index 6a25c35..58d2848 100644 --- a/README.md +++ b/README.md @@ -137,9 +137,9 @@ UniversalBle.connect(deviceId); // Disconnect from a device UniversalBle.disconnect(deviceId); -// Get notified for connection state changes +// Get connection state updates UniversalBle.onConnectionChanged = (String deviceId, BleConnectionState state) { - print('OnConnectionChanged $deviceId, $state'); + debugPrint('OnConnectionChanged $deviceId, $state'); } ``` @@ -168,7 +168,7 @@ UniversalBle.setNotifiable(deviceId, serviceId, characteristicId, BleInputProper // Get characteristic updates in `onValueChanged` UniversalBle.onValueChanged = (String deviceId, String characteristicId, Uint8List value) { - print('onValueChanged $deviceId, $characteristicId, ${hex.encode(value)}'); + debugPrint('onValueChanged $deviceId, $characteristicId, ${hex.encode(value)}'); } // Unsubscribe from a characteristic @@ -210,11 +210,31 @@ UniversalBle.enableBluetooth(); ## Command Queue -By default, all commands are executed in a queue, with each command waiting for the previous one to finish. This is because some platforms (e.g. Android) may fail to send consecutive commands without a delay between them. Therefore, it is a good idea to leave the queue enabled. +By default, all commands are executed in a global queue (`QueueType.global`), with each command waiting for the previous one to finish. + +If you want to parallelize commands between multiple devices, you can set: + +```dart +// Create a separate queue for each device. +UniversalBle.queueType = QueueType.perDevice; +``` + +You can also disable the queue completely and parallelize all commands, even for the same device, by using: ```dart // Disable queue -UniversalBle.queuesCommands = false; +UniversalBle.queueType = QueueType.none; +``` + +Keep in mind that some platforms (e.g. Android) may not handle well devices that fail to process consecutive commands without a minimum interval. Therefore, it is not advised to set `queueType` to `none`. + +You can get queue updates by setting: + +```dart +// Get queue state updates +UniversalBle.onQueueUpdate = (String id, int remainingItems) { + debugPrint("Queue: $id Remaining: $remainingItems"); +}; ``` ## Timeout diff --git a/example/lib/home/home.dart b/example/lib/home/home.dart index 19ab76e..c2440dc 100644 --- a/example/lib/home/home.dart +++ b/example/lib/home/home.dart @@ -22,7 +22,7 @@ class MyApp extends StatefulWidget { class _MyAppState extends State { final _scanResults = []; bool _isScanning = false; - bool _isQueueEnabled = true; + QueueType _queueType = QueueType.global; AvailabilityState? bleAvailabilityState; final List _services = [ @@ -45,7 +45,7 @@ class _MyAppState extends State { } /// Setup queue and timeout - UniversalBle.queuesCommands = _isQueueEnabled; + UniversalBle.queueType = _queueType; UniversalBle.timeout = const Duration(seconds: 10); UniversalBle.onAvailabilityChange = (state) { @@ -68,6 +68,10 @@ class _MyAppState extends State { } setState(() {}); }; + + // UniversalBle.onQueueUpdate = (String id, int remainingItems) { + // debugPrint("Queue: $id RemainingItems: $remainingItems"); + // }; } Future startScan() async { @@ -184,11 +188,15 @@ class _MyAppState extends State { }, ), PlatformButton( - text: _isQueueEnabled ? 'Disable Queue' : 'Enable Queue', + text: 'Queue: ${_queueType.name.toUpperCase()}', onPressed: () { setState(() { - _isQueueEnabled = !_isQueueEnabled; - UniversalBle.queuesCommands = _isQueueEnabled; + _queueType = switch (_queueType) { + QueueType.global => QueueType.perDevice, + QueueType.perDevice => QueueType.none, + QueueType.none => QueueType.global, + }; + UniversalBle.queueType = _queueType; }); }, ), diff --git a/lib/src/ble_command_queue.dart b/lib/src/ble_command_queue.dart index b2f5f12..bd7fdae 100644 --- a/lib/src/ble_command_queue.dart +++ b/lib/src/ble_command_queue.dart @@ -1,80 +1,50 @@ -import 'dart:async'; +import 'package:universal_ble/src/queue.dart'; +import 'package:universal_ble/universal_ble.dart'; -/// Original Author: Ryan Knell (https://github.com/rknell/dart_queue) - -/// Queue to execute Futures in order. -/// It awaits each future before executing the next one. +/// Execute commands in queue and manage queue per device class BleCommandQueue { - final Set _activeItems = {}; - int _lastProcessId = 0; - bool _isCancelled = false; - final List<_QueuedFuture> _nextCycle = []; + QueueType queueType = QueueType.global; + Duration? timeout = const Duration(seconds: 10); + OnQueueUpdate? onQueueUpdate; + final Queue _globalQueue = Queue(); + final Map _queueMap = {}; - Future add(Future Function() closure, {Duration? timeout}) { - if (_isCancelled) throw Exception('Queue Cancelled'); - final completer = Completer(); - _nextCycle.add(_QueuedFuture(closure, completer, timeout)); - _updateRemainingItems(); - if (_activeItems.isEmpty) _queueUpNext(); - return completer.future; + BleCommandQueue() { + _globalQueue.onRemainingItemsUpdate = (int items) { + onQueueUpdate?.call(QueueType.global.name, items); + }; } - void dispose() { - for (final item in _nextCycle) { - item.completer.completeError(Exception('Queue Cancelled')); + Future executeCommand( + Future Function() command, { + bool withTimeout = true, + String? deviceId, + }) { + Duration? duration = withTimeout ? timeout : null; + switch (queueType) { + case QueueType.none: + return duration != null ? command().timeout(duration) : command(); + case QueueType.global: + return _globalQueue.add(command, timeout: duration); + case QueueType.perDevice: + // If deviceId not available, use global queue + if (deviceId != null) { + return _getQueue(deviceId).add(command, timeout: duration); + } else { + return _globalQueue.add(command, timeout: duration); + } } - _nextCycle.removeWhere((item) => item.completer.isCompleted); - _isCancelled = true; } - void _queueUpNext() { - if (_nextCycle.isNotEmpty && !_isCancelled && _activeItems.length <= 1) { - final processId = _lastProcessId; - _activeItems.add(processId); - final item = _nextCycle.first; - _lastProcessId++; - _nextCycle.remove(item); - item.onComplete = () async { - _activeItems.remove(processId); - _updateRemainingItems(); - _queueUpNext(); + Queue _getQueue(String deviceId) { + Queue? queue = _queueMap[deviceId]; + if (queue == null) { + queue = Queue(); + queue.onRemainingItemsUpdate = (int items) { + onQueueUpdate?.call(deviceId, items); }; - unawaited(item.execute()); - } - } - - void _updateRemainingItems() { - // int remainingQueueItems = _nextCycle.length + _activeItems.length; - // onRemainingItemsUpdate?.call(_); - } -} - -class _QueuedFuture { - final Completer completer; - final Future Function() closure; - Function? onComplete; - final Duration? timeout; - - _QueuedFuture(this.closure, this.completer, this.timeout, {this.onComplete}); - - Future execute() async { - try { - T result; - if (timeout != null) { - result = await closure().timeout(timeout!); - } else { - result = await closure(); - } - if (result != null) { - completer.complete(result); - } else { - completer.complete(null); - } - await Future.microtask(() {}); - } catch (e, stack) { - completer.completeError(e, stack); - } finally { - if (onComplete != null) onComplete?.call(); + _queueMap[deviceId] = queue; } + return queue; } } diff --git a/lib/src/models/model_exports.dart b/lib/src/models/model_exports.dart index c2830d6..ea74218 100644 --- a/lib/src/models/model_exports.dart +++ b/lib/src/models/model_exports.dart @@ -1,3 +1,4 @@ +export 'package:universal_ble/src/models/queue_type.dart'; export 'package:universal_ble/src/models/uuid.dart'; export 'package:universal_ble/src/models/scan_filter.dart'; export 'package:universal_ble/src/models/ble_property.dart'; @@ -5,3 +6,5 @@ export 'package:universal_ble/src/models/ble_service.dart'; export 'package:universal_ble/src/models/availability_state.dart'; export 'package:universal_ble/src/models/ble_connection_state.dart'; export 'package:universal_ble/src/models/ble_scan_result.dart'; + + diff --git a/lib/src/models/queue_type.dart b/lib/src/models/queue_type.dart new file mode 100644 index 0000000..b516d1d --- /dev/null +++ b/lib/src/models/queue_type.dart @@ -0,0 +1,5 @@ +enum QueueType { + none, + perDevice, + global, +} diff --git a/lib/src/queue.dart b/lib/src/queue.dart new file mode 100644 index 0000000..313d9e4 --- /dev/null +++ b/lib/src/queue.dart @@ -0,0 +1,81 @@ +import 'dart:async'; + +/// Original Author: Ryan Knell (https://github.com/rknell/dart_queue) + +/// Queue to execute Futures in order. +/// It awaits each future before executing the next one. +class Queue { + final Set _activeItems = {}; + int _lastProcessId = 0; + bool _isCancelled = false; + final List<_QueuedFuture> _nextCycle = []; + Function(int)? onRemainingItemsUpdate; + + Future add(Future Function() closure, {Duration? timeout}) { + if (_isCancelled) throw Exception('Queue Cancelled'); + final completer = Completer(); + _nextCycle.add(_QueuedFuture(closure, completer, timeout)); + _updateRemainingItems(); + if (_activeItems.isEmpty) _queueUpNext(); + return completer.future; + } + + void dispose() { + for (final item in _nextCycle) { + item.completer.completeError(Exception('Queue Cancelled')); + } + _nextCycle.removeWhere((item) => item.completer.isCompleted); + _isCancelled = true; + } + + void _queueUpNext() { + if (_nextCycle.isNotEmpty && !_isCancelled && _activeItems.length <= 1) { + final processId = _lastProcessId; + _activeItems.add(processId); + final item = _nextCycle.first; + _lastProcessId++; + _nextCycle.remove(item); + item.onComplete = () async { + _activeItems.remove(processId); + _updateRemainingItems(); + _queueUpNext(); + }; + unawaited(item.execute()); + } + } + + void _updateRemainingItems() { + int remainingQueueItems = _nextCycle.length + _activeItems.length; + onRemainingItemsUpdate?.call(remainingQueueItems); + } +} + +class _QueuedFuture { + final Completer completer; + final Future Function() closure; + Function? onComplete; + final Duration? timeout; + + _QueuedFuture(this.closure, this.completer, this.timeout, {this.onComplete}); + + Future execute() async { + try { + T result; + if (timeout != null) { + result = await closure().timeout(timeout!); + } else { + result = await closure(); + } + if (result != null) { + completer.complete(result); + } else { + completer.complete(null); + } + await Future.microtask(() {}); + } catch (e, stack) { + completer.completeError(e, stack); + } finally { + if (onComplete != null) onComplete?.call(); + } + } +} diff --git a/lib/src/universal_ble.dart b/lib/src/universal_ble.dart index b07c0df..9cdda54 100644 --- a/lib/src/universal_ble.dart +++ b/lib/src/universal_ble.dart @@ -1,6 +1,7 @@ import 'dart:async'; import 'package:flutter/foundation.dart'; +import 'package:universal_ble/src/ble_command_queue.dart'; import 'package:universal_ble/src/universal_ble_linux/universal_ble_linux.dart'; import 'package:universal_ble/src/universal_ble_pigeon/universal_ble_pigeon_channel.dart'; import 'package:universal_ble/src/universal_ble_web/universal_ble_web.dart'; @@ -9,89 +10,95 @@ import 'package:universal_ble/universal_ble.dart'; class UniversalBle { /// Get platform specific implementation static UniversalBlePlatform _platform = _defaultPlatform(); + static final BleCommandQueue _bleCommandQueue = BleCommandQueue(); /// Set custom platform specific implementation (e.g. for testing) static void setInstance(UniversalBlePlatform instance) => _platform = instance; - /// Set global timeout for all commands - static Duration? timeout = const Duration(seconds: 10); - - static BleCommandQueue? _queue = BleCommandQueue(); - - /// Setup global queue for all commands, by default queue is enabled - static set queuesCommands(bool value) { - if (value) { - _queue ??= BleCommandQueue(); - UniversalBlePlatform.logInfo('Queue enabled'); - } else { - _queue?.dispose(); - _queue = null; - UniversalBlePlatform.logInfo('Queue disabled'); - } + /// Set global timeout for all commands. + /// Default timeout is 10 seconds + static set timeout(Duration? duration) { + _bleCommandQueue.timeout = duration; } - /// To get Bluetooth state availability - /// To get updates, set [onAvailabilityChange] listener + /// Set how commands will be executed. By default, all commands are executed in a global queue (`QueueType.global`), + /// with each command waiting for the previous one to finish. + /// + /// [QueueType.global] will execute commands of all devices in a single queue + /// [QueueType.perDevice] will execute command of each device in separate queues + /// [QueueType.none] will execute all commands in parallel + static set queueType(QueueType queueType) { + _bleCommandQueue.queueType = queueType; + UniversalBlePlatform.logInfo('Queue ${queueType.name}'); + } + + /// Get Bluetooth availability state + /// To be notified of updates, set [onAvailabilityChange] listener static Future getBluetoothAvailabilityState() async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.getBluetoothAvailabilityState(), - timeout: timeout, ); } - /// To Start scan, get scan results in [onScanResult] listener - /// might throw errors if Bluetooth is not available - /// `webRequestOptions` supported on Web only + /// Start scan. + /// Scan results will arrive in [onScanResult] listener + /// It might throw errors if Bluetooth is not available + /// `webRequestOptions` is supported on Web only static Future startScan({ ScanFilter? scanFilter, }) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.startScan(scanFilter: scanFilter), - timeout: null, + withTimeout: false, ); } - /// To Stop scan, set [onScanResult] listener to `null` if you don't need it anymore - /// might throw errors if Bluetooth is not available + /// Stop scan. + /// Set [onScanResult] listener to `null` if you don't need it anymore + /// It might throw errors if Bluetooth is not available static Future stopScan() async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.stopScan(), - timeout: null, + withTimeout: false, ); } - /// To connect to a device, get connection state in [onConnectionChanged] listener - /// preferred to stop scan before connecting - /// might throw errors if device is not connectable - /// `connectionTimeout` supported on Web only + /// Connect to a device. + /// Get notified of connection state changes in [onConnectionChanged] listener + /// It is advised to stop scanning before connecting + /// It might throw errors if device is not connectable + /// `connectionTimeout` is supported on Web only static Future connect( String deviceId, { Duration? connectionTimeout, }) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.connect(deviceId, connectionTimeout: connectionTimeout), - timeout: timeout, + deviceId: deviceId, ); } - /// To disconnect from a device, get connection state in [onConnectionChanged] listener + /// Disconnect from a device. + /// Get notified of connection state changes in [onConnectionChanged] listener static Future disconnect(String deviceId) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.disconnect(deviceId), - timeout: timeout, + deviceId: deviceId, ); } - /// To discover services of a device + /// Discover services of a device static Future> discoverServices(String deviceId) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.discoverServices(deviceId), - timeout: timeout, + deviceId: deviceId, ); } - /// To set a characteristic notifiable, set `bleInputProperty` to [BleInputProperty.notification] or [BleInputProperty.indication], get updates in [onValueChanged] listener + /// Set a characteristic notifiable. + /// Set `bleInputProperty` to [BleInputProperty.notification] or [BleInputProperty.indication] + /// Updates will arrive in [onValueChanged] listener /// To stop listening to a characteristic, set `bleInputProperty` to [BleInputProperty.disabled] static Future setNotifiable( String deviceId, @@ -99,31 +106,31 @@ class UniversalBle { String characteristic, BleInputProperty bleInputProperty, ) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.setNotifiable( deviceId, service, characteristic, bleInputProperty, ), - timeout: timeout, + deviceId: deviceId, ); } - /// To read a characteristic value - /// on iOS and MacOS, this command will also trigger [onValueChanged] listener + /// Read a characteristic value + /// On iOS and MacOS this command will also trigger [onValueChanged] listener static Future readValue( String deviceId, String service, String characteristic, ) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.readValue(deviceId, service, characteristic), - timeout: timeout, + deviceId: deviceId, ); } - /// To write a characteristic value + /// Write a characteristic value /// To write a characteristic value with response, set `bleOutputProperty` to [BleOutputProperty.withResponse] static Future writeValue( String deviceId, @@ -132,7 +139,7 @@ class UniversalBle { Uint8List value, BleOutputProperty bleOutputProperty, ) async { - await _executeCommand( + await _bleCommandQueue.executeCommand( () => _platform.writeValue( deviceId, service, @@ -140,68 +147,69 @@ class UniversalBle { value, bleOutputProperty, ), - timeout: timeout, + deviceId: deviceId, ); } - /// `requestMtu` not supported on `Linux` and `Web + /// Request MTU value + /// `requestMtu` is not supported on `Linux` and `Web static Future requestMtu(String deviceId, int expectedMtu) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.requestMtu(deviceId, expectedMtu), - timeout: timeout, + deviceId: deviceId, ); } - /// Pair commands are not supported on `iOS`, `MacOS` and `Web` + /// Check if a device is paired + /// Pair commands are not supported on `Apple` and `Web` static Future isPaired(String deviceId) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.isPaired(deviceId), - timeout: timeout, + deviceId: deviceId, ); } - /// To trigger pair request - /// might throw errors if device is already paired + /// Trigger pair request + /// It might throw an error if device is already paired static Future pair(String deviceId) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.pair(deviceId), - timeout: timeout, + deviceId: deviceId, ); } - /// To trigger unPair request - /// might throw errors if device is not paired + /// Unpair a device + /// It might throw an error if device is not paired static Future unPair(String deviceId) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.unPair(deviceId), - timeout: timeout, + deviceId: deviceId, ); } - /// To get connected devices to the system (connected by any app) - /// use [withServices] to filter devices by services - /// on `iOS`, `MacOS` [withServices] is required to get connected devices, else [1800] service will be used as default filter - /// on `Android`, `Linux` and `Windows`, if [withServices] is used, then internally all services will be discovered for each device first (either by connecting or by using cached services) + /// Get connected devices to the system (connected by any app) + /// Use [withServices] to filter devices by services + /// On `Apple`, [withServices] is required to get connected devices, else [1800] service will be used as default filter + /// On `Android`, `Linux` and `Windows`, if [withServices] is used, then internally all services will be discovered for each device first (either by connecting or by using cached services) /// Not supported on `Web` static Future> getConnectedDevices({ List? withServices, }) async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.getConnectedDevices(withServices), - timeout: timeout, ); } - /// Enabling Bluetooth, might throw errors if Bluetooth is not available + /// Enable Bluetooth + /// It might throw errors if Bluetooth is not available /// Not supported on `Web` and `Apple` static Future enableBluetooth() async { - return await _executeCommand( + return await _bleCommandQueue.executeCommand( () => _platform.enableBluetooth(), - timeout: timeout, ); } - /// To get Bluetooth state availability + /// Get Bluetooth state availability static set onAvailabilityChange(OnAvailabilityChange? onAvailabilityChange) { _platform.onAvailabilityChange = onAvailabilityChange; if (onAvailabilityChange != null) { @@ -211,19 +219,23 @@ class UniversalBle { } } - /// To get scan results + /// Get updates of remaining items of a queue + static set onQueueUpdate(OnQueueUpdate? onQueueUpdate) => + _bleCommandQueue.onQueueUpdate = onQueueUpdate; + + /// Get scan results static set onScanResult(OnScanResult? onScanResult) => _platform.onScanResult = onScanResult; - /// To get connection state changes + /// Get connection state changes static set onConnectionChanged(OnConnectionChanged? onConnectionChanged) => _platform.onConnectionChanged = onConnectionChanged; - /// To get characteristic value updates, set `bleInputProperty` in [setNotifiable] to [BleInputProperty.notification] or [BleInputProperty.indication] + /// Get characteristic value updates, set `bleInputProperty` in [setNotifiable] to [BleInputProperty.notification] or [BleInputProperty.indication] static set onValueChanged(OnValueChanged? onValueChanged) => _platform.onValueChanged = onValueChanged; - /// To get pair state changes, + /// Get pair state changes, static set onPairingStateChange(OnPairingStateChange pairingStateChange) => _platform.onPairingStateChange = pairingStateChange; @@ -234,12 +246,4 @@ class UniversalBle { } return UniversalBlePigeonChannel.instance; } - - static Future _executeCommand( - Future Function() command, { - required Duration? timeout, - }) { - return _queue?.add(command, timeout: timeout) ?? - (timeout != null ? command().timeout(timeout) : command()); - } } diff --git a/lib/src/universal_ble_platform_interface.dart b/lib/src/universal_ble_platform_interface.dart index 0372b29..5c504d8 100644 --- a/lib/src/universal_ble_platform_interface.dart +++ b/lib/src/universal_ble_platform_interface.dart @@ -84,3 +84,5 @@ typedef OnAvailabilityChange = void Function(AvailabilityState state); typedef OnPairingStateChange = void Function( String deviceId, bool isPaired, String? error); + +typedef OnQueueUpdate = void Function(String id, int remainingQueueItems); diff --git a/lib/universal_ble.dart b/lib/universal_ble.dart index bc868dd..c26951a 100644 --- a/lib/universal_ble.dart +++ b/lib/universal_ble.dart @@ -1,6 +1,5 @@ library universal_ble; -export 'package:universal_ble/src/ble_command_queue.dart'; export 'package:universal_ble/src/universal_ble_platform_interface.dart'; export 'package:universal_ble/src/universal_ble.dart'; export 'package:universal_ble/src/models/model_exports.dart'; diff --git a/pubspec.yaml b/pubspec.yaml index 7e85bd1..0f68d63 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -1,6 +1,6 @@ name: universal_ble description: A cross-platform (Android/iOS/macOS/Windows/Linux/Web) Bluetooth Low Energy (BLE) plugin for Flutter -version: 0.9.11 +version: 0.9.12 homepage: https://navideck.com repository: https://github.com/Navideck/universal_ble issue_tracker: https://github.com/Navideck/universal_ble/issues