Потоки данных только с Kotlin Coroutines?
С RxJava я привык к тому, что мои репозитории возвращают Observables данных, которые автоматически обновляются всякий раз, когда есть основное изменение. Я достигаю этого, просто имея субъект в своем репозитории, который получает уведомление с соответствующей информацией об изменении, и наблюдаемые объекты, такие как getAll(), уходят от этого.
В качестве примера возьмем такой псевдокод, как snippet:
fun getAll(): Observable<List<Model> {
subject
.filter { isChangeRelevant(it) }
.startWith(initialChangeEvent)
.map { queryAll() }
}
Мне было любопытно, как и можно ли добиться того же самого, используя только сопрограммы?
Ваш пример — это простой вопрос для любой реализации потоков, включая базовые потоки Java, потоки Java Reactive или сопрограммы. Таким образом, он не иллюстрирует слияние наблюдаемых цепочек, расходящихся цепочек, цепочек обратной связи или межпотокового взаимодействия, которые являются хлебом с маслом мира RxJava.
Ответы 1
Вы можете использовать Kotlin Coroutines Channels.
Если вы хотите, чтобы ваши значения испускались только как поток (чтобы вы могли for-each отключиться от него), вы можете использовать produce для их создания (который возвращает ReceiveChannel):
fun test(): ReceiveChannel<Int>{
return produce {
send(1)
send(5)
send(100)
}
}
Вы можете использовать for-each (или consumeEach) для значений test(), чтобы получить его значения.
Если вы хотите, чтобы ваш канал был точно таким же, как PublishSubject RxJava, вы можете использовать ConflatedBroadCastChannel и передать ему значения:
val broadCastChannel = ConflatedBroadcastChannel<Int>()
Вы можете использовать broadCastChannel.offer(value) для отправки значений на канал.
Чтобы получить значения из канала, вы можете использовать простой цикл for-each:
for (i in broadCastChannel.openSubscription()) {
//your values
}
См. github.com/kotlin/kotlinx.coroutines/blob/master/reactive/….