Последовательная связь Android с использованием потоков, обработчиков, потоков Kotlin, функций приостановки и составлени ⇐ Android

[Anonymous]

Я пытаюсь написать приложение для Android, которое требует последовательной связи с устройством Bluetooth. Некоторые из этих сообщений представляют собой команды/ответы, а некоторые — данные о событиях, которые просто отправляются с устройства. Все это пакетируется с идентификаторами типов пакетов. Я настроил последовательный поток Bluetooth, который обрабатывает потоки ввода/вывода bluetoothSocket. Я могу создавать пакеты команд и отправлять их. Поток принимает обработчик в качестве аргумента и отправляет сообщения в основной поток при получении полного пакета данных (несколько байтов в ByteArray).
Этот обработчик определяется в " callbackflow», который переопределяет функцию «handleMessage» и использует «trySend» для создания ByteArray.
Мне удалось отобразить полученные пакеты в реальном времени на компонуемый пользовательский интерфейс в виде шестнадцатеричных строк путем преобразования потока в SharedFlow в моем «источнике данных устройства Bluetooth» в StateFlow в моей ViewModel.
Я бы хотел, чтобы другие области моего приложения могли подписаться в SharedFlow и отфильтровывать пакеты, которые им интересны, но я также хотел бы иметь возможность последовательно запускать цепочки команд, не встраивая логику в пользовательский интерфейс.
Я подумал, что смогу добиться этого, создав функции приостановки, которые можно вызывать последовательно для выполнения ряда операций на внешнем устройстве. Например:
Отправить команду 1
Подождать ответа
Отправить команду 2
Подождать ответа
и т. д.
Я пытаясь понять, как они должны быть реализованы. По сути, я хочу инициировать отправку команды, а затем «прослушать» полученный поток пакетов на предмет правильного типа пакета, а затем вернуть пакет из функции приостановки.
Я новичок в Kotlin помимо того, что я довольно любитель в Android, поэтому я не уверен, что это «правильный» способ сделать что-то.
обработчик и поток обратного вызова выглядят следующим образом:
fun btInputStreamHandler(): Flow = callbackFlow {

val btMessageHandler = object : Handler(Looper.getMainLooper()) {
override fun handleMessage(msg: Message) {
super.handleMessage(msg)
val b = msg.data
val rxpacket = b.getByteArray("rxpacket")
if (rxpacket != null) {
//handleRxPacket(rxpacket)
trySend(rxpacket)
}
}
}
//this messageHandler is passed into the serial comms thread when a new device is connected
messageHandler = btMessageHandler

awaitClose {
messageHandler = null
}
}

val btInputPacketSharedFlow: SharedFlow = btInputStreamHandler()
.shareIn(CoroutineScope(Dispatchers.IO), SharingStarted.Eagerly)

В первую очередь я пытаюсь получить что-то вроде этого:
suspend fun getValueFromBluetoothDevice(): ByteArray {

//this is the serial comms thread that loads packets to be sent asynchronously
btSerialProcess?.sendCommand(DeviceCommand.getParameter1())

var flag: Boolean = true
var result: ByteArray = ByteArray(0)
while( flag ) {
//Wait for flow to emit the correct packet type
result = btInputPacketSharedFlow
.filterNotNull()
.first()
if( DeviceCommand.getPacketType(result) == CORRECT_PACKET_TYPE ) {
flag = false

}
}
return result
}

Очевидно, что здесь есть некоторые пробелы в моем понимании, потому что, хотя команда действительно отправляется и ответ возвращается (пользовательский интерфейс обновляется со строкой пакета), функция приостановки не возвращает значение и даже не возвращается.
Обновить
Я предполагаю, что это неправильный способ сделать это, но это вроде как работает:
suspend fun getCmdResult(cmd: DeviceCommand, packetType1: Int, packetType2: Int): ByteArray {
var result: ByteArray = ByteArray(0)
var flag: Boolean = true
val job = CoroutineScope(Dispatchers.IO).launch {
btInputPacketSharedFlow.collect { packet ->
result = packet
if (
(DeviceCommand.getPacketType1(result) == packetType1)
and
(DeviceCommand.getPacketType2(result) == packetType2)
){
flag = false

}
}
}
btSerialProcess?.sendCommand(cmd)
while( flag ) {
delay(50)
}
job.cancel()
return result
}


Подробнее здесь: https://stackoverflow.com/questions/792 ... ctions-and