Reduce reliance on CoroutineScope (#427)

As identified in #378, `Peripheral` objects eagerly lean on the
`CoroutineScope` receiver provided at `Peripheral` creation. This isn't
necessary and this is a small step in removing the crutches that were
being relied on.

Author: twyatt

core/src/androidMain/kotlin/BluetoothDevice.kt

@@ -17,6 +17,7 @@ import com.juul.kable.logs.Logging
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.ExecutorCoroutineDispatcher
 import kotlinx.coroutines.android.asCoroutineDispatcher
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.newSingleThreadContext
 
@@ -68,6 +69,7 @@ internal fun BluetoothDevice.connect(
     phy: Phy,
     state: MutableStateFlow<State>,
     mtu: MutableStateFlow<Int?>,
+    onCharacteristicChanged: MutableSharedFlow<ObservationEvent<ByteArray>>,
     logging: Logging,
     threading: Threading,
     invokeOnClose: () -> Unit,
@@ -76,7 +78,7 @@ internal fun BluetoothDevice.connect(
     // Disconnected before the connection request has kicked off the Connecting state (via Callback).
     state.value = State.Connecting.Bluetooth
 
-    val callback = Callback(state, mtu, logging, address)
+    val callback = Callback(state, mtu, onCharacteristicChanged, logging, address)
 
     val bluetoothGatt = when {
         Build.VERSION.SDK_INT >= Build.VERSION_CODES.O -> {

core/src/androidMain/kotlin/Connection.kt

@@ -2,13 +2,11 @@ package com.juul.kable
 
 import android.bluetooth.BluetoothGatt
 import android.bluetooth.BluetoothGatt.GATT_SUCCESS
-import com.juul.kable.ObservationEvent.CharacteristicChange
 import com.juul.kable.gatt.Callback
 import com.juul.kable.gatt.GattStatus
 import com.juul.kable.logs.Logger
 import com.juul.kable.logs.Logging
 import kotlinx.coroutines.CoroutineDispatcher
-import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
@@ -33,11 +31,6 @@ internal class Connection(
 
     private val logger = Logger(logging, tag = "Kable/Connection", identifier = bluetoothGatt.device.address)
 
-    val characteristicChanges = callback.onCharacteristicChanged
-        .map { (bluetoothGattCharacteristic, value) ->
-            CharacteristicChange(bluetoothGattCharacteristic.toLazyCharacteristic(), value)
-        }
-
     private val lock = Mutex()
     private var pending = false
 

core/src/androidMain/kotlin/Peripheral.kt

@@ -31,7 +31,6 @@ import kotlinx.atomicfu.atomic
 import kotlinx.atomicfu.updateAndGet
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.CoroutineStart.LAZY
-import kotlinx.coroutines.CoroutineStart.UNDISPATCHED
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
@@ -156,6 +155,7 @@ public class AndroidPeripheral internal constructor(
             phy,
             _state,
             _mtu,
+            observers.characteristicChanges,
             logging,
             threading,
             invokeOnClose = { connectJob.value = null },
@@ -165,13 +165,7 @@ public class AndroidPeripheral internal constructor(
     /** Creates a connect [Job] that completes when connection is established, or failure occurs. */
     private fun connectAsync() = scope.async(start = LAZY) {
         try {
-            val connection = establishConnection().also { _connection = it }
-
-            connection
-                .characteristicChanges
-                .onEach(observers.characteristicChanges::emit)
-                .launchIn(scope, start = UNDISPATCHED)
-
+            _connection = establishConnection()
             suspendUntilOrThrow<State.Connecting.Services>()
             discoverServices()
             onServicesDiscovered(ServicesDiscoveredPeripheral(this@AndroidPeripheral))

core/src/androidMain/kotlin/gatt/Callback.kt

@@ -10,6 +10,8 @@ import android.bluetooth.BluetoothProfile.STATE_CONNECTING
 import android.bluetooth.BluetoothProfile.STATE_DISCONNECTED
 import android.bluetooth.BluetoothProfile.STATE_DISCONNECTING
 import com.juul.kable.ConnectionLostException
+import com.juul.kable.ObservationEvent
+import com.juul.kable.ObservationEvent.CharacteristicChange
 import com.juul.kable.State
 import com.juul.kable.State.Disconnected.Status.Cancelled
 import com.juul.kable.State.Disconnected.Status.CentralDisconnected
@@ -35,25 +37,20 @@ import com.juul.kable.gatt.Response.OnServicesDiscovered
 import com.juul.kable.logs.Logger
 import com.juul.kable.logs.Logging
 import com.juul.kable.logs.detail
+import com.juul.kable.toLazyCharacteristic
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.CONFLATED
-import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.channels.onFailure
-import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.consumeAsFlow
 
 private typealias DisconnectedAction = () -> Unit
 
-internal data class OnCharacteristicChanged(
-    val characteristic: BluetoothGattCharacteristic,
-    val value: ByteArray,
-)
-
 internal class Callback(
     private val state: MutableStateFlow<State>,
     private val mtu: MutableStateFlow<Int?>,
+    private val onCharacteristicChanged: MutableSharedFlow<ObservationEvent<ByteArray>>,
     logging: Logging,
     macAddress: String,
 ) : BluetoothGattCallback() {
@@ -65,10 +62,6 @@ internal class Callback(
         disconnectedAction = action
     }
 
-    private val _onCharacteristicChanged = Channel<OnCharacteristicChanged>(UNLIMITED)
-    val onCharacteristicChanged: Flow<OnCharacteristicChanged> =
-        _onCharacteristicChanged.consumeAsFlow()
-
     val onResponse = Channel<Response>(CONFLATED)
     val onMtuChanged = Channel<OnMtuChanged>(CONFLATED)
 
@@ -126,7 +119,6 @@ internal class Callback(
         }
 
         if (newState == STATE_DISCONNECTING || newState == STATE_DISCONNECTED) {
-            _onCharacteristicChanged.close()
             onResponse.close(ConnectionLostException())
         }
     }
@@ -175,13 +167,13 @@ internal class Callback(
         characteristic: BluetoothGattCharacteristic,
     ) {
         val value = characteristic.value
-        val event = OnCharacteristicChanged(characteristic, value)
         logger.debug {
             message = "onCharacteristicChanged"
             detail(characteristic)
             detail(value)
         }
-        _onCharacteristicChanged.trySendOrLog(event)
+        val event = CharacteristicChange(characteristic.toLazyCharacteristic(), value)
+        onCharacteristicChanged.tryEmitOrLog(event)
     }
 
     override fun onDescriptorRead(
@@ -261,6 +253,14 @@ internal class Callback(
             }
         }
     }
+
+    private fun <E> MutableSharedFlow<E>.tryEmitOrLog(element: E) {
+        if (!tryEmit(element)) {
+            logger.warn {
+                message = "Callback was unable to deliver $element"
+            }
+        }
+    }
 }
 
 private val Int.disconnectedConnectionStatus: State.Disconnected.Status?

core/src/commonMain/kotlin/Observers.kt

@@ -40,11 +40,10 @@ internal expect fun Peripheral.observationHandler(): Observation.Handler
 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)
+    val characteristicChanges = MutableSharedFlow<ObservationEvent<T>>(extraBufferCapacity = Int.MAX_VALUE)
     private val observations = Observations()
 
     fun acquire(

core/src/jsMain/kotlin/Peripheral.kt

@@ -298,7 +298,7 @@ public class JsPeripheral internal constructor(
         .buffer
         .toByteArray()
 
-    private val observers = Observers<DataView>(this, logging, extraBufferCapacity = 64, observationExceptionHandler)
+    private val observers = Observers<DataView>(this, logging, observationExceptionHandler)
 
     public fun observeDataView(
         characteristic: Characteristic,