标签: kotlin-flow

lifeCycleScope.launch {
    viewModel.oneItem.collect {
        println("one")
    }

    viewModel.twoItem.collectLatest {
        println("two")
    }
}

我尝试了这段代码，但它只打印“一个”，而且似乎第二个collectLatest不起作用。这是为什么？

我有一个类，它采用协程调度程序作为我正在尝试测试的参数。在我的测试中，我通常@Before在每次测试运行之前设置我的类

@OptIn(ExperimentalCoroutinesApi::class)
@Before
fun setup() = runTest {
    .....
    val dispatcher = StandardTestDispatcher(testScheduler)
    scheduleService = ScheduleService(dispatcher)
}

我有一个正在尝试运行的测试，其中有一个SharedFlow我想检查其值，因此我也将其runTest与该测试一起使用

@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun testFullScheduleCreation() = runTest{
    ......
    val data = scheduleService.scheduleChangedListener.first()
}

当我尝试运行测试时出现错误

检测到使用不同的调度程序。如果需要使用多个测试协程调度程序，请创建一个 TestCoroutineScheduler 并将其传递给每个测试协程调度程序。

该错误是由于我使用的，@Before但我不确定如何在不将该设置方法中的所有代码复制到每个测试的情况下修复该错误

我如何使用它并为其添加过期时间，这意味着如果流中存在相同的值，我们仍然会收集它，因为在发出前一个重复值之后distinctUntilChanged()它的时间超过了毫秒。expiry

flow { 
  emit("A")    // printed
  emit("B")    // printed
  emit("A")    // printed
  emit("A")    // NOT printed because duplicate
  delay(5000)
  emit("A")    // printed because 5 seconds elapsed which is more than expiry
}
  .distinctUntilChanged(expiry = 2000)
  .collect {
    println(it)
  }

我想打印这个：

A
B
A
A

这是测试它的代码：

  @Test
  fun `distinctUntilChanged works as expected`(): Unit = runBlocking {
    flow {
      emit("A")    // printed
      emit("B")    // printed
      emit("A")    // printed
      emit("A")    // NOT printed because duplicate
      delay(5000)
      emit("A")    // printed because …

当我想将 Kotlin Flows 与普通回调一起使用时，我一直在寻找合适的解决方案或最佳实践。我的用例是我编写了一个在内部使用 Kotlin Flow 的 kotlin 库，我必须假设用户将使用 Java。所以我认为最好的解决方案是为我的流方法重载一个基本的回调接口，并以collect这样的方式调用它：

class KotlinClass {

    interface Callback {
        fun onResult(result: Int)
    }

    private fun foo() = flow {
        for (i in 1..3) {
            emit(i)
        }
    }

    fun bar(callback: Callback) {
        runBlocking {
            foo().collect { callback.onResult(it) }
        }
    }

  private fun main() {
    bar(object : Callback {
        override fun onResult(result: Int) {
            TODO("Not yet implemented")
        }
    })
}

在我的 Java 应用程序中，我可以像这样简单地使用它：

public class JavaClass {

    public void main() {
        KotlinClass libraryClass …

我正在尝试使用 Flows 测试 Kotlin 实现。我使用 Kotest 进行测试。此代码有效：

视图模型：

val detectedFlow = flow<String> {
    emit("123")
    delay(10L)
    emit("123")
}

测试：

class ScanViewModelTest : StringSpec({
    "when the flow contains values they are emitted" {
        val detectedString = "123"
        val vm = ScanViewModel()
        launch {
            vm.detectedFlow.collect {
                it shouldBe detectedString
            }
        }
    }
})

但是，在真正的 ViewModel 中我需要向流添加值，所以我使用ConflatedBroadcastChannel如下：

private val _detectedValues = ConflatedBroadcastChannel<String>()
val detectedFlow = _detectedValues.asFlow()

suspend fun sendDetectedValue(detectedString: String) {
    _detectedValues.send(detectedString)
}

然后在测试中我尝试：

"when the flow contains values they are emitted" …

我有一个关于 kotlin 流合并的问题。看看下面的乐趣。

suspend fun method(filter: String): Flow<List<Model>> {

// Search.
val models: List<Model> = repo.getModels(filter)  // suspend function

// Get favorites
val favoritesFlow: Flow<List<Int>> = otherRepo.getFavorites()

// Return models as Flow, but mark/unmark every model as favorite when favoritesFlow is updated.
??? val result = models + favoritesFlow ????

    return result
}

我需要返回模型列表流，但是当最喜欢的流更改时，我必须将每个模型标记或取消标记为最喜欢的。你知道我该怎么做吗？

我有一个具有以下属性的 android viewmodel 类

private val _trainingNavigationEvents = MutableSharedFlow<NavigationEventTraining>(replay = 0)
    val trainingNavigationEvents = _trainingNavigationEvents.asSharedFlow()

fun navigate(navigationEvent: NavigationEventTraining) {
        viewModelScope.launch {
            _trainingNavigationEvents.emit(navigationEvent)
        }
    }

我正在使用 SharedFlow，因为它解决了 SingleLiveEvent 问题。

当我尝试对代码进行单元测试时，问题就出现了。我不知道如何使用涡轮机（或提供的基元）来使其工作。

    @ExperimentalTime
    @Test
    fun `navigate`() = runBlockingTest {
        viewModel.handleIntent(TrainingViewModel.TrainingIntent.ShowQuestions)

        viewModel.navigationEvents.test {
            assertEquals(
                TrainingViewModel.TrainingNavigationEvent.NavigateToQuestions::class,
                expectItem()::class
            )
            cancelAndConsumeRemainingEvents()
        }
    }

我得到

kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1000 ms

我知道 SharedFlow 永远不会完成，这可能是部分原因，但我一直无法找到如何执行此操作的任何示例。

我正在使用 Junit 5 并使用 TestCoroutineDispatcher 类扩展。

我正在 kotlin 中尝试一些流程，并问自己一个问题：如果流程中的其中一个操作抛出异常，即使我使用 .catch，我的流程也会被取消吗？

如果没有，即使在使用 .catch 时发生异常，如何取消流程？

例子

fun testFlow() = flow {
   emit("Test")
   emit(Exception("Error"))
   emit("Test2") // This should not be emitted
}.catch { e -> println("Showing UI $e") }

另一个例子

fun testFlow2() = flow {
   emit("Test")
   throw Exception("Error")
   emit("Test2") // This should not be emitted
}.catch { e -> println("Showing UI $e") }

我有一个通过内容提供商从数据库获取数据的流程。

fun getDataFlow(): Flow<Result> {
    return flow {
      emit(Result.Loading)

      // fetchAll() is the method to fetch data via contentResolover.query()
      val results = fetchAll()
      emit(Result.Success(categories))
    }.catch { e ->
      emit(Result.Error(e))
    }
  }

那么当ContentProvider数据发生变化（onChange被调用）时，如何触发重新获取数据呢？

val contentObserver = object : ContentObserver(null) {
      override fun onChange(selfChange: Boolean) {
        super.onChange(selfChange)
      }
    }

我们知道StateFlow和SharedFlow很热门。

\n

StateFlow 是一个热流\xe2\x80\x94，只要该流被收集\n或者垃圾收集根中存在对其的任何其他引用，它就会保留在内存中。

\n

\n

SharedFlow 是一个热流，它向所有从其收集的使用者发送值。\n

\n

水流本身是冷的。

\n

流是类似于序列的冷流......

\n

我有一个问题无法找到直接答案。LiveData 是热的还是冷的？