如何在 Kotlin CoRoutine 中编写 rx concatArrayEager 等效项？

Question

我想将我的 rxJava 代码转换为 Kotlin CoRoutine。

下面的代码同时调用了 api 和 db 并将数据返回给 UI，无论哪个先出现。让我们说如果数据库响应恰好比 api 更快。在这种情况下，api 响应将继续，直到它接收到与 db 同步的数据，尽管它本可以更早地完成 UI 更新。

我该怎么做？

class MoviesRepository @Inject constructor(val apiInterface: ApiInterface,
                                        val MoviesDao: MoviesDao) {

fun getMovies(): Observable<List<Movie>> {
    val observableFromApi = getMoviesFromApi()
    val observableFromDb = getMoviesFromDb()
    return Observable.concatArrayEager(observableFromApi, observableFromDb)
}

fun getMoviesFromApi(): Observable<List<Movie>> {

    return apiInterface.getMovies()
            .doOnNext { it ->
                it.data?.let { it1 -> MoviesDao.insertAllMovies(it1) }
                println("Size of Movies from API %d", it.data?.size)
            }
            .map({ r -> r.data })
}

fun getMoviesFromDb(): Observable<List<Movie>> {
    return MoviesDao.queryMovies()
            .toObservable()
            .doOnNext {
                //Print log it.size :)
            }
}

}

Answer 1

作为第一步，您应该为您的ApiInterface 和MovieDao 调用创建suspend funs。如果他们有一些基于回调的 API，你可以关注these official instructions。

你现在应该有

suspend fun ApiInterface.suspendGetMovies(): List<Movie>

和

suspend fun MoviesDao.suspendQueryMovies(): List<Movie>

现在你可以编写这段代码了：

launch(UI) {
    val fromNetwork = async(UI) { apiInterface.suspendGetMovies() }
    val fromDb = async(UI) { MoviesDao.suspendQueryMovies() }
    select<List<Movie>> {
        fromNetwork.onAwait { it }
        fromDb.onAwait { it }
    }.also { movies ->
        // act on the movies
    }
}

亮点是 select 调用，它将同时在 Deferreds 上等待并在最先完成的那个上执行操作。

如果您想确保根据网络的结果采取行动，则需要更多代码，例如：

    val action = { movies: List<Movie> ->
        // act on the returned movie list
    }
    var gotNetworkResult = false
    select<List<Movie>> {
        fromNetwork.onAwait { gotNetworkResult = true; it }
        fromDb.onAwait { it }
    }.also(action)
    if (!gotNetworkResult) {
        action(fromNetwork.await())
    }

仅当数据库结果在网络结果之前出现时，此代码才会对数据库结果起作用，在所有情况下都会处理。

Answer 2

类似的东西应该可以工作：

data class Result(val fromApi: ???, val fromDB: ???)

fun getMovies(): Result {
    val apiRes = getMoviesFromApiAsync()
    val dbRes = getMoviesFromDbAsync()
    return Result(apiRes.await(), dbRes.await())
}

fun getMoviesFromApiAsync() = async {

    return apiInterface.getMovies()
            .doOnNext { it ->
                it.data?.let { it1 -> MoviesDao.insertAllMovies(it1) }
                println("Size of Movies from API %d", it.data?.size)
            }
            .map({ r -> r.data })
}

fun getMoviesFromDbAsync() = async {
    return MoviesDao.queryMovies()           
}

我不知道你要返回什么，所以我只输入了???。