如何阻止 Flowable 发射更多项目答案

【问题标题】：How to stop a Flowable from emitting more items如何阻止 Flowable 发射更多项目
【发布时间】：2020-01-12 19:00:21
【问题描述】：

我有一个关于Flowables 的问题。对于这个问题，我已经有了一些解决方案，但我想仔细检查这些是否是最好的解决方案。

上下文

我有一个交互器，它应该为数据库上的食谱添加书签。它看起来像这样：

    /**
     * This Interactor marks a recipe as "bookmarked" on the DB. The Interactor actually switches
     * the isBookmarked value of the related recipeId. If it was marked as true, it switches its value
     * to false. If it was false, then it switches its value to true.
     */
    class BookmarkRecipeInteractorImpl(
        private val recipesCacheRepository: RecipesCacheRepository
    ) : BookmarkRecipeInteractor {
        override fun execute(recipeId: Int, callback: BookmarkRecipeInteractor.Callback) {
            // Fetches the recipe from DB. The getRecipeById(recipeId) function returns a Flowable.
            // Internally, within the RecipesCacheRepository, I'm using room.
            recipesCacheRepository.getRecipeById(recipeId).flatMap { originalRecipe ->
                // Switches the isBookmarked value
                val updatedRecipe = originalRecipe.copy(
                    isBookmarked = !originalRecipe.isBookmarked
                )
                // Update the DB
                recipesCacheRepository.updateRecipe(updatedRecipe)
                // Here's the issue, since I'm updating a DB record and the getRecipeById returns
                // a Flowable, as soon as I update the DB, the getRecipeById is going to get triggered
                // again, and switch the value again, and again, and again...
            }
            .subscribe(
                {
                    callback.onSuccessfullyBookmarkedRecipe(it.response)
                },
                {
                    callback.onErrorFetchingRecipes()
                }
            )
        }
    }

因此，如果您按照代码进行操作，则错误非常简单。我陷入了一个循环，我不断地更改食谱记录。

可能的解决方案

1) 在我的 DAO 上有两个不同的函数，一个叫做 getRecipeByIdFlowable(id)，它返回一个 Flowable，另一个叫做 getRecipeByIdSingle(id)，它返回一个 rx.Single。这样我就可以通过存储库公开getRecipeByIdSingle(id) 并使用它而不是返回Flowable 的函数。这样我就切断了循环。

专业人士：它有效。

Con：我不喜欢在我的 DAO 上使用这样的功能。

2) 将Disposable 保存在lateinit 属性上，并在订阅者触发onNext() 时立即处理它。

专业人士：它有效。

Con：我不喜欢做这样的事情，感觉很老套。

3) 使用...getRecipeById(recipeId).take(1).flatMap...，所以它只处理第一个发射的对象。

专业人士：它有效，看起来很整洁。

Con：我不确定是否有更好的方法。

问题

理想情况下，我想调用一些函数，只允许我禁用Flowable 行为，并防止它在数据库更改时发出更多项目。到目前为止，我最喜欢的解决方案是#3，但我不确定这是否是正确的方法。

谢谢！

编辑 1

我只是在此处添加有关用例的更多信息。我需要一个给定 recipeId 的交互器，将 DB 上的 isBookmarked 值更改为相反的值。

数据库记录如下：

    data class DbRecipeDto(
        @PrimaryKey
        val id: Int,
        val name: String,
        val ingredients: List<String>,
        val isBookmarked: Boolean = false
    )

我知道也许还有其他一些方法可以让我以不同的方式解决这个问题。也许我可以传递 recipeId 参数和书签（布尔）参数，然后运行更新查询。

但是这个用例完全是虚构的，只是一个例子;我试图弄清楚如果数据库发生变化，如何防止 Flowable 发出更多项目。

【问题讨论】：

不确定您的用例是什么，但我只是想知道：为什么在从房间检索数据库的同时需要更新数据库？难道你不能有一个单独的 observable 来更新 db 吗？
@ChristilynArjona 谢谢，我编辑了我的问题以提供有关用例的更多信息。我不确定我是否理解您所说的，但我需要获取 DbRecipeDTO 记录以了解其isBookmarked 值，然后更新数据库。如果您还有其他需要知道的，或者我没有在这里解释的，请告诉我。
选项 3：take(1).
我做 RX 已经几个月了，但很确定如果你想让你的 flowable 停止工作（并通过代理，停止发射项目），你应该处理它......当然，你可以take(1) from the stream，这意味着您将创建一个 flowable，它仅从上游 flowable 中发出第一个值，但是您不妨用一个 no 对整个流进行建模？
@ThomasCook 是的，我在这里没有解释的事情（因为我不想让这变得太混乱）是我在其他地方使用getRecipesById()，实际上我需要它表现为 Flowable 并在每次行更改时发出项目。话虽这么说，我不想在我的 DAO 中创建两个函数（选项 #1）我想重用这个函数并让它表现得像一个单一的。

标签： android rx-java android-room

【解决方案1】：

您应该在getRecipeById(recipeId) 的末尾调用.take(1).singleOrError()。

这将获取通过调用getRecipeById 检索到的Flowable 发出的第一项（或错误）并将其包装在Single 中。在我看来，这与您想要实现的语义正确匹配。

此外，如果我没记错的话，因为您将通过这样做订阅Single，所以在下游调用singleOrError 消耗第一项后，您的Flowable 将不会继续工作。

【讨论】：