-
|
Hey @twyatt, can you help me with a structured concurrency issue here? I'm trying to achieve the following:
Because the What I did so far: private val scanner = Scanner {
filters = listOf(Filter.Service(Device.ADVERTISEMENT_UUID))
logging {
engine = SystemLogEngine
level = Logging.Level.Data
format = Logging.Format.Multiline
data = Hex
}
}
class DeviceManager(val scope: CoroutineScope) {
private val scanScope = scope.childScope()
private val readScope = scope.childScope()
private val _data = MutableStateFlow<List<DeviceData>>(listOf())
val data = _data.asSharedFlow()
fun start() {
scanScope.launch {
val advertisement = scanner.advertisements.first()
scope.launch {
val peripheral = scope.peripheral(advertisement) {
logging {
level = Logging.Level.Events
}
}
peripheral.connect()
peripheral.state.collect {
when (it) {
State.Connected -> readScope.launch { collect(peripheral) }
is State.Disconnected -> start()
else -> {}
}
}
}
}
}
private suspend fun collect(peripheral: Peripheral) {
peripheral.observe(TX_CHARACTERISTIC).shareIn(readScope, SharingStarted.Eagerly).onSubscription {
peripheral.write(RX_CHARACTERISTIC, "get".encodeToByteArray())
}.collect { data ->
val deviceData = DeviceData(data)
_data.value = listOf(deviceData)
peripheral.write(RX_CHARACTERISTIC, "cancel".encodeToByteArray()) // This will trigger disconnect
readScope.cancelChildren()
}
}Unfortunately using this solution I have issues on upcoming connections First connection: Second start when first device was disconnected: Third scan, looks like two coroutines are running here: I'm sure there is more elegant and flow-ish way do achieve this and to work properly, help! |
Beta Was this translation helpful? Give feedback.
Answered by
twyatt
Apr 6, 2023
Replies: 1 comment 5 replies
-
|
I haven't had time to test it, so apologies if it totally falls short, but let me know if something like this is what you're shooting for?: val data = MutableSharedFlow<ByteArray>()
val scope = CoroutineScope(..)
while (true) {
val advertisement = scanner.advertisements.first()
val peripheral = scope.peripheral(advertisement)
peripheral.connect()
val readScope = scope.childScope()
peripheral.observe(tx) {
peripheral.write(rx, "get".encodeToByteArray())
}.onEach(data::emit).launchIn(readScope)
peripheral.state.first { it is Disconnected }
readScope.cancel()
} |
Beta Was this translation helpful? Give feedback.
5 replies
-
|
Hey @twyatt thanks for the answer. Your snippet works but I have again some similar issue. Every next scan and connect on same device, Kable shows more and more duplicated logs for On third connect the situation is JobCancellationException: Is it possible the Peripheral scope to spawn new coroutines in parallel trying to connect in same moment to single device? |
Beta Was this translation helpful? Give feedback.
-
|
Looks like the |
Beta Was this translation helpful? Give feedback.
-
|
Probably a bug in Kable. I'll try to find time to look into it soon. |
Beta Was this translation helpful? Give feedback.
-
|
Tracking this potential bug via #458. |
Beta Was this translation helpful? Give feedback.
Answer selected by
azlekov
-
|
Just FYI the workaround which seems to work is everything related with the peripheral to use separate val data = MutableSharedFlow<ByteArray>()
val scope = CoroutineScope(..)
while (true) {
val advertisement = scanner.advertisements.first()
val readScope = scope.childScope()
val peripheral = readScope.peripheral(advertisement)
peripheral.connect()
peripheral.observe(tx) {
peripheral.write(rx, "get".encodeToByteArray())
}.onEach(data::emit).launchIn(readScope)
peripheral.state.first { it is Disconnected }
readScope.cancel()
}Notice that peripheral is using also the new scope: val peripheral = readScope.peripheral(advertisement) |
Beta Was this translation helpful? Give feedback.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Tracking this potential bug via #458.