kotlin - 用 Kotlin 序列替换同步回调

Question

在下面的示例中，回调被调用多次，直到输入被消耗。我如何把它变成 Sequence或 Flow适合即时消费？我看到许多(严重的是，负载 - 没有人阅读 SO)类似的一次性回调协程问题，使用 suspendCoroutine 解决。但我看不到如何在这里应用它。我敢肯定有一些涉及 Channel()我无法工作......

任何指针表示赞赏!

    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()

Answer 1

非常感谢(但没有积分!)@marko-topolnik 提供 channelFlow 的知识.我知道会有一个优雅的解决方案

    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) })
    }