diff --git a/lib/src/managers/ble_manager.dart b/lib/src/managers/ble_manager.dart index 709d01c..ca81d66 100644 --- a/lib/src/managers/ble_manager.dart +++ b/lib/src/managers/ble_manager.dart @@ -396,7 +396,14 @@ class BleManager extends BleGattManager { } streamController.onCancel = () async { - if (_streamControllers.containsKey(streamIdentifier)) { + // A canceled controller may finish closing after a replacement + // subscription has already installed a new controller for the same + // characteristic. Only the controller that still owns the map entry may + // tear down the native notification subscription. + if (identical( + _streamControllers[streamIdentifier], + streamController, + )) { final canceledController = _streamControllers.remove(streamIdentifier); _subscriptionSetups.remove(streamIdentifier); if (canceledController != null && !canceledController.isClosed) { diff --git a/lib/src/models/devices/open_earable_v2_audio_response_manager.dart b/lib/src/models/devices/open_earable_v2_audio_response_manager.dart index 29f61cd..ffed8cb 100644 --- a/lib/src/models/devices/open_earable_v2_audio_response_manager.dart +++ b/lib/src/models/devices/open_earable_v2_audio_response_manager.dart @@ -24,7 +24,7 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager { /// Maximum time to wait for each protocol notification. final Duration notificationTimeout; - bool _operationInProgress = false; + Future _operationQueue = Future.value(); @override Future uploadAudioBuffer({ @@ -79,7 +79,9 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager { try { reportProgress(AudioResponseUploadPhase.starting, 0); - var statusFuture = _nextTransferStatus(statusIterator, transferId); + var statusFuture = _guardPendingFuture( + _nextTransferStatus(statusIterator, transferId), + ); await _write( AudioResponseBleUuids.transferControlCharacteristicUuid, AudioResponseTransferControl.start( @@ -117,7 +119,9 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager { continue; } - statusFuture = _nextTransferStatus(statusIterator, transferId); + statusFuture = _guardPendingFuture( + _nextTransferStatus(statusIterator, transferId), + ); for (var credit = 0; credit < status.credits && sentOffset < samples.length; credit++) { @@ -160,7 +164,9 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager { AudioResponseUploadPhase.committing, lastAcknowledgedSamples, ); - statusFuture = _nextTransferStatus(statusIterator, transferId); + statusFuture = _guardPendingFuture( + _nextTransferStatus(statusIterator, transferId), + ); await _write( AudioResponseBleUuids.transferControlCharacteristicUuid, AudioResponseTransferControl.commit( @@ -196,7 +202,9 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager { ); try { - final resultFuture = _nextResult(resultIterator, config.id); + final resultFuture = _guardPendingFuture( + _nextResult(resultIterator, config.id), + ); await _write( AudioResponseBleUuids.configCharacteristicUuid, config.toBytes(), @@ -210,17 +218,24 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager { }); } - Future _runExclusive(Future Function() operation) async { - if (_operationInProgress) { - throw StateError('An audio response operation is already in progress'); - } + Future _runExclusive(Future Function() operation) { + final result = Completer(); + _operationQueue = _operationQueue.then((_) async { + try { + result.complete(await operation()); + } on Object catch (error, stackTrace) { + result.completeError(error, stackTrace); + } + }); + return result.future; + } - _operationInProgress = true; - try { - return await operation(); - } finally { - _operationInProgress = false; - } + Future _guardPendingFuture(Future future) { + // Notifications must be awaited before the corresponding write so a fast + // response cannot be missed. If that write fails, keep the pending wait + // from surfacing a second, detached asynchronous error during cleanup. + future.ignore(); + return future; } Future _write( diff --git a/test/ble_manager_subscription_test.dart b/test/ble_manager_subscription_test.dart new file mode 100644 index 0000000..faf7839 --- /dev/null +++ b/test/ble_manager_subscription_test.dart @@ -0,0 +1,82 @@ +import 'dart:async'; +import 'dart:typed_data'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:open_earable_flutter/src/managers/ble_manager.dart'; +import 'package:universal_ble/universal_ble.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + test('a stale controller cannot tear down its replacement subscription', + () async { + const deviceId = 'device'; + const serviceId = '7467b395-9043-4453-bc5c-2d8e8b10680a'; + const characteristicId = '7467b39c-9043-4453-bc5c-2d8e8b10680a'; + final platform = _FakeUniversalBlePlatform(); + UniversalBle.setInstance(platform); + final manager = BleManager(); + + platform.updateConnection(deviceId, true); + final oldStream = await manager.subscribe( + deviceId: deviceId, + serviceId: serviceId, + characteristicId: characteristicId, + ); + final oldSubscription = oldStream.listen((_) {}); + oldSubscription.pause(); + + // Closing a paused stream defers its onCancel callback. Reconnect and + // subscribe again before that callback is allowed to run. + platform.updateConnection(deviceId, false); + platform.updateConnection(deviceId, true); + final replacementStream = await manager.subscribe( + deviceId: deviceId, + serviceId: serviceId, + characteristicId: characteristicId, + ); + final replacementValue = Completer>(); + final replacementSubscription = replacementStream.listen( + replacementValue.complete, + ); + + oldSubscription.resume(); + await oldSubscription.asFuture(); + await oldSubscription.cancel(); + await Future.delayed(Duration.zero); + + platform.updateCharacteristicValue( + deviceId, + characteristicId, + Uint8List.fromList([7]), + null, + ); + + expect(await replacementValue.future, [7]); + expect( + platform.notificationChanges, + [ + BleInputProperty.notification, + BleInputProperty.notification, + ], + ); + await replacementSubscription.cancel(); + }); +} + +class _FakeUniversalBlePlatform extends UniversalBlePlatform { + final List notificationChanges = []; + + @override + Future setNotifiable( + String deviceId, + String service, + String characteristic, + BleInputProperty bleInputProperty, + ) async { + notificationChanges.add(bleInputProperty); + } + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} diff --git a/test/open_earable_v2_audio_response_manager_test.dart b/test/open_earable_v2_audio_response_manager_test.dart index 3adefe1..959d59a 100644 --- a/test/open_earable_v2_audio_response_manager_test.dart +++ b/test/open_earable_v2_audio_response_manager_test.dart @@ -176,6 +176,72 @@ void main() { expect(result.response, [10, 20]); }); + test('queues an operation requested while another is finishing', () async { + final bleManager = _FakeBleGattManager(); + final manager = OpenEarableV2AudioResponseManager( + bleManager: bleManager, + deviceId: 'device', + ); + final firstWrite = Completer(); + final configWrites = []; + + bleManager.onWrite = (write) { + if (write.characteristicId != + AudioResponseBleUuids.configCharacteristicUuid) { + return; + } + final config = AudioResponseConfig.fromBytes(write.bytes); + configWrites.add(config.id); + if (config.id == 1) { + firstWrite.complete(); + return; + } + bleManager.emitResult( + AudioResponseResult( + id: config.id, + points: 1, + frequencies: [100], + response: [config.id], + ), + ); + }; + + final firstResult = manager.measureAudioResponse( + AudioResponseConfig( + id: 1, + transfer_id: 42, + volume: 0.5, + points: 1, + frequencies: [100], + ), + ); + await firstWrite.future; + final secondResult = manager.measureAudioResponse( + AudioResponseConfig( + id: 2, + transfer_id: 42, + volume: 0.5, + points: 1, + frequencies: [100], + ), + ); + await Future.delayed(Duration.zero); + + expect(configWrites, [1]); + bleManager.emitResult( + AudioResponseResult( + id: 1, + points: 1, + frequencies: [100], + response: [1], + ), + ); + + expect((await firstResult).id, 1); + expect((await secondResult).id, 2); + expect(configWrites, [1, 2]); + }); + test('allows device default measurement points', () async { final bleManager = _FakeBleGattManager(); final manager = OpenEarableV2AudioResponseManager( @@ -392,6 +458,53 @@ void main() { .map((write) => AudioResponseTransferControl.fromBytes(write.bytes)); expect(controls.map(_controlType), [0, 2]); }); + + test('does not leak a pending status error when a write fails', () async { + final bleManager = _FakeBleGattManager(); + final manager = OpenEarableV2AudioResponseManager( + bleManager: bleManager, + deviceId: 'device', + ); + var failStartWrite = true; + + bleManager.onWrite = (write) { + if (!failStartWrite || + write.characteristicId != + AudioResponseBleUuids.transferControlCharacteristicUuid) { + return; + } + final control = AudioResponseTransferControl.fromBytes(write.bytes); + if (_controlType(control) == 0) { + failStartWrite = false; + throw StateError('start write failed'); + } + }; + + await expectLater( + manager.uploadAudioBuffer( + transferId: 42, + samples: const [0], + samplingRate: 16000, + ), + throwsA( + isA().having( + (error) => error.message, + 'message', + 'start write failed', + ), + ), + ); + await Future.delayed(Duration.zero); + + final controls = bleManager.writes + .where( + (write) => + write.characteristicId == + AudioResponseBleUuids.transferControlCharacteristicUuid, + ) + .map((write) => AudioResponseTransferControl.fromBytes(write.bytes)); + expect(controls.map(_controlType), [0, 2]); + }); }); }