问题标签 [kotlin-flow]

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

视图模型：

测试：

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

然后在测试中我尝试：

测试只是挂起并且永远不会完成。我尝试了各种各样的事情：launch或者runBlockingTest代替runBlocking，将发送和收集放在相同或单独的协同程序中，offer而不是send......似乎没有什么可以解决它。我究竟做错了什么？

更新：如果我手动创建流程，它可以工作：

那么，这是asFlow()方法中的错误吗？

我想读取文件内容并为其内容的每一行发出一个流。所以，我必须实现一个具有以下签名的函数：

有没有办法实现这个功能？

我的代码应该通过流将 SharedPreferences 更改为可观察的存储，所以我有这样的代码

或这个

然后我观察token、userId、companyId的这个偏好，然后登录但是有一个奇怪的事情，因为我需要构建应用程序三次，比如更改token不会导致tokenFlow发出任何东西，然后第二次新的userId不会导致userIdFlow发出任何东西，然后在第三次登录后我可以注销/登录并且它可以工作。注销时，我正在清除 prefs 令牌、userId、companyId 中的所有 3 个属性存储。

我使用 Spring Boot 和 WebFlux 运行基于 Netty 的 Kotlin 应用程序。详细情况如下：

• 爪哇 11;
• 科特林 1.3.61；
• Spring Boot 2.2.5.RELEASE；
• Spring Vault 核心 2.2.2.RELEASE。

我在 web 层上得到一个文件。WebFlux 从中创建一个Part( org.springframework.http.codec.multipart)。数据作为大小为 4Kb的块流存储在 Project ReactorFlux中：PartDataBuffer

由于符合框架的一致性，我将其转换Flux为 Kotlin 的Flow.

然后我在方法中使用同步 Vault 客户端encrypt(...)异步提交块（据我所知）flatMapMerge（注意encrypt(...)不是suspend，它是 HTTP 客户端顶部到远程加密提供程序的包装器）：

我已经检查了这个答案https://stackoverflow.com/a/58659423/6612401并发现基于流的方法应该与flow { emit(...)}.

我的问题是我可以使用这种基于流的方法而不使用suspend函数吗？或者有没有更好的方法，考虑到我正在使用runBlocking(Dispatchers.IO)一个suspend fold(...)函数。

代码如下：

PS 该fold(...)函数将打开的块收集到 aByteArrayOutputStream以稍后计算哈希，并将加密的块收集到 aStringBuilder作为加密文件的结果。

PPS 我已经尝试过我的方法。该方法在我的 Core i5 8gen 4 物理内核机器上平均提交 5-7 个并行请求。它完成了它的工作，但没有那么快。如果 Vault 不在本地部署，我每 1 Mb 加密大约需要 1 秒。我知道这取决于网络的延迟。我什至不考虑 Vault 一侧的加密速度，由于只有 4Kb 的块大小，它快如闪电。有什么方法可以提高并发速度吗？

PPPS 我试过concurrency = MAX_CONCURRENT_REQUESTS在flatMapMerge{...}. 到目前为止，没有什么显着的结果。最好将其保留为默认值。

以上是我的代码的简化版本。

我想要做的是，doWork()等到所有值都被释放，最后返回一个基于 last ProcessStatus.SUCCESSor的布尔值ProcessStatus.FAILURE。

现在，上面的代码发生的事情是，一旦processWork()返回流。doWork()调用包括single()and 在内的所有运算符，因为工作仍在进行中 ProcessStatus.FAILURE 或 ProcessStatus.SUCCESS 仍未发出，使其异常。

如何让doWork()return 语句等待并仅在流程完成时返回？

编辑1：

原因，我必须使用频道是因为，这是 Android 代码的一部分，channel.offer()实际上并不是像上面示例中那样的新协程，而是从 Android 调用BroadcastReceiver。

由于流程很冷，我不希望用户离开活动来阻止任务完成和通知。

最近作为 Kotlin 协程的一部分引入了该类。StateFlow

我目前正在尝试它并在尝试对我的 ViewModel 进行单元测试时遇到问题。我想要实现的目标：测试我StateFlow在 ViewModel 中以正确的顺序接收所有状态值。

我的代码如下：

视图模型：

我的测试：

访问发出的最后一个值有效。但我要测试的是所有发出的值都以正确的顺序发出。使用这段代码：viewModel.userWallet.toList(res) //doesn't works 我收到以下错误：

我想我错过了一些明显的东西。但不确定为什么当我刚刚开始使用 Coroutine 和 Flow 时，当不使用我已经使用的 runBlockingTest 时似乎会发生此错误。

编辑：作为临时解决方案，我将其作为实时数据进行测试：

我需要创建一个 API，它应该是流，它收集事件。问题是这些事件可能来自一个通道（我需要一个用于 PublishSubject 的模拟）和一个流（它执行网络请求）。

我也不确定这是否是最好的解决方案，所以让我知道我是否可以做得更好。

我在做什么：

我的 API：

所以 statusFlow 应该返回一个流，我可以从中接收来自流和通道的数据。

我试图通过consumeAsflow将通道转换为流，但它不起作用。

我看到一个解决方案

正确的方法是什么？

GlobalScope 或自定义 CoroutineScope 实例都不起作用：

绝对没有输出或抛出错误

为什么？

更奇怪的是，当我添加一个runBlocking{}块时，一切都会执行：

输出：

我想使用 Flow 作为存储库中所有函数的返回类型。例如：

此函数应调用 2 个数据源：远程（将数据保存在服务器上）和本地（将服务器返回的数据保存在本地）。问题是我如何实现这个场景：

1. 尝试使用 RemoteDataSource 保存数据
2. 如果 1. 失败 - 尝试 N 次 M 超时
3. 如果数据最终从服务器返回 - 在本地与 LocalDataSource 相同
4. 使用本地保存的数据返回流

RemoteDataSource 和 LocalDataSource 都有create相同的签名：

所以它们都返回数据流。如果您对如何实施它有任何想法，我将不胜感激。

------ 更新 #1 ------

可能解决方案的一部分：

这是一个可行的想法吗？

我正在从数据库中加载 Flow>。基于此结果，我想将每个元素 a 与另一个数据库调用转换为 Pair。

所以基本上是这样的：

调用 2 还返回一个带有 ElementB 的流。返回类型必须是 List>。

如何使用 Kotlin Flow API 实现这一目标？