定时刷新Kotlin流(轮询)

Question

我应该如何定期"刷新";Kotlin流，获取新的数据。

我有一个存储在房间数据库中的图书id列表。从网络api中，我使用存储的id获得实际的书籍。

我的目标是监听数据库的变化，并将书id映射到实际的书。

问题是，我应该如何从网络api定期重新获取图书数据，但仍然保持侦听数据库所有的时间?因此，数据库中图书id的更改应该总是触发网络api调用(bookapi .get())。

fun getBookesFlow(): Flow<List<Book>> {
return booksDao.all().map { ids -> 
booksApi.get(ids)
}
}
fun getPollingFlow(): Flow<List<Book>> {
return flow {
while(true) {
// Reset the flow every 30 seconds
getBookesFlow().collect { bookes -> // This is just idea, not working code
emit(books)
}
delay(30.seconds)
}
}
}
fun getBooks(): {
viewModelScope.launch {
getPollingFlow().collect { books ->
_uiState.update { it.copy( books = books) }
}
}
}

Answer 1

我们不能强制流提供值，因此轮询并不真正适用于它们。相反，我们在其他流之上创建流来提供我们需要的行为。

在您的情况下，我们需要一个流来观察booksDao.all()，然后它将定期重新发出相同的数据。我们可以使用transformLatest():

轻松地做到这一点

fun getBookesFlow(): Flow<List<Book>> {
return booksDao.all()
.transformLatest {
while (true) {
emit(it)
delay(30.seconds)
}
}.map { ids -> 
booksApi.get(ids)
}
}

booksDao.all()每当有一个新的项目,我们开始无限循环,发出相同的数据到下游流。对于每个新项，前一个循环被取消，因此我们只发出最新的项。

更好的是，我们可以创建一个流操作符来完成上面的操作:

fun getBookesFlow(): Flow<List<Book>> {
return booksDao.all()
.emitLatestPeriodically(30.seconds)
.map { ids -> 
booksApi.get(ids)
}
}
fun <T> Flow<T>.emitLatestPeriodically(interval: Duration): Flow<T> = transformLatest {
while (true) {
emit(it)
delay(interval)
}
}

如果你不喜欢使用实验函数(虽然它们相当稳定)，那么你可以使用flatMapLatest()代替。

Answer 2

您可以通过将数据库更新和30秒计时器作为更新源来建模。要合并两个流，使用combine。

private val timerFlow = flow {
while (currentCoroutineContext().isActive)
{
emit(Unit)
delay(30.seconds)
}
}
fun getPollingFlow(): Flow<List<Book>> =
timerFlow.combine(booksDao.all()) { _, ids ->
booksApi.get(ids)
}.distinctUntilChanged()