From 4667d158b7126f0712a6cf601fbce2e9fbb3fc5c Mon Sep 17 00:00:00 2001 From: fotidim Date: Tue, 11 Aug 2026 19:59:49 +0200 Subject: [PATCH 1/2] Fix Apple queued write throughput (#271) --- CHANGELOG.md | 1 + README.md | 2 + .../universal_ble/UniversalBleHelper.swift | 14 +++ .../universal_ble/UniversalBlePlugin.swift | 89 +++++++++----- lib/src/queue.dart | 52 ++++++-- lib/src/universal_ble.dart | 8 ++ lib/src/utils/ble_command_queue.dart | 42 ++++++- test/universal_ble_write_queue_test.dart | 112 ++++++++++++++++++ 8 files changed, 276 insertions(+), 44 deletions(-) create mode 100644 test/universal_ble_write_queue_test.dart diff --git a/CHANGELOG.md b/CHANGELOG.md index b7581651..63833378 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ ## Unreleased * Windows: Harden BLE connection lifetime, asynchronous callbacks, and notification subscription handling * Android: instantly close the GATT client on disconnect +* Apple: Prevent queued writes from degrading throughput by pipelining them through the native GATT queue ## 2.1.1 * Android: Fix BluetoothDevice null-safety compile error under Kotlin 2.x diff --git a/README.md b/README.md index a6e4bbab..b598ca09 100644 --- a/README.md +++ b/README.md @@ -592,6 +592,8 @@ 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`. +On iOS, macOS, and Chrome running on Apple platforms, consecutive writes are submitted to the native GATT queue ahead of completion so the platform can maintain throughput. Later queued non-write commands still wait for all preceding writes. + You can get queue updates by setting: ```dart diff --git a/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift b/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift index 45001b55..b5b44cf5 100644 --- a/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift +++ b/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift @@ -197,6 +197,20 @@ class CharacteristicWriteFuture { } } +class PendingWriteWithoutResponse { + let deviceId: String + let characteristic: CBCharacteristic + let data: Data + let result: (Result) -> Void + + init(deviceId: String, characteristic: CBCharacteristic, data: Data, result: @escaping (Result) -> Void) { + self.deviceId = deviceId + self.characteristic = characteristic + self.data = data + self.result = result + } +} + class CharacteristicNotifyFuture { let deviceId: String let characteristicId: String diff --git a/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift b/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift index f73f417c..72a0ef96 100644 --- a/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift +++ b/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift @@ -95,7 +95,7 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral private var activeServiceDiscoveries: [String: UniversalBleAsyncServiceDiscovery] = [:] private var characteristicReadFutures = [CharacteristicReadFuture]() private var characteristicWriteFutures = [CharacteristicWriteFuture]() - private var characteristicWriteWithoutResponseFutures = [CharacteristicWriteFuture]() + private var pendingWriteWithoutResponse = [PendingWriteWithoutResponse]() private var characteristicNotifyFutures = [CharacteristicNotifyFuture]() private var discoverServicesFutures = [DiscoverServicesFuture]() private var rssiReadFutures = [RssiReadFuture]() @@ -305,6 +305,15 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral } return false } + pendingWriteWithoutResponse.removeAll { pending in + if pending.deviceId == deviceId { + pending.result( + Result.failure(createFlutterError(code: .deviceDisconnected, message: "Device Disconnected")) + ) + return true + } + return false + } characteristicNotifyFutures.removeAll { future in if future.deviceId == deviceId { future.result( @@ -425,6 +434,10 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral completion(Result.failure(createFlutterError(code: .deviceNotFound, message: "Unknown deviceId:\(self)"))) return } + guard peripheral.state == .connected else { + completion(Result.failure(createFlutterError(code: .deviceDisconnected, message: "Device Disconnected"))) + return + } guard let gattCharacteristic = peripheral.getCharacteristic(characteristic, of: service) else { completion(Result.failure(createFlutterError(code: .characteristicNotFound, message: "Unknown characteristic:\(characteristic)"))) return @@ -437,20 +450,35 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral completion(Result.failure(createFlutterError(code: .characteristicDoesNotSupportWrite, message: "Characteristic does not support write withResponse"))) return } - } else if type == CBCharacteristicWriteType.withoutResponse { - if !gattCharacteristic.properties.contains(.writeWithoutResponse) { - completion(Result.failure(createFlutterError(code: .characteristicDoesNotSupportWriteWithoutResponse, message: "Characteristic does not support write withoutResponse"))) - return - } + characteristicWriteFutures.append( + CharacteristicWriteFuture( + deviceId: deviceId, + characteristicId: gattCharacteristic.uuid.uuidStr, + serviceId: gattCharacteristic.service?.uuid.uuidStr, + result: completion + ) + ) + peripheral.writeValue(value.data, for: gattCharacteristic, type: .withResponse) + return } - peripheral.writeValue(value.data, for: gattCharacteristic, type: type) - // Wait for future response - let future = CharacteristicWriteFuture(deviceId: deviceId, characteristicId: gattCharacteristic.uuid.uuidStr, serviceId: gattCharacteristic.service?.uuid.uuidStr, result: completion) - if type == CBCharacteristicWriteType.withResponse { - characteristicWriteFutures.append(future) + if !gattCharacteristic.properties.contains(.writeWithoutResponse) { + completion(Result.failure(createFlutterError(code: .characteristicDoesNotSupportWriteWithoutResponse, message: "Characteristic does not support write withoutResponse"))) + return + } + + if peripheral.canSendWriteWithoutResponse { + peripheral.writeValue(value.data, for: gattCharacteristic, type: .withoutResponse) + completion(Result.success(())) } else { - characteristicWriteWithoutResponseFutures.append(future) + pendingWriteWithoutResponse.append( + PendingWriteWithoutResponse( + deviceId: deviceId, + characteristic: gattCharacteristic, + data: value.data, + result: completion + ) + ) } } @@ -652,27 +680,32 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral } public func peripheralIsReady(toSendWriteWithoutResponse peripheral: CBPeripheral) { - characteristicWriteWithoutResponseFutures.removeAll { future in - if future.deviceId == peripheral.uuid.uuidString { - future.result(Result.success({}())) - return true + let deviceId = peripheral.uuid.uuidString + while peripheral.state == .connected && peripheral.canSendWriteWithoutResponse { + guard let index = pendingWriteWithoutResponse.firstIndex(where: { $0.deviceId == deviceId }) else { + return } - return false + let pending = pendingWriteWithoutResponse.remove(at: index) + peripheral.writeValue(pending.data, for: pending.characteristic, type: .withoutResponse) + pending.result(Result.success(())) } } public func peripheral(_ peripheral: CBPeripheral, didWriteValueFor characteristic: CBCharacteristic, error: Error?) { - characteristicWriteFutures.removeAll { future in - if future.deviceId == peripheral.uuid.uuidString && future.characteristicId == characteristic.uuid.uuidStr && future.serviceId == characteristic.service?.uuid.uuidStr { - if let flutterError = error?.toFlutterError() { - UniversalBleLogger.shared.logError("WRITE_FAILED <- \(peripheral.uuid.uuidString) \(characteristic.uuid.uuidStr): \(flutterError.message ?? "")") - future.result(Result.failure(flutterError)) - } else { - future.result(Result.success({}())) - } - return true - } - return false + guard let index = characteristicWriteFutures.firstIndex(where: { + $0.deviceId == peripheral.uuid.uuidString && + $0.characteristicId == characteristic.uuid.uuidStr && + $0.serviceId == characteristic.service?.uuid.uuidStr + }) else { + return + } + + let future = characteristicWriteFutures.remove(at: index) + if let flutterError = error?.toFlutterError() { + UniversalBleLogger.shared.logError("WRITE_FAILED <- \(peripheral.uuid.uuidString) \(characteristic.uuid.uuidStr): \(flutterError.message ?? "")") + future.result(Result.failure(flutterError)) + } else { + future.result(Result.success(())) } } diff --git a/lib/src/queue.dart b/lib/src/queue.dart index 1203bce7..40ea2763 100644 --- a/lib/src/queue.dart +++ b/lib/src/queue.dart @@ -3,20 +3,40 @@ 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. +/// It awaits each future before executing the next one unless consecutive +/// commands opt into [addConcurrent]. class Queue { - final Set _activeItems = {}; + final Map _activeItems = {}; int _lastProcessId = 0; bool _isCancelled = false; final List<_QueuedFuture> _nextCycle = []; Function(int)? onRemainingItemsUpdate; - Future add(Future Function() closure, [Duration? timeout]) { + Future add(Future Function() closure, [Duration? timeout]) => + _add(closure, timeout); + + Future addConcurrent( + Future Function() closure, [ + Duration? timeout, + ]) => _add(closure, timeout, canRunConcurrently: true); + + Future _add( + Future Function() closure, + Duration? timeout, { + bool canRunConcurrently = false, + }) { if (_isCancelled) throw Exception('Queue Cancelled'); final completer = Completer(); - _nextCycle.add(_QueuedFuture(closure, completer, timeout)); + _nextCycle.add( + _QueuedFuture( + closure, + completer, + timeout, + canRunConcurrently: canRunConcurrently, + ), + ); _updateRemainingItems(); - if (_activeItems.isEmpty) _queueUpNext(); + _queueUpNext(); return completer.future; } @@ -29,12 +49,11 @@ class Queue { } void _queueUpNext() { - if (_nextCycle.isNotEmpty && !_isCancelled && _activeItems.length <= 1) { + while (_nextCycle.isNotEmpty && !_isCancelled && _canRunNext()) { final processId = _lastProcessId; - _activeItems.add(processId); - final item = _nextCycle.first; + final item = _nextCycle.removeAt(0); + _activeItems[processId] = item.canRunConcurrently; _lastProcessId++; - _nextCycle.remove(item); item.onComplete = () async { _activeItems.remove(processId); _updateRemainingItems(); @@ -44,6 +63,12 @@ class Queue { } } + bool _canRunNext() { + if (_activeItems.isEmpty) return true; + return _nextCycle.first.canRunConcurrently && + _activeItems.values.every((canRunConcurrently) => canRunConcurrently); + } + void _updateRemainingItems() { int remainingQueueItems = _nextCycle.length + _activeItems.length; onRemainingItemsUpdate?.call(remainingQueueItems); @@ -55,8 +80,15 @@ class _QueuedFuture { final Future Function() closure; Function? onComplete; final Duration? timeout; + final bool canRunConcurrently; - _QueuedFuture(this.closure, this.completer, this.timeout, {this.onComplete}); + _QueuedFuture( + this.closure, + this.completer, + this.timeout, { + this.onComplete, + this.canRunConcurrently = false, + }); Future execute() async { try { diff --git a/lib/src/universal_ble.dart b/lib/src/universal_ble.dart index f3d9e306..23ee9c7c 100644 --- a/lib/src/universal_ble.dart +++ b/lib/src/universal_ble.dart @@ -344,6 +344,9 @@ class UniversalBle { Duration? timeout, String? queueId, }) async { + // CoreBluetooth (and Chromium on Apple platforms) owns a native GATT + // operation queue. Let consecutive writes fill it without letting either + // writes or other operations cross a queue barrier. await _bleCommandQueue.queueCommand( () => _platform.writeValue( deviceId, @@ -357,9 +360,14 @@ class UniversalBle { timeout: timeout, deviceId: deviceId, queueId: queueId, + canRunConcurrently: _usesNativeWriteQueue, ); } + static bool get _usesNativeWriteQueue => + defaultTargetPlatform == TargetPlatform.iOS || + defaultTargetPlatform == TargetPlatform.macOS; + /// Requests an MTU (Maximum Transmission Unit) value for the connection. /// /// **⚠️ Note:** Requesting an MTU is a *best-effort* operation. On many platforms diff --git a/lib/src/utils/ble_command_queue.dart b/lib/src/utils/ble_command_queue.dart index 299d620f..3badecb8 100644 --- a/lib/src/utils/ble_command_queue.dart +++ b/lib/src/utils/ble_command_queue.dart @@ -16,6 +16,7 @@ class BleCommandQueue { String? deviceId, Duration? timeout, String? queueId, + bool canRunConcurrently = false, }) { Duration? timeoutDuration = timeout ?? this.timeout; if (timeoutDuration == null) { @@ -23,13 +24,22 @@ class BleCommandQueue { command, deviceId: deviceId, queueId: queueId, + canRunConcurrently: canRunConcurrently, ); } return switch (queueType) { - QueueType.global => _queue(queueId).add(command, timeoutDuration), - QueueType.perDevice => _queue( - queueId ?? deviceId, - ).add(command, timeoutDuration), + QueueType.global => _add( + _queue(queueId), + command, + timeoutDuration, + canRunConcurrently, + ), + QueueType.perDevice => _add( + _queue(queueId ?? deviceId), + command, + timeoutDuration, + canRunConcurrently, + ), QueueType.none => command().timeout(timeoutDuration), }; } @@ -38,14 +48,34 @@ class BleCommandQueue { Future Function() command, { String? deviceId, String? queueId, + bool canRunConcurrently = false, }) { return switch (queueType) { - QueueType.global => _queue(queueId).add(command), - QueueType.perDevice => _queue(queueId ?? deviceId).add(command), + QueueType.global => _add( + _queue(queueId), + command, + null, + canRunConcurrently, + ), + QueueType.perDevice => _add( + _queue(queueId ?? deviceId), + command, + null, + canRunConcurrently, + ), QueueType.none => command(), }; } + Future _add( + Queue queue, + Future Function() command, + Duration? timeout, + bool canRunConcurrently, + ) => canRunConcurrently + ? queue.addConcurrent(command, timeout) + : queue.add(command, timeout); + Queue _queue(String? id) { final queueKey = id ?? globalQueueId; return _queueMap[queueKey] ?? _newQueue(queueKey); diff --git a/test/universal_ble_write_queue_test.dart b/test/universal_ble_write_queue_test.dart new file mode 100644 index 00000000..6c1d5d1c --- /dev/null +++ b/test/universal_ble_write_queue_test.dart @@ -0,0 +1,112 @@ +import 'dart:async'; + +import 'package:flutter/foundation.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:universal_ble/universal_ble.dart'; + +import 'universal_ble_test_mock.dart'; + +void main() { + setUp(() { + UniversalBle.clearQueue(); + UniversalBle.queueType = QueueType.global; + }); + + tearDown(() { + UniversalBle.clearQueue(); + debugDefaultTargetPlatformOverride = null; + }); + + test('pipelines consecutive Apple writes', () async { + debugDefaultTargetPlatformOverride = TargetPlatform.macOS; + final platform = _PendingWritePlatform(); + UniversalBle.setInstance(platform); + + final first = _write(1); + final second = _write(2); + + expect(platform.started, [1, 2]); + + platform.pending[1].complete(); + platform.pending[0].complete(); + await first; + await second; + }); + + test('keeps Android writes strictly serialized', () async { + debugDefaultTargetPlatformOverride = TargetPlatform.android; + final platform = _PendingWritePlatform(); + UniversalBle.setInstance(platform); + + final first = _write(1); + final second = _write(2); + + expect(platform.started, [1]); + + platform.pending.single.complete(); + await first; + await pumpEventQueue(); + expect(platform.started, [1, 2]); + + platform.pending[1].complete(); + await second; + }); + + test('does not pipeline an Apple write across a queue barrier', () async { + debugDefaultTargetPlatformOverride = TargetPlatform.macOS; + final platform = _PendingWritePlatform(); + UniversalBle.setInstance(platform); + + final read = UniversalBle.read('device', '180a', '202a'); + final write = _write(1); + + expect(platform.reads, 1); + expect(platform.started, isEmpty); + + platform.readPending.complete(Uint8List(0)); + await read; + await pumpEventQueue(); + expect(platform.started, [1]); + + platform.pending.single.complete(); + await write; + }); +} + +Future _write(int value) => + UniversalBle.write('device', '180a', '202a', Uint8List.fromList([value])); + +class _PendingWritePlatform extends UniversalBlePlatformMock { + final started = []; + final pending = >[]; + final readPending = Completer(); + var reads = 0; + + @override + Future readValue( + String deviceId, + String service, + String characteristic, { + Duration? timeout, + }) { + reads++; + return readPending.future; + } + + @override + Future readRssi(String deviceId) => throw UnimplementedError(); + + @override + Future writeValue( + String deviceId, + String service, + String characteristic, + Uint8List value, + BleOutputProperty bleOutputProperty, + ) { + started.add(value.single); + final completion = Completer(); + pending.add(completion); + return completion.future; + } +} From c4d3afef29d0963d3f7fb073f0432d92ee9b8fec Mon Sep 17 00:00:00 2001 From: fotidim Date: Tue, 11 Aug 2026 20:31:29 +0200 Subject: [PATCH 2/2] Refine Apple write pipelining --- .../universal_ble/UniversalBleHelper.swift | 9 +--- .../universal_ble/UniversalBlePlugin.swift | 42 +++++++++---------- .../universal_ble_platform_interface.dart | 3 ++ lib/src/universal_ble.dart | 8 +--- .../universal_ble_pigeon_channel.dart | 5 +++ .../universal_ble_web/universal_ble_web.dart | 5 +++ lib/src/utils/ble_command_queue.dart | 38 ++++++++++++++++- test/universal_ble_test_mock.dart | 5 +++ test/universal_ble_write_queue_test.dart | 24 +++++------ 9 files changed, 89 insertions(+), 50 deletions(-) diff --git a/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift b/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift index b5b44cf5..781b3672 100644 --- a/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift +++ b/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift @@ -197,18 +197,11 @@ class CharacteristicWriteFuture { } } -class PendingWriteWithoutResponse { +struct PendingWriteWithoutResponse { let deviceId: String let characteristic: CBCharacteristic let data: Data let result: (Result) -> Void - - init(deviceId: String, characteristic: CBCharacteristic, data: Data, result: @escaping (Result) -> Void) { - self.deviceId = deviceId - self.characteristic = characteristic - self.data = data - self.result = result - } } class CharacteristicNotifyFuture { diff --git a/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift b/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift index 72a0ef96..1ad92039 100644 --- a/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift +++ b/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift @@ -443,10 +443,9 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral return } - let type = bleOutputProperty == .withoutResponse ? CBCharacteristicWriteType.withoutResponse : CBCharacteristicWriteType.withResponse - - if type == CBCharacteristicWriteType.withResponse { - if !gattCharacteristic.properties.contains(.write) { + switch bleOutputProperty { + case .withResponse: + guard gattCharacteristic.properties.contains(.write) else { completion(Result.failure(createFlutterError(code: .characteristicDoesNotSupportWrite, message: "Characteristic does not support write withResponse"))) return } @@ -459,26 +458,25 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral ) ) peripheral.writeValue(value.data, for: gattCharacteristic, type: .withResponse) - return - } - - if !gattCharacteristic.properties.contains(.writeWithoutResponse) { - completion(Result.failure(createFlutterError(code: .characteristicDoesNotSupportWriteWithoutResponse, message: "Characteristic does not support write withoutResponse"))) - return - } + case .withoutResponse: + guard gattCharacteristic.properties.contains(.writeWithoutResponse) else { + completion(Result.failure(createFlutterError(code: .characteristicDoesNotSupportWriteWithoutResponse, message: "Characteristic does not support write withoutResponse"))) + return + } - if peripheral.canSendWriteWithoutResponse { - peripheral.writeValue(value.data, for: gattCharacteristic, type: .withoutResponse) - completion(Result.success(())) - } else { - pendingWriteWithoutResponse.append( - PendingWriteWithoutResponse( - deviceId: deviceId, - characteristic: gattCharacteristic, - data: value.data, - result: completion + if peripheral.canSendWriteWithoutResponse { + peripheral.writeValue(value.data, for: gattCharacteristic, type: .withoutResponse) + completion(Result.success(())) + } else { + pendingWriteWithoutResponse.append( + PendingWriteWithoutResponse( + deviceId: deviceId, + characteristic: gattCharacteristic, + data: value.data, + result: completion + ) ) - ) + } } } diff --git a/lib/src/interfaces/universal_ble_platform_interface.dart b/lib/src/interfaces/universal_ble_platform_interface.dart index aacd62c4..a8c0d0f1 100644 --- a/lib/src/interfaces/universal_ble_platform_interface.dart +++ b/lib/src/interfaces/universal_ble_platform_interface.dart @@ -6,6 +6,9 @@ import 'package:universal_ble/src/utils/universal_logger.dart'; import 'package:universal_ble/universal_ble.dart'; abstract class UniversalBlePlatform { + /// Whether consecutive writes may be submitted before earlier writes finish. + bool get supportsWritePipelining => false; + // Do not use these directly to push updates OnScanResult? onScanResultUpdate; OnConnectionChange? onConnectionChange; diff --git a/lib/src/universal_ble.dart b/lib/src/universal_ble.dart index 23ee9c7c..3afccce1 100644 --- a/lib/src/universal_ble.dart +++ b/lib/src/universal_ble.dart @@ -347,7 +347,7 @@ class UniversalBle { // CoreBluetooth (and Chromium on Apple platforms) owns a native GATT // operation queue. Let consecutive writes fill it without letting either // writes or other operations cross a queue barrier. - await _bleCommandQueue.queueCommand( + await _bleCommandQueue.queueWrite( () => _platform.writeValue( deviceId, BleUuidParser.string(service), @@ -360,14 +360,10 @@ class UniversalBle { timeout: timeout, deviceId: deviceId, queueId: queueId, - canRunConcurrently: _usesNativeWriteQueue, + pipelined: _platform.supportsWritePipelining, ); } - static bool get _usesNativeWriteQueue => - defaultTargetPlatform == TargetPlatform.iOS || - defaultTargetPlatform == TargetPlatform.macOS; - /// Requests an MTU (Maximum Transmission Unit) value for the connection. /// /// **⚠️ Note:** Requesting an MTU is a *best-effort* operation. On many platforms diff --git a/lib/src/universal_ble_pigeon/universal_ble_pigeon_channel.dart b/lib/src/universal_ble_pigeon/universal_ble_pigeon_channel.dart index 133031c1..5d829d5d 100644 --- a/lib/src/universal_ble_pigeon/universal_ble_pigeon_channel.dart +++ b/lib/src/universal_ble_pigeon/universal_ble_pigeon_channel.dart @@ -16,6 +16,11 @@ class UniversalBlePigeonChannel extends UniversalBlePlatform final _channel = UniversalBlePlatformChannel(); + @override + bool get supportsWritePipelining => + defaultTargetPlatform == TargetPlatform.iOS || + defaultTargetPlatform == TargetPlatform.macOS; + @override Future getBluetoothAvailabilityState() => _executeWithErrorHandling(() => _channel.getBluetoothAvailabilityState()); diff --git a/lib/src/universal_ble_web/universal_ble_web.dart b/lib/src/universal_ble_web/universal_ble_web.dart index a118adc1..8b2cc50b 100644 --- a/lib/src/universal_ble_web/universal_ble_web.dart +++ b/lib/src/universal_ble_web/universal_ble_web.dart @@ -21,6 +21,11 @@ class UniversalBleWeb extends UniversalBlePlatform { final Map> _serviceCache = {}; bool _isScanning = false; + @override + bool get supportsWritePipelining => + defaultTargetPlatform == TargetPlatform.iOS || + defaultTargetPlatform == TargetPlatform.macOS; + @override Future getConnectionState(String deviceId) async { BluetoothDevice? device = _getDeviceById(deviceId); diff --git a/lib/src/utils/ble_command_queue.dart b/lib/src/utils/ble_command_queue.dart index 3badecb8..15fa51ac 100644 --- a/lib/src/utils/ble_command_queue.dart +++ b/lib/src/utils/ble_command_queue.dart @@ -16,11 +16,37 @@ class BleCommandQueue { String? deviceId, Duration? timeout, String? queueId, + }) => _queueCommand( + command, + deviceId: deviceId, + timeout: timeout, + queueId: queueId, + ); + + Future queueWrite( + Future Function() command, { + String? deviceId, + Duration? timeout, + String? queueId, + bool pipelined = false, + }) => _queueCommand( + command, + deviceId: deviceId, + timeout: timeout, + queueId: queueId, + canRunConcurrently: pipelined, + ); + + Future _queueCommand( + Future Function() command, { + String? deviceId, + Duration? timeout, + String? queueId, bool canRunConcurrently = false, }) { Duration? timeoutDuration = timeout ?? this.timeout; if (timeoutDuration == null) { - return queueCommandWithoutTimeout( + return _queueCommandWithoutTimeout( command, deviceId: deviceId, queueId: queueId, @@ -48,6 +74,16 @@ class BleCommandQueue { Future Function() command, { String? deviceId, String? queueId, + }) => _queueCommandWithoutTimeout( + command, + deviceId: deviceId, + queueId: queueId, + ); + + Future _queueCommandWithoutTimeout( + Future Function() command, { + String? deviceId, + String? queueId, bool canRunConcurrently = false, }) { return switch (queueType) { diff --git a/test/universal_ble_test_mock.dart b/test/universal_ble_test_mock.dart index c1f6fdb4..f106f757 100644 --- a/test/universal_ble_test_mock.dart +++ b/test/universal_ble_test_mock.dart @@ -75,6 +75,11 @@ abstract class UniversalBlePlatformMock extends UniversalBlePlatform { throw UnimplementedError(); } + @override + Future readRssi(String deviceId) { + throw UnimplementedError(); + } + @override Future requestConnectionPriority( String deviceId, diff --git a/test/universal_ble_write_queue_test.dart b/test/universal_ble_write_queue_test.dart index 6c1d5d1c..f91774f4 100644 --- a/test/universal_ble_write_queue_test.dart +++ b/test/universal_ble_write_queue_test.dart @@ -1,6 +1,6 @@ import 'dart:async'; +import 'dart:typed_data'; -import 'package:flutter/foundation.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:universal_ble/universal_ble.dart'; @@ -14,12 +14,10 @@ void main() { tearDown(() { UniversalBle.clearQueue(); - debugDefaultTargetPlatformOverride = null; }); - test('pipelines consecutive Apple writes', () async { - debugDefaultTargetPlatformOverride = TargetPlatform.macOS; - final platform = _PendingWritePlatform(); + test('pipelines writes when supported by the platform', () async { + final platform = _PendingWritePlatform(supportsWritePipelining: true); UniversalBle.setInstance(platform); final first = _write(1); @@ -33,8 +31,7 @@ void main() { await second; }); - test('keeps Android writes strictly serialized', () async { - debugDefaultTargetPlatformOverride = TargetPlatform.android; + test('serializes writes when pipelining is unsupported', () async { final platform = _PendingWritePlatform(); UniversalBle.setInstance(platform); @@ -52,9 +49,8 @@ void main() { await second; }); - test('does not pipeline an Apple write across a queue barrier', () async { - debugDefaultTargetPlatformOverride = TargetPlatform.macOS; - final platform = _PendingWritePlatform(); + test('does not pipeline a write across a queue barrier', () async { + final platform = _PendingWritePlatform(supportsWritePipelining: true); UniversalBle.setInstance(platform); final read = UniversalBle.read('device', '180a', '202a'); @@ -77,6 +73,11 @@ Future _write(int value) => UniversalBle.write('device', '180a', '202a', Uint8List.fromList([value])); class _PendingWritePlatform extends UniversalBlePlatformMock { + @override + final bool supportsWritePipelining; + + _PendingWritePlatform({this.supportsWritePipelining = false}); + final started = []; final pending = >[]; final readPending = Completer(); @@ -93,9 +94,6 @@ class _PendingWritePlatform extends UniversalBlePlatformMock { return readPending.future; } - @override - Future readRssi(String deviceId) => throw UnimplementedError(); - @override Future writeValue( String deviceId,