Skip to content

Commit ccb9b60

Browse files
twyatttwyatt
authored
Fix BLE I/O cancellation handling (#453)
1 parent 15b2019 commit ccb9b60

6 files changed

Lines changed: 80 additions & 38 deletions

File tree

‎core/src/androidMain/kotlin/BluetoothDevice.kt‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import kotlinx.coroutines.android.asCoroutineDispatcher
2020
import kotlinx.coroutines.flow.MutableSharedFlow
2121
import kotlinx.coroutines.flow.MutableStateFlow
2222
import kotlinx.coroutines.newSingleThreadContext
23+
import kotlin.coroutines.CoroutineContext
2324

2425
internal sealed class Threading {
2526

@@ -64,6 +65,7 @@ internal fun BluetoothDevice.threading(): Threading =
6465
* @param phy is only used on API level >= 26.
6566
*/
6667
internal fun BluetoothDevice.connect(
68+
parentCoroutineContext: CoroutineContext,
6769
context: Context,
6870
transport: Transport,
6971
phy: Phy,
@@ -89,7 +91,7 @@ internal fun BluetoothDevice.connect(
8991
else -> connectGatt(context, false, callback)
9092
} ?: return null
9193

92-
return Connection(bluetoothGatt, threading.dispatcher, callback, logging, invokeOnClose)
94+
return Connection(parentCoroutineContext, bluetoothGatt, threading.dispatcher, callback, logging, invokeOnClose)
9395
}
9496

9597
private val Transport.intValue: Int

‎core/src/androidMain/kotlin/Connection.kt‎

Lines changed: 29 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -4,12 +4,19 @@ import android.bluetooth.BluetoothGatt
44
import android.bluetooth.BluetoothGatt.GATT_SUCCESS
55
import com.juul.kable.gatt.Callback
66
import com.juul.kable.gatt.GattStatus
7+
import com.juul.kable.gatt.Response
78
import com.juul.kable.logs.Logger
89
import com.juul.kable.logs.Logging
910
import kotlinx.coroutines.CoroutineDispatcher
11+
import kotlinx.coroutines.CoroutineScope
12+
import kotlinx.coroutines.Deferred
13+
import kotlinx.coroutines.Job
14+
import kotlinx.coroutines.async
15+
import kotlinx.coroutines.cancel
1016
import kotlinx.coroutines.sync.Mutex
1117
import kotlinx.coroutines.sync.withLock
1218
import kotlinx.coroutines.withContext
19+
import kotlin.coroutines.CoroutineContext
1320

1421
public class OutOfOrderGattCallbackException internal constructor(
1522
message: String,
@@ -18,6 +25,7 @@ public class OutOfOrderGattCallbackException internal constructor(
1825
private val GattSuccess = GattStatus(GATT_SUCCESS)
1926

2027
internal class Connection(
28+
parentCoroutineContext: CoroutineContext,
2129
internal val bluetoothGatt: BluetoothGatt,
2230
internal val dispatcher: CoroutineDispatcher,
2331
private val callback: Callback,
@@ -29,10 +37,12 @@ internal class Connection(
2937
callback.invokeOnDisconnected(::close)
3038
}
3139

40+
private val scope = CoroutineScope(parentCoroutineContext + Job(parentCoroutineContext[Job]))
41+
3242
private val logger = Logger(logging, tag = "Kable/Connection", identifier = bluetoothGatt.device.address)
3343

3444
private val lock = Mutex()
35-
private var pending = false
45+
private var deferredResponse: Deferred<Response>? = null
3646

3747
/**
3848
* Executes specified [BluetoothGatt] [action].
@@ -51,28 +61,33 @@ internal class Connection(
5161
suspend inline fun <reified T> execute(
5262
crossinline action: BluetoothGatt.() -> Boolean,
5363
): T = lock.withLock {
54-
if (pending) {
55-
// Discard response as we've performed another `execute` without the previous finishing. This happens if a
56-
// previous `execute` was cancelled after invoking GATT action, but before receiving response from callback
57-
// channel. See https://github.com/JuulLabs/kable/issues/326 for more details.
58-
val response = callback.onResponse.receive()
59-
pending = false
60-
logger.warn {
61-
message = "Discarded response"
62-
detail("response", response.toString())
64+
deferredResponse?.let {
65+
if (it.isActive) {
66+
// Discard response as we've performed another `execute` without the previous finishing. This happens if
67+
// a previous `execute` was cancelled after invoking GATT action, but before receiving response from
68+
// callback channel. See the following issues for more details:
69+
// https://github.com/JuulLabs/kable/issues/326
70+
// https://github.com/JuulLabs/kable/issues/450
71+
val response = it.await()
72+
logger.warn {
73+
message = "Discarded response"
74+
detail("response", response.toString())
75+
}
6376
}
6477
}
6578

6679
withContext(dispatcher) {
67-
pending = true
68-
action.invoke(bluetoothGatt) || throw GattRequestRejectedException()
80+
if (!bluetoothGatt.action()) throw GattRequestRejectedException()
6981
}
82+
val deferred = scope.async { callback.onResponse.receive() }
83+
deferredResponse = deferred
7084

7185
val response = try {
72-
callback.onResponse.receive().also { pending = false }
86+
deferred.await()
7387
} catch (e: ConnectionLostException) {
7488
throw ConnectionLostException(cause = e)
7589
}
90+
deferredResponse = null
7691

7792
if (response.status != GattSuccess) throw GattStatusException(response.toString())
7893

@@ -110,6 +125,7 @@ internal class Connection(
110125
}
111126

112127
fun close() {
128+
scope.cancel()
113129
bluetoothGatt.close()
114130
invokeOnClose.invoke()
115131
}

‎core/src/androidMain/kotlin/Peripheral.kt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ import kotlinx.coroutines.flow.filter
4343
import kotlinx.coroutines.flow.launchIn
4444
import kotlinx.coroutines.flow.onEach
4545
import kotlinx.coroutines.flow.update
46+
import kotlinx.coroutines.job
4647
import kotlinx.coroutines.withContext
4748
import kotlin.coroutines.CoroutineContext
4849

@@ -144,6 +145,7 @@ internal class BluetoothDeviceAndroidPeripheral(
144145
private fun establishConnection(): Connection {
145146
logger.info { message = "Connecting" }
146147
return bluetoothDevice.connect(
148+
scope.coroutineContext,
147149
applicationContext,
148150
transport,
149151
phy,

‎core/src/appleMain/kotlin/CentralManager.kt‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -48,16 +48,19 @@ public class CentralManager internal constructor() {
4848
.firstOrNull() as? CBPeripheral
4949

5050
internal suspend fun connectPeripheral(
51-
cbPeripheral: CBPeripheral,
51+
peripheral: CBPeripheralCoreBluetoothPeripheral,
5252
logging: Logging,
53-
delegate: PeripheralDelegate,
5453
options: Map<Any?, *>? = null,
5554
): Connection {
55+
val parentCoroutineContext = peripheral.connectionScope.coroutineContext
56+
val cbPeripheral = peripheral.cbPeripheral
57+
val identifier = cbPeripheral.identifier.UUIDString
58+
val delegate = PeripheralDelegate(peripheral.canSendWriteWithoutResponse, logging, identifier)
5659
withContext(dispatcher) {
5760
cbPeripheral.delegate = delegate
5861
cbCentralManager.connectPeripheral(cbPeripheral, options)
5962
}
60-
return Connection(delegate, logging, cbPeripheral.identifier.UUIDString)
63+
return Connection(parentCoroutineContext, delegate, logging, identifier)
6164
}
6265

6366
internal suspend fun cancelPeripheralConnection(

‎core/src/appleMain/kotlin/Connection.kt‎

Lines changed: 34 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -2,47 +2,68 @@ package com.juul.kable
22

33
import com.juul.kable.logs.Logger
44
import com.juul.kable.logs.Logging
5+
import kotlinx.coroutines.CoroutineName
6+
import kotlinx.coroutines.CoroutineScope
7+
import kotlinx.coroutines.Deferred
8+
import kotlinx.coroutines.Job
9+
import kotlinx.coroutines.async
10+
import kotlinx.coroutines.cancel
511
import kotlinx.coroutines.sync.Semaphore
612
import kotlinx.coroutines.sync.withPermit
13+
import kotlin.coroutines.CoroutineContext
714

815
internal class Connection(
16+
parentCoroutineContext: CoroutineContext,
917
val delegate: PeripheralDelegate,
1018
logging: Logging,
1119
identifier: String,
1220
) {
1321

22+
private val scope = CoroutineScope(parentCoroutineContext + Job(parentCoroutineContext[Job]) + CoroutineName("Kable/Connection@$identifier"))
23+
1424
private val logger = Logger(logging, tag = "Kable/Connection", identifier = identifier)
1525

1626
// Using Semaphore as Mutex never relinquished lock when multiple concurrent `withLock`s are executed.
1727
val semaphore = Semaphore(1)
1828

19-
private var pending = false
29+
private var deferredResponse: Deferred<PeripheralDelegate.Response>? = null
2030

2131
suspend inline fun <T> execute(
2232
action: () -> Unit,
2333
): T = semaphore.withPermit {
24-
if (pending) {
25-
// Discard response as we've performed another `execute` without the previous finishing. This happens if a
26-
// previous `execute` was cancelled after invoking GATT action, but before receiving response from callback
27-
// channel. See https://github.com/JuulLabs/kable/issues/326 for more details.
28-
val response = delegate.response.receive()
29-
pending = false
30-
logger.warn {
31-
message = "Discarded response"
32-
detail("response", response.toString())
34+
deferredResponse?.let {
35+
if (it.isActive) {
36+
// Discard response as we've performed another `execute` without the previous finishing. This happens if
37+
// a previous `execute` was cancelled after invoking GATT action, but before receiving response from
38+
// callback channel. See the following issues for more details:
39+
// https://github.com/JuulLabs/kable/issues/326
40+
// https://github.com/JuulLabs/kable/issues/450
41+
val response = it.await()
42+
logger.warn {
43+
message = "Discarded response"
44+
detail("response", response.toString())
45+
}
3346
}
3447
}
3548

36-
pending = true
3749
action.invoke()
38-
val response = delegate.response.receive()
39-
pending = false
50+
val deferred = scope.async { delegate.response.receive() }
51+
deferredResponse = deferred
52+
53+
val response = try {
54+
deferred.await()
55+
} catch (e: ConnectionLostException) {
56+
throw ConnectionLostException(cause = e)
57+
}
58+
deferredResponse = null
59+
4060
val error = response.error
4161
if (error != null) throw IOException(error.description, cause = null)
4262
response as T
4363
}
4464

4565
fun close() {
66+
scope.cancel()
4667
delegate.close()
4768
}
4869
}

‎core/src/appleMain/kotlin/Peripheral.kt‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import com.juul.kable.logs.Logging
3030
import com.juul.kable.logs.detail
3131
import kotlinx.atomicfu.atomic
3232
import kotlinx.atomicfu.updateAndGet
33+
import kotlinx.coroutines.CoroutineName
3334
import kotlinx.coroutines.CoroutineScope
3435
import kotlinx.coroutines.CoroutineStart.LAZY
3536
import kotlinx.coroutines.CoroutineStart.UNDISPATCHED
@@ -109,15 +110,15 @@ public fun CoroutineScope.peripheral(
109110

110111
internal class CBPeripheralCoreBluetoothPeripheral(
111112
parentCoroutineContext: CoroutineContext,
112-
private val cbPeripheral: CBPeripheral,
113+
internal val cbPeripheral: CBPeripheral,
113114
observationExceptionHandler: ObservationExceptionHandler,
114115
private val onServicesDiscovered: ServicesDiscoveredAction,
115116
private val logging: Logging,
116117
) : CoreBluetoothPeripheral {
117118

118119
private val job = SupervisorJob(parentCoroutineContext.job) // todo: Disconnect/dispose CBPeripheral on completion?
119-
private val scope = CoroutineScope(parentCoroutineContext + job)
120-
private val connectionScope = CoroutineScope(scope.coroutineContext + Job(scope.coroutineContext[Job]))
120+
private val scope = CoroutineScope(parentCoroutineContext + job + CoroutineName("Kable/Peripheral@${cbPeripheral.identifier.UUIDString}"))
121+
internal val connectionScope = CoroutineScope(scope.coroutineContext + Job(scope.coroutineContext[Job]) + CoroutineName("Kable/Connect@${cbPeripheral.identifier.UUIDString}"))
121122

122123
private val centralManager: CentralManager = CentralManager.Default
123124

@@ -154,7 +155,7 @@ internal class CBPeripheralCoreBluetoothPeripheral(
154155
.launchIn(scope)
155156
}
156157

157-
private val canSendWriteWithoutResponse = MutableStateFlow(cbPeripheral.canSendWriteWithoutResponse)
158+
internal val canSendWriteWithoutResponse = MutableStateFlow(cbPeripheral.canSendWriteWithoutResponse)
158159

159160
private val _discoveredServices = atomic<List<DiscoveredService>?>(null)
160161
private val discoveredServices: List<DiscoveredService>
@@ -189,10 +190,7 @@ internal class CBPeripheralCoreBluetoothPeripheral(
189190
}.launchIn(connectionScope)
190191

191192
try {
192-
// todo: Create in `connectPeripheral`.
193-
val delegate = PeripheralDelegate(canSendWriteWithoutResponse, logging, cbPeripheral.identifier.UUIDString)
194-
195-
val connection = centralManager.connectPeripheral(cbPeripheral, logging, delegate).also {
193+
val connection = centralManager.connectPeripheral(this@CBPeripheralCoreBluetoothPeripheral, logging).also {
196194
_connection.value = it
197195
}
198196

0 commit comments

Comments
 (0)