标签: kotlin-flow

我得到了一个类型流，我需要从中Flow<List<T>>得到一个类型流。Flow<T>

我试过：

outerFlow.onEach { items -> items.onEach { flow { emit(it) } }

这是行不通的。我怎样才能实现这个目标？

我有一个Flow<List<Int?>>，我想收集这个流，但直到我得到一个空的 Int。然后流程应该被取消。例如，

val flow = flowOf(
    listOf(1, 2),
    listOf(3, null),
    listOf(4)
)
flow.collectUntilNull {
    println(it)
}

我想要的输出是：

[1, 2]
[3, null]

我知道有一个函数Flow.takeWhile，但它不会发出谓词返回 false 的值。就我而言，我也想要那个。

[1, 2]
[3, null]

由于collectWhile是内部的，我无法使用此代码。虽然我想我可以复制粘贴该collectWhile实现到我的代码中。但还有其他方法可以做到这一点吗？

我正在尝试combine()组合 3 个流程，但没有任何反应。这是我的代码：

combine(flowX, flowY, flowZ) { x, y, z ->
    println("$x, $y, $z") // this line is never reached
}

我知道我的流程很好，因为它有效：

val x = flowX.first()
val y = flowY.first()
val z = flowZ.first()

println("$x, $y, $z") // this line is executed and prints the right values

为什么我的流量会发出 with ，first()但不会发出combine()？

我有一个下载文件的流程。该流程发出下载进度。我想在手机通知中心显示进度，但每秒只更新一次值，以防止设备滞后。

我的流程：

return callbackFlow {
  someJavaCallback { progress ->
    trySend(progress)
  }

  close()
}

我的 CoroutineWorker，显示通知并下载文件：

myFlow.collect { // update notification }
Result.Success()

我的问题是，如何“限制”收集，例如收集 1%，但 1 秒后它收集 5% 并忽略这两点之间的所有值

我需要在 MVVM 中获取用户身份验证状态。在存储库中我这样做：

override fun getAuthResponse() = callbackFlow  {
    val listener = AuthStateListener {
        Log.d(TAG, "currentUser: " + (currentUser == null)) //Line2
        trySend(it.currentUser == null)
    }
    auth.addAuthStateListener(listener)
    awaitClose {
        auth.removeAuthStateListener(listener)
    }
}

“Line2”将始终打印，true因为用户未经过身份验证。然后在 ViewModel 中我有：

fun getAuthResponse() = repo.getAuthResponse()

以及内部活动：

setContent {
    //...
    val response = viewModel.getAuthResponse().collectAsState(initial = false).value
    Log.d(TAG, "response: $response") //Line1
}

由于 setContent 是一个可组合函数，因此当我打开应用程序时，它会触发两次。这意味着“Line1”处的日志语句被触发两次。当它第一次点火时，我得到：

response: false
currentUser: true

因此，即使我在前一行调用了 getAuthResponse()，响应也会在之前打印。问题是，由于某种原因，即使当前用户为空，我也会在活动中打印出该用户不为空。当它第二次触发时，我得到了正确的数据：

response: true
currentUser: true

为什么我得到一个非空的用户对象？trySend 会发出假数据吗？

我有2个流量。第一个流每 50 毫秒更新一次。我有第二个流，它等于第一个流，但我希望它每 300 毫秒从原始流生成最新值。我找到了debounce流程扩展，但它不起作用（文档中的注释）：

Note that the resulting flow does not emit anything as long as the original flow emits items faster than every timeoutMillis milliseconds.

因此，当我尝试每 300 毫秒发出一次带有去抖动的值时，它根本不会发出，因为原始流比我需要的要快。那么我怎样才能做这样的事情：

1. 延迟300ms
2. 检查原始流量的最新值
3. 发出这个值
4. 重复

我现在的流程：

// fast flow (50ms)
val orientationFlow = merge(_orientationFlow, locationFlow)

val cameraBearingFlow = orientationFlow.debounce(300ms)

PS 这种方法不适合，因为我们延迟了值，所以 300 毫秒后它就不再新鲜了。我需要在 300 毫秒后获取最新值：

val cameraBearingFlow = azimuthFlow.onEach { 
        delay(ORIENTATION_UPDATE_DELAY)
        it
    }

我在流程内发出异常并得到以下异常。

IllegalStateException: Flow exception transparency is violated:
    Previous 'emit' call has thrown exception java.lang.NullPointerException, but then emission attempt of value 'planetbeyond.domain.api.Resource$Error@85b4d28' has been detected.
    Emissions from 'catch' blocks are prohibited in order to avoid unspecified behaviour, 'Flow.catch' operator can be used instead.
    For a more detailed explanation, please refer to Flow documentation.
       at kotlinx.coroutines.flow.internal.SafeCollector.exceptionTransparencyViolated(SafeCollector.kt:140)
       at kotlinx.coroutines.flow.internal.SafeCollector.checkContext(SafeCollector.kt:104)
       at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:83)
       at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:66)
       at planetbeyond.domain.use_cases.OptionSelectedCountUsecase$invoke$1.invokeSuspend(OptionSelectedCountUsecase.kt:20)

OptionSelectedCountUsecase.kt

class OptionSelectedCountUsecase @Inject constructor(
private val repository: Repository
) {
    operator fun invoke(questionId: Int): Flow<Resource<List<OptionSelectedCountModel>>> = flow …

我使用以下代码行从可组合函数内的房间数据库中提取数据：

val messagesUiState：MessagesUiState 通过 messageViewModel.allMessages.collectAsStateWithLifecycle()

如果我将视图模型属性更改为函数，我会从流中获得多次发射，并且 compose 将重新渲染多次，导致 ui 闪烁：

val messagesUiState: MessagesUiState by messageViewModel.allMessages().collectAsStateWithLifecycle()

我想使用函数的原因是这样我可以将参数传递给 Room 数据库。

ViewModel 内的 allMessages

 val allMessages: StateFlow<MessagesUiState> = chatRepository.getAllMessages()
        .map(MessagesUiState::Success)
        .stateIn(scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5_000),
            MessagesUiState.Loading)

房间查询：

@Query("SELECT * FROM message")
    fun getAll(): Flow<List<Message>>

当我按如下方式更改所有消息时，我会收到多次发射并多次重新渲染。我希望它只发射一次，以便用户界面不会随着每次发射而闪烁：

fun allMessages(): StateFlow<MessagesUiState> = chatRepository.getAllMessages()
        .map(MessagesUiState::Success)
        .stateIn(scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5_000),
            MessagesUiState.Loading)

和

val messagesUiState: MessagesUiState by messageViewModel.allMessages().collectAsStateWithLifecycle()

谁能帮助我理解这种行为？

我在某处读到这是因为 onEach 没有根据发出的值执行任何处理，但它只是改变它们；但是collect很可能对收集到的值执行一些异步或繁重的任务，这可能会阻塞线程。

这对我来说似乎不正确；如果我们对 onEach 函数中的值执行一些繁重的任务并且它有可能阻塞线程怎么办？

在不接收更新的情况下获取流发出的最后一个元素的最佳方法是什么。

问题： 我使用流来观察某些共享偏好的变化，但有时我想知道该偏好的当前值。我总是使用两个函数，一个用于观察流中的值，另一个用于直接捕获数据，有没有办法仅使用观察器函数来存档相同的行为？

suspend fun getBreadcrumb(): Breadcrumb =
        withContext(Dispatchers.IO) context@ {
            ...
        }

fun observeBreadcrumb(): Flow<Breadcrumb> {
        ....
    }