Add support for handling observation exceptions (#254)

Author: twyatt

README.md

@@ -377,7 +377,17 @@ platform characteristic will be used to to resume streaming characteristic chang
 
 Failures related to notifications/indications are propagated via the [`observe`] [`Flow`], for example, if the
 associated characteristic is invalid or cannot be found, then a `NoSuchElementException` is propagated via the
-[`observe`] [`Flow`].
+[`observe`] [`Flow`]. An [`ObservationExceptionHandler`] may be registered with the [`Peripheral`] to control which
+failures are propagated through (and terminate) the [`observe`] [`Flow`], for example:
+
+```kotlin
+scope.peripheral(advertisement) {
+    observationExceptionHandler { cause ->
+        // Log failure instead of propagating associated `observe` flow.
+        println("Observation failure suppressed: $cause")
+    }
+}
+```
 
 In scenarios where an I/O operation needs to be performed upon subscribing to the [`observe`] [`Flow`], an `onSubscribe`
 action may be specified:

core/api/core.api

@@ -191,6 +191,10 @@ public final class com/juul/kable/NotReadyException : com/juul/kable/NotConnecte
 	public fun <init> ()V
 }
 
+public final class com/juul/kable/ObservationExceptionPeripheral {
+	public final fun getState ()Lkotlinx/coroutines/flow/StateFlow;
+}
+
 public abstract interface annotation class com/juul/kable/ObsoleteKableApi : java/lang/annotation/Annotation {
 }
 
@@ -219,6 +223,7 @@ public final class com/juul/kable/PeripheralBuilder {
 	public final fun getPhy ()Lcom/juul/kable/Phy;
 	public final fun getTransport ()Lcom/juul/kable/Transport;
 	public final fun logging (Lkotlin/jvm/functions/Function1;)V
+	public final fun observationExceptionHandler (Lkotlin/jvm/functions/Function3;)V
 	public final fun onServicesDiscovered (Lkotlin/jvm/functions/Function2;)V
 	public final fun setPhy (Lcom/juul/kable/Phy;)V
 	public final fun setTransport (Lcom/juul/kable/Transport;)V

core/src/androidMain/kotlin/Peripheral.kt

@@ -117,6 +117,7 @@ public fun CoroutineScope.peripheral(
         bluetoothDevice,
         builder.transport,
         builder.phy,
+        builder.observationExceptionHandler,
         builder.onServicesDiscovered,
         builder.logging,
     )
@@ -129,6 +130,7 @@ public class AndroidPeripheral internal constructor(
     private val bluetoothDevice: BluetoothDevice,
     private val transport: Transport,
     private val phy: Phy,
+    observationExceptionHandler: ObservationExceptionHandler,
     private val onServicesDiscovered: ServicesDiscoveredAction,
     private val logging: Logging,
 ) : Peripheral {
@@ -167,7 +169,7 @@ public class AndroidPeripheral internal constructor(
      */
     public val mtu: StateFlow<Int?> = _mtu.asStateFlow()
 
-    private val observers = Observers<ByteArray>(this, logging)
+    private val observers = Observers<ByteArray>(this, logging, exceptionHandler = observationExceptionHandler)
 
     @Volatile
     private var _discoveredServices: List<DiscoveredService>? = null

core/src/androidMain/kotlin/PeripheralBuilder.kt

@@ -87,6 +87,11 @@ public actual class PeripheralBuilder internal actual constructor() {
         onServicesDiscovered = action
     }
 
+    internal var observationExceptionHandler: ObservationExceptionHandler = { cause -> throw cause }
+    public actual fun observationExceptionHandler(handler: ObservationExceptionHandler) {
+        observationExceptionHandler = handler
+    }
+
     /** Preferred transport for GATT connections to remote dual-mode devices. */
     public var transport: Transport = Transport.Le
 

core/src/appleMain/kotlin/Peripheral.kt

@@ -81,6 +81,7 @@ public actual fun CoroutineScope.peripheral(
     return ApplePeripheral(
         coroutineContext,
         advertisement.cbPeripheral,
+        builder.observationExceptionHandler,
         builder.onServicesDiscovered,
         builder.logging,
     )
@@ -90,6 +91,7 @@ public actual fun CoroutineScope.peripheral(
 public class ApplePeripheral internal constructor(
     parentCoroutineContext: CoroutineContext,
     private val cbPeripheral: CBPeripheral,
+    observationExceptionHandler: ObservationExceptionHandler,
     private val onServicesDiscovered: ServicesDiscoveredAction,
     private val logging: Logging,
 ) : Peripheral {
@@ -105,7 +107,7 @@ public class ApplePeripheral internal constructor(
     private val _state = MutableStateFlow<State>(State.Disconnected())
     override val state: StateFlow<State> = _state.asStateFlow()
 
-    private val observers = Observers<NSData>(this, logging)
+    private val observers = Observers<NSData>(this, logging, exceptionHandler = observationExceptionHandler)
 
     internal val platformIdentifier = cbPeripheral.identifier.UUIDString
 

core/src/appleMain/kotlin/PeripheralBuilder.kt

@@ -49,4 +49,9 @@ public actual class PeripheralBuilder internal actual constructor() {
     public actual fun onServicesDiscovered(action: ServicesDiscoveredAction) {
         onServicesDiscovered = action
     }
+
+    internal var observationExceptionHandler: ObservationExceptionHandler = { cause -> throw cause }
+    public actual fun observationExceptionHandler(handler: ObservationExceptionHandler) {
+        observationExceptionHandler = handler
+    }
 }

core/src/commonMain/kotlin/ObservationEvent.kt

@@ -11,15 +11,10 @@ internal sealed class ObservationEvent<T> {
 
     data class Error<T>(
         override val characteristic: Characteristic,
-        val cause: Throwable,
+        val cause: Exception,
     ) : ObservationEvent<T>()
 }
 
-internal fun <T> dematerialize(event: ObservationEvent<T>): T = when (event) {
-    is ObservationEvent.Error -> throw event.cause
-    is ObservationEvent.CharacteristicChange -> event.data
-}
-
 internal fun <T> ObservationEvent<T>.isAssociatedWith(characteristic: Characteristic): Boolean =
     this.characteristic.characteristicUuid == characteristic.characteristicUuid &&
         this.characteristic.serviceUuid == characteristic.serviceUuid

core/src/commonMain/kotlin/Observers.kt

@@ -2,12 +2,15 @@ package com.juul.kable
 
 import co.touchlab.stately.collections.IsoMutableList
 import co.touchlab.stately.isolate.IsolateState
+import com.juul.kable.ObservationEvent.CharacteristicChange
+import com.juul.kable.ObservationEvent.Error
 import com.juul.kable.logs.Logging
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.filter
-import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.mapNotNull
 import kotlinx.coroutines.flow.onCompletion
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.onSubscription
 import kotlin.coroutines.cancellation.CancellationException
 
@@ -38,6 +41,7 @@ internal class Observers<T>(
     private val peripheral: Peripheral,
     private val logging: Logging,
     extraBufferCapacity: Int = 0,
+    private val exceptionHandler: ObservationExceptionHandler,
 ) {
 
     val characteristicChanges = MutableSharedFlow<ObservationEvent<T>>(extraBufferCapacity = extraBufferCapacity)
@@ -60,10 +64,26 @@ internal class Observers<T>(
         }
 
         return characteristicChanges
-            .onSubscription { observation.onSubscription(onSubscription) }
+            .onSubscription {
+                try {
+                    observation.onSubscription(onSubscription)
+                } catch (e: Exception) {
+                    exceptionHandler(ObservationExceptionPeripheral(peripheral), e)
+                }
+            }
             .filter { event -> event.isAssociatedWith(characteristic) }
-            .map(::dematerialize)
-            .onCompletion { observation.onCompletion(onSubscription) }
+            .onEach { event ->
+                if (event is Error)
+                    exceptionHandler(ObservationExceptionPeripheral(peripheral), event.cause)
+            }
+            .mapNotNull { event -> (event as? CharacteristicChange)?.data }
+            .onCompletion {
+                try {
+                    observation.onCompletion(onSubscription)
+                } catch (e: Exception) {
+                    exceptionHandler(ObservationExceptionPeripheral(peripheral), e)
+                }
+            }
     }
 
     suspend fun onConnected() {
@@ -74,7 +94,7 @@ internal class Observers<T>(
             } catch (cancellation: CancellationException) {
                 throw cancellation
             } catch (e: Exception) {
-                characteristicChanges.emit(ObservationEvent.Error(characteristic, e))
+                characteristicChanges.emit(Error(characteristic, e))
             }
         }
     }

core/src/commonMain/kotlin/Peripheral.kt

@@ -176,8 +176,12 @@ public interface Peripheral {
      * **indication** (if [characteristic] supports both notifications and indications, then only **notification** is
      * used).
      *
-     * Failures related to notifications are propagated via the returned [observe] [Flow], for example, if the specified
+     * Failures related to observations are propagated via the returned [observe] [Flow], for example, if the specified
      * [characteristic] is invalid or cannot be found then returned [Flow] terminates with a [NoSuchElementException].
+     * An [ObservationExceptionHandler] may be registered with the [Peripheral] to control which failures are propagated
+     * through (and terminate) the observation [Flow]. When registered, only exceptions thrown from the
+     * [ObservationExceptionHandler] are propagated (and terminate) the returned observation [Flow]. See
+     * [PeripheralBuilder.observationExceptionHandler] for more details.
      *
      * The optional [onSubscription] parameter is functionally identical to using the
      * [onSubscription][kotlinx.coroutines.flow.onSubscription] operator on the returned [Flow] except it has the

core/src/commonMain/kotlin/PeripheralBuilder.kt

@@ -1,6 +1,7 @@
 package com.juul.kable
 
 import com.juul.kable.logs.LoggingBuilder
+import kotlinx.coroutines.flow.StateFlow
 
 public expect class ServicesDiscoveredPeripheral {
 
@@ -24,9 +25,37 @@ public expect class ServicesDiscoveredPeripheral {
     ): Unit
 }
 
+public class ObservationExceptionPeripheral internal constructor(peripheral: Peripheral) {
+    public val state: StateFlow<State> = peripheral.state
+}
+
 internal typealias ServicesDiscoveredAction = suspend ServicesDiscoveredPeripheral.() -> Unit
+internal typealias ObservationExceptionHandler = suspend ObservationExceptionPeripheral.(cause: Exception) -> Unit
 
 public expect class PeripheralBuilder internal constructor() {
     public fun logging(init: LoggingBuilder)
     public fun onServicesDiscovered(action: ServicesDiscoveredAction)
+
+    /**
+     * Registers an [ObservationExceptionHandler] for the [Peripheral]. When registered, observation failures are
+     * passed to the [ObservationExceptionHandler] instead of through [observation][Peripheral.observe] flows. Any
+     * exceptions in the [ObservationExceptionHandler] will propagate through (and terminate) the associated
+     * [observation][Peripheral.observe] flow.
+     *
+     * Some failures are due to connection drops before the connection state has propagated from the system, the
+     * [ObservationExceptionHandler] can be useful for ignoring failures that precursor a connection drop:
+     *
+     * ```
+     * scope.peripheral(advertisement) {
+     *     observationExceptionHandler { cause ->
+     *         // Only propagate failure if we don't see a disconnect within a second.
+     *         withTimeoutOrNull(1_000L) {
+     *             state.first { it is Disconnected }
+     *         } ?: throw IllegalStateException("Observation failure occurred.", cause)
+     *         println("Ignored failure associated with disconnect: $cause")
+     *     }
+     * }
+     * ```
+     */
+    public fun observationExceptionHandler(handler: ObservationExceptionHandler)
 }