Optimize observation handling

core/api/core.api

@@ -17,7 +17,7 @@ public final class com/juul/kable/AndroidPeripheral : com/juul/kable/Peripheral
 	public fun disconnect (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public final fun getMtu ()Lkotlinx/coroutines/flow/StateFlow;
 	public fun getServices ()Ljava/util/List;
-	public fun getState ()Lkotlinx/coroutines/flow/Flow;
+	public fun getState ()Lkotlinx/coroutines/flow/StateFlow;
 	public fun observe (Lcom/juul/kable/Characteristic;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
 	public fun read (Lcom/juul/kable/Characteristic;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun read (Lcom/juul/kable/Descriptor;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
@@ -58,7 +58,7 @@ public final class com/juul/kable/CharacteristicKt {
 	public static final fun characteristicOf (Ljava/lang/String;Ljava/lang/String;)Lcom/juul/kable/Characteristic;
 }
 
-public final class com/juul/kable/ConnectionLostException : java/io/IOException {
+public final class com/juul/kable/ConnectionLostException : com/juul/kable/NotConnectedException {
 	public fun <init> ()V
 }
 
@@ -154,7 +154,11 @@ public final class com/juul/kable/ManufacturerData {
 	public final fun getData ()[B
 }
 
-public final class com/juul/kable/NotReadyException : java/io/IOException {
+public class com/juul/kable/NotConnectedException : java/io/IOException {
+	public fun <init> ()V
+}
+
+public final class com/juul/kable/NotReadyException : com/juul/kable/NotConnectedException {
 	public fun <init> ()V
 }
 
@@ -168,7 +172,7 @@ public abstract interface class com/juul/kable/Peripheral {
 	public abstract fun connect (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun disconnect (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun getServices ()Ljava/util/List;
-	public abstract fun getState ()Lkotlinx/coroutines/flow/Flow;
+	public abstract fun getState ()Lkotlinx/coroutines/flow/StateFlow;
 	public abstract fun observe (Lcom/juul/kable/Characteristic;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
 	public abstract fun read (Lcom/juul/kable/Characteristic;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun read (Lcom/juul/kable/Descriptor;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;

core/build.gradle.kts

@@ -34,6 +34,7 @@ kotlin {
             dependencies {
                 api(libs.kotlinx.coroutines.core)
                 api(libs.uuid)
+                implementation(libs.tuulbox.collections)
             }
         }
 
@@ -72,9 +73,6 @@ kotlin {
 
         val appleMain by creating {
             dependsOn(commonMain)
-            dependencies {
-                implementation(libs.stately)
-            }
         }
 
         val appleTest by creating

core/src/androidMain/kotlin/Connection.kt

@@ -2,7 +2,7 @@ package com.juul.kable
 
 import android.bluetooth.BluetoothGatt
 import android.bluetooth.BluetoothGatt.GATT_SUCCESS
-import com.juul.kable.AndroidObservationEvent.CharacteristicChange
+import com.juul.kable.ObservationEvent.CharacteristicChange
 import com.juul.kable.gatt.Callback
 import com.juul.kable.gatt.GattStatus
 import kotlinx.coroutines.CoroutineDispatcher

core/src/androidMain/kotlin/Observations.kt

@@ -0,0 +1,11 @@
+package com.juul.kable
+
+internal actual fun Peripheral.observationHandler(): Observation.Handler = object : Observation.Handler {
+    override suspend fun startObservation(characteristic: Characteristic) {
+        (this@observationHandler as AndroidPeripheral).startObservation(characteristic)
+    }
+
+    override suspend fun stopObservation(characteristic: Characteristic) {
+        (this@observationHandler as AndroidPeripheral).stopObservation(characteristic)
+    }
+}

core/src/androidMain/kotlin/Peripheral.kt

@@ -41,6 +41,7 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.onCompletion
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.withContext
@@ -127,7 +128,7 @@ public enum class Priority { Low, Balanced, High }
 
 public class AndroidPeripheral internal constructor(
     parentCoroutineContext: CoroutineContext,
-    internal val bluetoothDevice: BluetoothDevice,
+    private val bluetoothDevice: BluetoothDevice,
     private val transport: Transport,
     private val phy: Phy,
     private val onServicesDiscovered: ServicesDiscoveredAction,
@@ -136,12 +137,15 @@ public class AndroidPeripheral internal constructor(
 
     private val logger = Logger(logging, tag = "Kable/Peripheral", identifier = bluetoothDevice.address)
 
+    internal val platformIdentifier = bluetoothDevice.address
+
     private val _state = MutableStateFlow<State>(State.Disconnected())
-    public override val state: Flow<State> = _state.asStateFlow()
+    public override val state: StateFlow<State> = _state.asStateFlow()
 
     private val receiver = registerBluetoothStateBroadcastReceiver { state ->
         if (state == STATE_OFF) {
             closeConnection()
+            observers.onConnectionLost()
             _state.value = State.Disconnected()
         }
     }
@@ -166,7 +170,7 @@ public class AndroidPeripheral internal constructor(
      */
     public val mtu: StateFlow<Int?> = _mtu.asStateFlow()
 
-    private val observers = Observers(this, logging)
+    private val observers = Observers<ByteArray>(this, logging)
 
     @Volatile
     private var _platformServices: List<PlatformService>? = null
@@ -205,14 +209,15 @@ public class AndroidPeripheral internal constructor(
             connection
                 .characteristicChanges
                 .onEach(observers.characteristicChanges::emit)
+                .onCompletion { observers.onConnectionLost() }
                 .launchIn(scope, start = UNDISPATCHED)
 
             suspendUntilOrThrow<State.Connecting.Services>()
             discoverServices()
             onServicesDiscovered(ServicesDiscoveredPeripheral(this@AndroidPeripheral))
             _state.value = State.Connecting.Observes
             logger.verbose { message = "Configuring characteristic observations" }
-            observers.rewire()
+            observers.onConnected()
         } catch (t: Throwable) {
             closeConnection()
             logger.error(t) { message = "Failed to connect" }
@@ -226,6 +231,9 @@ public class AndroidPeripheral internal constructor(
     private fun closeConnection() {
         _connection?.close()
         _connection = null
+
+        observers.onConnectionLost()
+
         // Avoid trampling existing `Disconnected` state (and its properties) by only updating if not already `Disconnected`.
         _state.update { previous -> previous as? State.Disconnected ?: State.Disconnected() }
     }
@@ -381,19 +389,16 @@ public class AndroidPeripheral internal constructor(
 
     internal suspend fun stopObservation(characteristic: Characteristic) {
         val platformCharacteristic = platformServices.findCharacteristic(characteristic)
+        setConfigDescriptor(platformCharacteristic, enable = false)
 
-        try {
-            setConfigDescriptor(platformCharacteristic, enable = false)
-        } finally {
-            logger.debug {
-                message = "setCharacteristicNotification"
-                detail(characteristic)
-                detail("value", "false")
-            }
-            connection
-                .bluetoothGatt
-                .setCharacteristicNotification(platformCharacteristic, false)
+        logger.debug {
+            message = "setCharacteristicNotification"
+            detail(characteristic)
+            detail("value", "false")
         }
+        connection
+            .bluetoothGatt
+            .setCharacteristicNotification(platformCharacteristic, false)
     }
 
     private suspend fun setConfigDescriptor(
@@ -506,3 +511,6 @@ private fun checkBluetoothAdapterState(
         throw BluetoothDisabledException("Bluetooth adapter state is $actualName ($actual), but $expectedName ($expected) was required.")
     }
 }
+
+internal actual val Peripheral.identifier: String
+    get() = (this as AndroidPeripheral).platformIdentifier

core/src/appleMain/kotlin/Observations.kt

@@ -0,0 +1,11 @@
+package com.juul.kable
+
+internal actual fun Peripheral.observationHandler(): Observation.Handler = object : Observation.Handler {
+    override suspend fun startObservation(characteristic: Characteristic) {
+        (this@observationHandler as ApplePeripheral).startNotifications(characteristic)
+    }
+
+    override suspend fun stopObservation(characteristic: Characteristic) {
+        (this@observationHandler as ApplePeripheral).stopNotifications(characteristic)
+    }
+}