RxSwift：为啥 flatMapLatest 从不执行 onCompleted()？

Posted 2023-03-10

【中文标题】RxSwift：为啥 flatMapLatest 从不执行 onCompleted()？

【英文标题】：RxSwift: Why does flatMapLatest never execute onCompleted()?RxSwift：为什么 flatMapLatest 从不执行 onCompleted()？

【发布时间】：2018-08-09 18:48:07

【问题描述】：

在我的 Swift UIViewController 中，我尝试订阅变量类型的类成员，通过 flatMapLatest 调用运行它，然后在所有订阅者上执行 flatMapLatest 可观察对象中的 onCompleted() 调用。然而，虽然 onNext() 被调用，但 onCompleted() 永远不会被调用，我不知道为什么。

我的班级成员定义为：

private let privateVar = Variable<String>("")

在我的 viewDidLoad() 方法中，我设置了 observables：

let localVar = self.privateVar.asObservable().distinctUntilChanged()

localVar.subscribe(onNext:  [weak self] sent in print("first onNext called") )
        .disposed(by: self.disposeBag)

let mappedVar = localVar.flatMapLatest  self.ajaxLoad(var1: $0) .share()

mappedVar.subscribe(
  onNext:  [weak self] queryRes in
    print("onNext called!")
  ,
  onCompleted:  [weak self] in
    print("onCompleted called!")
  
)
.disposed(by: self.disposeBag)

还有我的 ajaxLoad 方法：

func ajaxLoad(var1 myVar: String) -> Observable<QueryResponse> 
  return Observable.create  observable in
    apollo.fetch(query: MyQuery())  (result, _) in
      observable.onNext(result?.data?.myQuery)
      observable.onCompleted()
    

    return Disposables.create()
  

我对 ReactiveX 还很陌生，所以我可能对 Rx 生命周期的实际情况有些模糊。为什么在 flatMapLatest 调用中可能会调用 onNext，而不是 onCompleted？任何帮助，将不胜感激。提前致谢！

【参考方案1】：

flatMap 运算符不会发出您在块内返回的任何可观察对象的已完成事件。

以下代码清楚地说明了这一点。 .just(_) 发出元素，然后发出完成的事件，不会终止订阅。

_ = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
    .debug("before flatmap")
    .flatMap  .just($0 * 2) 
    .debug("after flatmap")
    .subscribe()

事实上，Variable 仅在释放时才发出完成。见source v4.0。 请注意，RxSwift 4 中不推荐使用 Variable，建议您改用 RxCocoa 的类似 BehaviorRelay。

deinit 
    _subject.on(.completed)

【讨论】：

但请注意，内部序列中未处理的错误将终止外部序列（flatMap .error(SomeError("an error")) ）

【参考方案2】：

既然你说你是新人，“有点朦胧”……

请记住，每当localVar 更改时，它都会发出一个新值并调用ajaxLoad(var1:)。 ajaxLoad(var1:) 的结果然后被推送到您订阅的 onNext 闭包。

还要记住，如果 Observable 发出 .completed，它就死了。它不能再发出任何东西。

所以flatMapLatest 不能完成（除非它的源代码完成。）如果它完成了，它会杀死整个管道并且不会对localVar 进行更多更改通过管道路由， ajaxLoad(var:1) 不会被新值再次调用，也不会被推送到订阅的 onNext 方法。

可以将一系列 observables 想象成一台 Rube Goldberg 机器，其中 completed 关闭机器，error 破坏机器。您应该关闭机器的唯一时间是 source（在本例中为 localVar）完成发出值或 sink（目标，在本例中为 onNext :closure) 不再需要任何值。