Swift Combine sink 在第一个错误后停止接收值

Question

我正在将我的项目从 RxSwift 转移到 Combine 我有一个逻辑，我希望发布者每次单击按钮时都发出事件。实际点击按钮执行pushMe.send()

pushMe
            .print("Debug")
            .flatMap { (res) -> AnyPublisher<Bool, Error> in
                return Future<Bool, Error>.init { closure in
                    closure(.failure(Errors.validationFail))
                }.eraseToAnyPublisher()
            }
            .sink(receiveCompletion: { completion in
                print("Completion received")
            }, receiveValue: { value in
                print("Value = \(value)")
            })
            .store(in: &subscriptions)

控制台结果

Debug: receive value: (true)
Completion received
Debug: receive value: (true)
Debug: receive value: (true)

我不明白为什么接收器只在第一个事件上收到错误。其余的点击被忽略。

Answer 1

flatMap 是做什么的 -

1. 订阅给定的发布者（比如 XPublisher）。
2. 发送发出的错误和输出值（未完成事件/完成） 通过 XPublisher 到下游。

所以如果你在 flat map 内部处理错误，（这意味着 flatMap 内部的发布者不会发出错误），那么 flatMap 永远不会向下游发送错误。

pushMe
    .print("Debug")
    .flatMap { (res) -> AnyPublisher<Bool, Never> in //<= here
        return Future<Bool, Error>.init { closure in
            closure(.failure(Errors.validationFail))
        }
        .replaceError(with: false) //<= here
        .eraseToAnyPublisher()
    }
    .sink(receiveCompletion: { completion in
        print("Completion received")
    }, receiveValue: { value in
        print("Value = \(value)")
    })
    .store(in: &subscriptions)

否则，您可以在 fatMap 之外处理错误。这里的问题是，一旦出现错误，整个订阅/可取消取消。 （在下面的示例中，错误已替换为 false 值）

pushMe
    .print("Debug")
    .flatMap { (res) -> AnyPublisher<Bool, Error> in 
        return Future<Bool, Error>.init { closure in
            closure(.failure(Errors.validationFail))
        }
        .eraseToAnyPublisher()
    }
    .replaceError(with: false) //<= here
    .sink(receiveCompletion: { completion in
        print("Completion received")
    }, receiveValue: { value in
        print("Value = \(value)")
    })
    .store(in: &subscriptions)

上面的代码发生了什么。

1. FlatMap error 出局。
2. 将错误替换为 false（因此将收到一个 false 值）
3. 订阅因流中的错误而取消。

Answer 2

规则是，如果错误沿管道传播，则取消整个管道。因此，如果您的 Future 产生错误，它会作为错误传递给 Sink，因此管道会一直取消到 Publisher。

防止这种情况的模式是处理FlatMap 内部的错误。基本上，这里有两个 管道：一个以pushMe 开头，一个以Future 开头。只是不要让 Future 管道产生的错误“泄漏”到pushMe 管道中，这样pushMe 管道就不会被取消。相反，在 FlatMap 中捕获错误，如果您想将某些内容传递给您的 Sink，请从其中传递某种 value 来告诉您的 Sink 输入错误。

在您的情况下，一个简单的解决方案是将您的 FlatMap 类型更改为 <Bool,Never>，并将 true 或 false 作为 Bool 传递以指示将来验证是否成功。

或者，如果将有关错误的更多详细信息传递到管道中对您很重要，请将您的 FlatMap 的类型更改为 <Result<Bool,Error>,Never> 并将错误信息打包到 Result 对象的 .failure 大小写中。

Answer 3

这就是Publishers 在 Combine 中的工作方式。

Publisher 可以发出值或发出完成事件 - 一旦发出完成事件，Publisher 就完成了，它不能发出任何其他值或另一个完成事件。如果Publisher 发出错误，则错误将作为failure 完成发出，这意味着一旦发出错误，Publisher 就会完成并且它不能发出更多值。

有几个组合运算符设计用于在不完成 Publisher 的情况下处理错误。例如，查看catch 运算符。

Answer 4

首先，感谢大家对这个问题的帮助。

@matt 的答案是可能的解决方案之一。 另一种解决方案是每次单击按钮时创建新管道。 我使用这种方法是因为我在失败的发布者下面有一系列步骤，并且我无法进一步依赖虚拟的真/假结果。

      Just<String>()
            .sink(receiveValue: { value in
                startProcess()
                        .sink(receiveCompletion: { (completion:
                                Subscribers.Completion<Failure>) in
                            // can handle ALL types of error of pipe line in ONE place
                        }, receiveValue: { (v: P.Output) in
                            // handle correct result           
                        })
            })
            .store(in: &subscriptions)

    func startProcess() -> AnyPublisher<Bool, Error> {
        Future<Bool, Error>.init { closure in
                    // action 1 
                    closure(.success(true))
                }
                .flatMap { (b: Bool) -> AnyPubilsher<Void, Error> in
                    Future<Bool, Error>.init { closure in
                        // action 2
                        closure(.success(()))
                    }
                }
    }

好处是你可以在一个地方处理所有类型的错误，如果第二个 sink()