I’m trying to write an Android app which requires some serial comms with a bluetooth device. Some of this comms is command/response and some of it is event data that just gets sent from the device. All of it is packetized with packet type identifiers. I’ve setup a bluetooth serial thread which handles bluetoothSocket input/output streams. I can create command packets and send them. The thread takes a Handler as an arguement and sends messages to the main thread when a complete data packet is received (a handful of bytes in a ByteArray).

This Handler is defined within a “callbackflow” which overrides the “handleMessage” function and uses “trySend” to emit the ByteArray.

I’ve been able to get the received packets to display in realtime on the composable ui as hexstrings by converting the flow to a SharedFlow within my “bluetooth device datasource” to a StateFlow within my ViewModel.

I’d like other areas of my app to be able to subscribe to the SharedFlow and filter out the packets that they are interested in but I’d also like to be able to run chains of commands sequentially without building the logic into the Ui.

I thought I might be able to achieve this by creating suspend functions that can be called sequentially to carry out a series of operations on the external device. E.g:
Send command 1
Wait for response
Send command 2
Wait for response
etc.

I’m trying to work out how these are supposed to be implemented. I basically want to trigger sending the command and then “listen” on the received packet flow for the correct packet type, then return the packet from the suspend function.

I’m new to Kotlin on top of being fairly amateur in Android so I’m not sure if any of this is the “correct” way to do things.

the handler and callbackflow is like this:

fun btInputStreamHandler(): Flow<ByteArray> = 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<ByteArray> = btInputStreamHandler()
        .shareIn(CoroutineScope(Dispatchers.IO), SharingStarted.Eagerly)

In the first instance I’m trying to end up with something like this:

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
    }

Clearly there are some gaps in my understanding here because although the command does get sent and the response does comeback (the UI updates with the packet string) the suspend function doesn’t return the value and doesn’t even return.

Update
I assume this is the wrong way to do it but it sort of works:

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
    }