diff --git a/packages/flutter_reactive_ble/lib/src/connected_device_operation.dart b/packages/flutter_reactive_ble/lib/src/connected_device_operation.dart index de8eb833d..a6d6be954 100644 --- a/packages/flutter_reactive_ble/lib/src/connected_device_operation.dart +++ b/packages/flutter_reactive_ble/lib/src/connected_device_operation.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:flutter_reactive_ble/flutter_reactive_ble.dart'; import 'package:meta/meta.dart'; import 'package:reactive_ble_platform_interface/reactive_ble_platform_interface.dart'; @@ -22,6 +24,11 @@ abstract class ConnectedDeviceOperation { Future isDisconnected, ); + Future>> subscribeToCharacteristicAsync( + CharacteristicInstance characteristic, + Future isDisconnected, + ); + Future requestMtu(String deviceId, int mtu); Future> discoverServices(String deviceId); @@ -101,6 +108,102 @@ class ConnectedDeviceOperationImpl implements ConnectedDeviceOperation { return autosubscribingRepeater.stream; } + @override + Future>> subscribeToCharacteristicAsync( + CharacteristicInstance characteristic, + Future isDisconnected, + ) async { + final specificCharacteristicValueStream = characteristicValueStream + .where((update) => update.characteristic == characteristic) + .map((update) => update.result.dematerialize()); + + StreamController>? controller; + StreamSubscription>? valueSubscription; + final pendingActions = >)>[]; + + void startListeningToValues(Stream> source) { + valueSubscription = source.listen( + (data) { + if (controller != null && controller.hasListener) { + controller.add(data); + } else { + pendingActions.add((c) => c.add(data)); + } + }, + onError: (Object error, StackTrace stackTrace) { + if (controller != null && controller.hasListener) { + controller.addError(error, stackTrace); + } else { + pendingActions.add((c) => c.addError(error, stackTrace)); + } + }, + onDone: () { + if (controller != null && controller.hasListener) { + controller.close(); + } else { + pendingActions.add((c) => c.close()); + } + }, + ); + } + + startListeningToValues(specificCharacteristicValueStream); + + try { + await _blePlatform + .subscribeToNotifications(characteristic) + .firstWhere((_) => true, orElse: () {}); + } catch (_) { + await valueSubscription?.cancel(); + rethrow; + } + + var isSubscribed = true; + + controller = StreamController>.broadcast( + onListen: () { + if (!isSubscribed) { + isSubscribed = true; + startListeningToValues( + _blePlatform + .subscribeToNotifications(characteristic) + .asyncExpand((_) => specificCharacteristicValueStream), + ); + return; + } + if (pendingActions.isNotEmpty) { + final actions = List.of(pendingActions); + pendingActions.clear(); + for (final action in actions) { + action(controller!); + } + } + }, + onCancel: () async { + isSubscribed = false; + pendingActions.clear(); + await valueSubscription?.cancel(); + valueSubscription = null; + await _blePlatform + .stopSubscribingToNotifications(characteristic) + .catchError((Object e) => + // ignore: avoid_print + print("Error unsubscribing from notifications: $e")); + }, + ); + + unawaited( + isDisconnected.then((_) async { + pendingActions.clear(); + await valueSubscription?.cancel(); + valueSubscription = null; + await controller?.close(); + }), + ); + + return controller.stream; + } + @override Future requestMtu(String deviceId, int mtu) async => _blePlatform.requestMtuSize(deviceId, mtu); diff --git a/packages/flutter_reactive_ble/lib/src/reactive_ble.dart b/packages/flutter_reactive_ble/lib/src/reactive_ble.dart index c3503c1a7..b07e5a3a4 100644 --- a/packages/flutter_reactive_ble/lib/src/reactive_ble.dart +++ b/packages/flutter_reactive_ble/lib/src/reactive_ble.dart @@ -412,6 +412,19 @@ class FlutterReactiveBle { yield* (await resolveSingle(characteristic)).subscribe(); } + /// Subscribes to updates from the characteristic specified, waiting for the + /// BLE subscription to be completed before returning. + /// + /// This stream terminates automatically when the device is disconnected. + /// + /// This method assumes there is a single characteristic with the ids specified in [characteristic]. If there are + /// multiple characteristics with the same id on a device, use [resolve] to find them all. Or use + /// [getDiscoveredServices] to select the [Service]s and [Characteristic]s you're interested in. + Future>> subscribeToCharacteristicAsync(QualifiedCharacteristic characteristic) async { + await initialize(); + return (await resolveSingle(characteristic)).subscribeAsync(); + } + Future> resolve(QualifiedCharacteristic characteristic) async { final services = await getDiscoveredServices(characteristic.deviceId); return services @@ -559,6 +572,29 @@ class Characteristic { ); } + /// Subscribes to updates from the characteristic specified, waiting for the + /// BLE subscription to be completed before returning. + /// + /// This stream terminates automatically when the device is disconnected. + Future>> subscribeAsync() async { + _assertValidity(); + + final isDisconnected = _lib.connectedDeviceStream + .where((update) => + update.deviceId == service.deviceId && + (update.connectionState == DeviceConnectionState.disconnecting || + update.connectionState == DeviceConnectionState.disconnected)) + .cast() + .firstWhere((_) => true, orElse: () {}); + + await _lib.initialize(); + + return _lib._connectedDeviceOperator.subscribeToCharacteristicAsync( + _ids, + isDisconnected, + ); + } + // A Characteristic becomes invalid when its device gets disconnected. After reconnecting to the device, // services and characteristics have to be rediscovered bool _valid = true; diff --git a/packages/flutter_reactive_ble/test/connected_device_operation_test.dart b/packages/flutter_reactive_ble/test/connected_device_operation_test.dart index 4320c9151..5ed2078a6 100644 --- a/packages/flutter_reactive_ble/test/connected_device_operation_test.dart +++ b/packages/flutter_reactive_ble/test/connected_device_operation_test.dart @@ -341,6 +341,130 @@ void main() { ])); }); }); + + group('Subscribe to characteristic async', () { + test('It waits for subscription to complete before returning stream', + () async { + final subscriptionCompleter = Completer(); + when(_blePlatform.subscribeToNotifications(charDevice)) + .thenAnswer((_) => subscriptionCompleter.future.asStream()); + when(_blePlatform.charValueUpdateStream) + .thenAnswer((_) => Stream.fromIterable([ + valueUpdate1, + valueUpdateOtherDevice, + valueUpdate2, + valueUpdateSameDeviceOtherChar, + ])); + + var completed = false; + final streamFuture = _sut + .subscribeToCharacteristicAsync( + charDevice, terminateCompleter.future) + .then((stream) { + completed = true; + return stream; + }); + + await Future.delayed(Duration.zero); + expect(completed, isFalse); + + subscriptionCompleter.complete(); + final stream = await streamFuture; + expect(completed, isTrue); + + expect( + stream, + emitsInOrder(>[ + [1], + [2], + ]), + ); + }); + + test( + 'It buffers notifications emitted before the returned stream is listened to', + () async { + final charValueController = + StreamController.broadcast(); + addTearDown(charValueController.close); + + when(_blePlatform.charValueUpdateStream) + .thenAnswer((_) => charValueController.stream); + + final stream = await _sut.subscribeToCharacteristicAsync( + charDevice, terminateCompleter.future); + + // Emit a notification before any listener is attached to `stream` + charValueController.add(valueUpdate1); + await Future.delayed(Duration.zero); + + final received = >[]; + final subscription = stream.listen(received.add); + addTearDown(subscription.cancel); + + charValueController.add(valueUpdate2); + await Future.delayed(Duration.zero); + + expect(received, >[ + [1], + [2], + ]); + }); + + test( + 'It stops subscribing to notifications when stream subscription is cancelled', + () async { + final charValueController = + StreamController.broadcast(); + addTearDown(charValueController.close); + + when(_blePlatform.charValueUpdateStream) + .thenAnswer((_) => charValueController.stream); + + final stream = await _sut.subscribeToCharacteristicAsync( + charDevice, terminateCompleter.future); + + final subscription = stream.listen((_) {}); + await subscription.cancel(); + + verify(_blePlatform.stopSubscribingToNotifications(charDevice)) + .called(1); + }); + + test('It closes stream when device disconnects', () async { + final charValueController = + StreamController.broadcast(); + addTearDown(charValueController.close); + + when(_blePlatform.charValueUpdateStream) + .thenAnswer((_) => charValueController.stream); + + final stream = await _sut.subscribeToCharacteristicAsync( + charDevice, terminateCompleter.future); + + terminateCompleter.complete(const ConnectionStateUpdate( + deviceId: '123', + connectionState: DeviceConnectionState.disconnected, + failure: null, + )); + + await expectLater(stream, emitsDone); + }); + + test('It propagates error if subscribeToNotifications fails', + () async { + when(_blePlatform.subscribeToNotifications(charDevice)) + .thenAnswer((_) => Stream.error(Exception('BLE failure'))); + when(_blePlatform.charValueUpdateStream) + .thenAnswer((_) => const Stream.empty()); + + await expectLater( + _sut.subscribeToCharacteristicAsync( + charDevice, terminateCompleter.future), + throwsException, + ); + }); + }); }); group('Negotiate mtusize', () { diff --git a/packages/flutter_reactive_ble/test/reactive_ble_test.dart b/packages/flutter_reactive_ble/test/reactive_ble_test.dart index 1f65fee5d..8a2bdaddf 100644 --- a/packages/flutter_reactive_ble/test/reactive_ble_test.dart +++ b/packages/flutter_reactive_ble/test/reactive_ble_test.dart @@ -589,6 +589,87 @@ void main() { }); }); + group('Subscribe to characteristic async', () { + const update = ConnectionStateUpdate( + deviceId: '123', + connectionState: DeviceConnectionState.disconnected, + failure: null, + ); + + late QualifiedCharacteristic char; + late Stream> valueStream; + late Stream?> resultStream; + + setUp(() async { + char = _createChar(); + when(_deviceConnector.deviceConnectionStateUpdateStream).thenAnswer((_) => Stream.fromIterable([update])); + when(_deviceOperation.getDiscoverServices(char.deviceId)).thenAnswer((_) async => [ + DiscoveredService( + serviceId: char.serviceId, + serviceInstanceId: "11", + characteristicIds: [char.characteristicId], + includedServices: [], + characteristics: [ + DiscoveredCharacteristic( + characteristicId: Uuid.parse("1234"), + characteristicInstanceId: "101", + serviceId: char.serviceId, + isReadable: true, + isWritableWithResponse: true, + isWritableWithoutResponse: true, + isNotifiable: true, + isIndicatable: true, + ), + DiscoveredCharacteristic( + characteristicId: char.characteristicId, + characteristicInstanceId: "101", + serviceId: char.serviceId, + isReadable: true, + isWritableWithResponse: true, + isWritableWithoutResponse: true, + isNotifiable: true, + isIndicatable: true, + ), + ], + ) + ]); + + valueStream = Stream.fromIterable([ + [1], + [2] + ]); + + when(_deviceOperation.subscribeToCharacteristicAsync(any, any)).thenAnswer((_) async => valueStream); + + resultStream = await _sut.subscribeToCharacteristicAsync(char); + }); + + test("It subscribes to the correct characteristic", () { + verify(_deviceOperation.subscribeToCharacteristicAsync( + CharacteristicInstance( + characteristicId: char.characteristicId, + characteristicInstanceId: "101", + serviceId: char.serviceId, + serviceInstanceId: "11", + deviceId: char.deviceId, + ), + any, + )).called(1); + }); + + test('It emits correct value', () { + expect( + resultStream, + emitsInOrder( + >[ + [1], + [2] + ], + ), + ); + }); + }); + group('Discover Services', () { const deviceId = '123'; diff --git a/packages/flutter_reactive_ble/test/reactive_ble_test.mocks.dart b/packages/flutter_reactive_ble/test/reactive_ble_test.mocks.dart index 030c1276a..c290086f5 100644 --- a/packages/flutter_reactive_ble/test/reactive_ble_test.mocks.dart +++ b/packages/flutter_reactive_ble/test/reactive_ble_test.mocks.dart @@ -423,6 +423,22 @@ class MockConnectedDeviceOperation extends _i1.Mock returnValue: _i4.Stream>.empty(), ) as _i4.Stream>); @override + _i4.Future<_i4.Stream>> subscribeToCharacteristicAsync( + _i2.CharacteristicInstance? characteristic, + _i4.Future? isDisconnected, + ) => + (super.noSuchMethod( + Invocation.method( + #subscribeToCharacteristicAsync, + [ + characteristic, + isDisconnected, + ], + ), + returnValue: + _i4.Future<_i4.Stream>>.value(_i4.Stream>.empty()), + ) as _i4.Future<_i4.Stream>>); + @override _i4.Future requestMtu( String? deviceId, int? mtu, diff --git a/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/PluginController.kt b/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/PluginController.kt index 22468cc26..1172aa9ac 100644 --- a/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/PluginController.kt +++ b/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/PluginController.kt @@ -287,8 +287,15 @@ class PluginController { result: Result, ) { val request = pb.NotifyCharacteristicRequest.parseFrom(call.arguments as ByteArray) - charNotificationHandler.subscribeToNotifications(request) - result.success(null) + charNotificationHandler.subscribeToNotifications( + request, + onSubscriptionCompleted = { + result.success(null) + }, + onSubscriptionError = { error -> + result.error("char_notification_error", error.message, null) + }, + ) } private fun stopNotifications( diff --git a/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/ble/BleClient.kt b/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/ble/BleClient.kt index 758709414..24904635d 100644 --- a/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/ble/BleClient.kt +++ b/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/ble/BleClient.kt @@ -45,7 +45,7 @@ interface BleClient { deviceId: String, characteristicId: UUID, characteristicInstanceId: Int, - ): Observable + ): Observable> fun writeCharacteristicWithResponse( deviceId: String, diff --git a/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/ble/ReactiveBleClient.kt b/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/ble/ReactiveBleClient.kt index 9fa076c1a..2cfe6e098 100644 --- a/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/ble/ReactiveBleClient.kt +++ b/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/ble/ReactiveBleClient.kt @@ -232,7 +232,7 @@ open class ReactiveBleClient(private val context: Context) : BleClient { deviceId: String, characteristicId: UUID, characteristicInstanceId: Int, - ): Observable { + ): Observable> { return getConnection(deviceId) .flatMap { deviceConnection -> setupNotificationOrIndication( @@ -241,10 +241,6 @@ open class ReactiveBleClient(private val context: Context) : BleClient { characteristicInstanceId, ) } - // now we have setup the subscription and we want the actual value - .flatMap { notificationObservable -> - notificationObservable - } } override fun negotiateMtuSize( diff --git a/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/channelhandlers/CharNotificationHandler.kt b/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/channelhandlers/CharNotificationHandler.kt index a699eb58c..fbf1360ff 100644 --- a/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/channelhandlers/CharNotificationHandler.kt +++ b/packages/reactive_ble_mobile/android/src/main/kotlin/com/signify/hue/flutterreactiveble/channelhandlers/CharNotificationHandler.kt @@ -32,20 +32,43 @@ class CharNotificationHandler(private val bleClient: com.signify.hue.flutterreac unsubscribeFromAllNotifications() } - fun subscribeToNotifications(request: pb.NotifyCharacteristicRequest) { + fun subscribeToNotifications( + request: pb.NotifyCharacteristicRequest, + onSubscriptionCompleted: () -> Unit = {}, + onSubscriptionError: (Throwable) -> Unit = {}, + ) { val charUuid = uuidConverter .uuidFromByteArray(request.characteristic.characteristicUuid.data.toByteArray()) + var isCompleted = false + fun completeSubscription() { + if (!isCompleted) { + isCompleted = true + onSubscriptionCompleted() + } + } + subscriptionMap.remove(request.characteristic)?.dispose() val subscription = bleClient.setupNotification( request.characteristic.deviceId, charUuid, request.characteristic.characteristicInstanceId.toInt(), ) + .observeOn(AndroidSchedulers.mainThread()) + .doOnDispose(::completeSubscription) + .flatMap { notificationObservable -> + notificationObservable + .observeOn(AndroidSchedulers.mainThread()) + .doOnSubscribe { completeSubscription() } + } .observeOn(AndroidSchedulers.mainThread()) .subscribe({ value -> handleNotificationValue(request.characteristic, value) }, { + if (!isCompleted) { + isCompleted = true + onSubscriptionError(it) + } when (it) { is BleDisconnectedException -> { subscriptionMap.remove(request.characteristic)?.dispose() diff --git a/packages/reactive_ble_mobile/android/src/test/kotlin/com/signify/hue/flutterreactiveble/channelhandlers/CharNotificationHandlerTest.kt b/packages/reactive_ble_mobile/android/src/test/kotlin/com/signify/hue/flutterreactiveble/channelhandlers/CharNotificationHandlerTest.kt new file mode 100644 index 000000000..7d1494725 --- /dev/null +++ b/packages/reactive_ble_mobile/android/src/test/kotlin/com/signify/hue/flutterreactiveble/channelhandlers/CharNotificationHandlerTest.kt @@ -0,0 +1,136 @@ +package com.signify.hue.flutterreactiveble.channelhandlers + +import com.google.common.truth.Truth.assertThat +import com.google.protobuf.ByteString +import com.signify.hue.flutterreactiveble.ble.BleClient +import io.flutter.plugin.common.EventChannel +import io.mockk.MockKAnnotations +import io.mockk.every +import io.mockk.impl.annotations.MockK +import io.mockk.verify +import io.reactivex.Observable +import io.reactivex.android.plugins.RxAndroidPlugins +import io.reactivex.schedulers.Schedulers +import io.reactivex.subjects.PublishSubject +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test +import java.nio.ByteBuffer +import java.util.UUID +import com.signify.hue.flutterreactiveble.ProtobufModel as pb + +class CharNotificationHandlerTest { + @MockK + private lateinit var bleClient: BleClient + + @MockK(relaxed = true) + private lateinit var eventSink: EventChannel.EventSink + + private lateinit var sut: CharNotificationHandler + + @BeforeEach + fun setup() { + MockKAnnotations.init(this) + RxAndroidPlugins.setMainThreadSchedulerHandler { Schedulers.trampoline() } + RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() } + sut = CharNotificationHandler(bleClient) + sut.onListen(null, eventSink) + } + + @AfterEach + fun teardown() { + sut.onCancel(null) + RxAndroidPlugins.reset() + } + + @Test + fun `completes subscription only after setupNotification emits inner observable`() { + val outerSubject = PublishSubject.create>() + val innerSubject = PublishSubject.create() + + every { bleClient.setupNotification(any(), any(), any()) }.returns(outerSubject) + + val uuid = UUID.randomUUID() + val uuidBytes = + ByteBuffer.wrap(ByteArray(16)) + .putLong(uuid.mostSignificantBits) + .putLong(uuid.leastSignificantBits) + .array() + + val charAddress = + pb.CharacteristicAddress.newBuilder() + .setDeviceId("device1") + .setServiceUuid(pb.Uuid.newBuilder().setData(ByteString.copyFrom(uuidBytes))) + .setServiceInstanceId("10") + .setCharacteristicUuid(pb.Uuid.newBuilder().setData(ByteString.copyFrom(uuidBytes))) + .setCharacteristicInstanceId("11") + .build() + + val request = + pb.NotifyCharacteristicRequest.newBuilder() + .setCharacteristic(charAddress) + .build() + + var completed = false + sut.subscribeToNotifications( + request, + onSubscriptionCompleted = { + completed = true + }, + ) + + assertThat(completed).isFalse() + + outerSubject.onNext(innerSubject) + assertThat(completed).isTrue() + + innerSubject.onNext(byteArrayOf(1, 2, 3)) + verify(exactly = 1) { eventSink.success(any()) } + } + + @Test + fun `reports subscription error if setupNotification fails before emitting inner observable`() { + val outerSubject = PublishSubject.create>() + + every { bleClient.setupNotification(any(), any(), any()) }.returns(outerSubject) + + val uuid = UUID.randomUUID() + val uuidBytes = + ByteBuffer.wrap(ByteArray(16)) + .putLong(uuid.mostSignificantBits) + .putLong(uuid.leastSignificantBits) + .array() + + val charAddress = + pb.CharacteristicAddress.newBuilder() + .setDeviceId("device1") + .setServiceUuid(pb.Uuid.newBuilder().setData(ByteString.copyFrom(uuidBytes))) + .setServiceInstanceId("10") + .setCharacteristicUuid(pb.Uuid.newBuilder().setData(ByteString.copyFrom(uuidBytes))) + .setCharacteristicInstanceId("11") + .build() + + val request = + pb.NotifyCharacteristicRequest.newBuilder() + .setCharacteristic(charAddress) + .build() + + var completed = false + var receivedError: Throwable? = null + sut.subscribeToNotifications( + request, + onSubscriptionCompleted = { + completed = true + }, + onSubscriptionError = { error -> + receivedError = error + }, + ) + + val expectedError = IllegalStateException("Notification setup failed") + outerSubject.onError(expectedError) + + assertThat(completed).isFalse() + assertThat(receivedError).isEqualTo(expectedError) + } +}