Замена обратных вызовов syn c последовательностью Kotlin - PullRequest
Новая
       1

Замена обратных вызовов syn c последовательностью Kotlin

1 голос
/ 31 марта 2020

В приведенном ниже примере обратный вызов вызывается много раз, пока не будет использован ввод. Как превратить это в Sequence или Flow, подходящий для потребления на лету? Я вижу много (серьезно, загружает - никто не читает SO) аналогичных вопросов сопрограмм для обратных вызовов one-shot , разрешенных с помощью 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()

1 Ответ

1 голос
/ 31 марта 2020

Огромное спасибо (но без очков!) @ 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) })
    }
...