是否有适当的方法来链接协程流？

Question

我正在寻找一种方法来链接多个流，就像在 RxJava 中链接操作一样。这是当前实现的代码：

            driverPreferencesFlow
                    .stateIn(lifecycleScope)
                    .transform<DriverPreferences, Response<DriverCredentials>> { it ->
                        Log.d(App.TAG, "Got driver from cache, ask for driver from server")
                        repo.getDriver(it.driver.cell, it.driver.secret)
                    }
                    .onStart {
                    }
                    .onCompletion { e ->
                    }
                    .catch { e ->
                        Log.e(App.TAG, "Something went wrong on loading with driver", e)
                        Response.Error.Exception(e)
                    }
                    .collect { it ->
                        Log.d(App.TAG, "Got driver from server")
                        Log.d(App.TAG, "Collect new driver state ${it.javaClass}")
                        _newDriverState.value = it
                    }

在此实现中，调用了第二个操作/调用 (repo.getDriver())，但从未完成。似乎已暂停。

你有什么方法来完成类似的任务？

Answer 1

我不确定你为什么使用transform 而不是map 或flatMap，但根据documentation 你必须emit 转换函数中的值，所以我会假设它看起来像：

.transform<DriverPreferences, Response<DriverCredentials>> { it ->
    Log.d(App.TAG, "Got driver from cache, ask for driver from server")
    emit(repo.getDriver(it.driver.cell, it.driver.secret))
}

Answer 2

当您想从一个Flow 获取发射然后基于第一个的元素创建另一个时，您需要使用flatMap 操作（flatMapConcat/flatMapMerge），方法与中相同RxJava。这是一个包含两个整数流的简单示例，您可以在其中看到 flatMap 和 map 运算符的用法：

import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flatMapMerge
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

val a = flowOf(1, 2, 3)
val b = flowOf(3, 4, 5)

val result = a.flatMapConcat { intFromA ->
  b.map { intFromB ->
    intFromA + intFromB
  }
}

runBlocking {
  launch {
    result.collect {
      print(it)
    }
  }
}

是的，正如@romstn 正确指出的那样，transform 运算符需要手动调用 emit 函数，类似于 RxJava 中的 Observable.create()：https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/transform.html