Replacing sync callbacks with Kotlin sequence

user_628318 0
Viewed 364

In the below sample, a callback is invoked many times until input is consumed. How do I make this into a Sequence or Flow suitable for on-the-fly consumption? I see many (seriously, loads - does no one read SO) similar coroutine questions for one-shot callbacks resolved using suspendCoroutine but I cannot see how to apply that here. I'm sure there's something involving a Channel() that I could not make work ...

Any pointers appreciated!

    fun String.inputStream(): InputStream = object {}.javaClass.getResourceAsStream(this)

    fun main() {
        val input = BufferedReader(InputStreamReader("/file.txt".inputStream()))
        input.mark(1000)
        readeR(input, Consumer { println(it) } ) // Synchronous call
        input.reset()
        sequenceR(input).forEach { println(it) } // Async-sequence
    }

    // This is a mockup of existing (Java) code, cannot change this method
    fun readeR(reader: BufferedReader, handleR: Consumer<String>) {
        do {
            val line = reader.readLine()
            if (line != null && line.startsWith("R")) handleR.accept(line)
        } while (line != null)
    }

    // TODO
    fun sequenceR(r: BufferedReader) = sequence {
        readeR(r, Consumer { yield(it) }) // <<< Q1 How to build sequence from the callbacks
        yield(null) // <<< Q2 How to close the sequence?
    }.constrainOnce()
kotlin-coroutines kotlin
1 Answers

Huge thanks (but no points!) to @marko-topolnik for knowledge of channelFlow. I knew there would an elegant solution out there

    fun main() {
        val input = BufferedReader(InputStreamReader("/file.txt".inputStream()))
        runBlocking {
            flowR(reader).collect { println(it) } // Async-supplied iterator
        }
    }

    fun flowR(r: BufferedReader): Flow<String> = channelFlow {
        readeR(r, Consumer<String> { sendBlocking(it) })
    }
user_6283180
Related