How can I wrap reading data packets from socket in...

# coroutines

[dkhusainov]

How can I wrap reading data packets from socket inside a Flow? With callbackFlow or a channel and a separate coroutine that reads from socket and sends to channel I end up with a Flow which isn't cancelable because socket.receive is a blocking call(which supposedly never returns) inside a while loop.

[Dominaezzz]

Yeah, callbackFlow . If there's no way to cancel socket.receive , then there's not much you can do about it.

[Adam Powell]

you can close the socket

[dkhusainov]

Closing socket is called from onCompletion, which is not called until the blocking call is finished

[Dominaezzz]

If you're using channelFlow I'd expect you to be closing the socket in awaitClose .

[dkhusainov]

Yes, but the code is either not going to even reach registering the callback because the suspending lamda is doing a blocking loop, or the callback is not called because another coroutine which is running the blocking loop can't complete.

[dkhusainov]

In short - I don't see a way of implementing this in a self contained way

[louiscad]

You can use launch(<http://Dispatchers.IO|Dispatchers.IO>) in callbackFlow

[hmole]

Working variant if anyone intersted

fun receiveUdpBroadcasts(
    port: Int,
    bufferSize: Int = DEFAULT_BUFFER_SIZE
): Flow<UdpPacket> = callbackFlow {
    val socket = DatagramSocket(port)

    suspend fun receiveAndSend() {
        val packet = DatagramPacket(ByteArray(bufferSize), bufferSize)

        socket.receive(packet)

        if (isActive && !channel.isClosedForSend) {
            channel.send(
                UdpPacket(
                    data = packet.data.toByteString(),
                    sender = socket.inetAddress
                )
            )
        }
    }

    launch(<http://Dispatchers.IO|Dispatchers.IO>) {
        while (isActive) {
            try {
                receiveAndSend()
            } catch (socketClosed: SocketException) {
                break
            }
        }
    }

    awaitClose(socket::close)
}