有没有一种有效的方法来跟踪 rxjs 中的订阅答案

【问题标题】：Is there an efficient way to keep track of Subscriptions in rxjs有没有一种有效的方法来跟踪 rxjs 中的订阅
【发布时间】：2019-12-08 16:58:56
【问题描述】：

假设我有很多包含这样结构的组件

ngOnInit() {
    this.fb.user$.subscribe(data => {
        if (data != null) {
          //User is logged in
          this.fb.afs.collection('col').doc(data.info).valueChanges().subscribe(info => {
            this.info = info;
          })
          //Lots of other subscriptions following...
        } else {
          //User is logged out
        }
    })
}

一旦用户注销，就会引发 Firebase 权限异常，因为

this.fb.afs.collection('col').doc(data.info).valueChanges().subscribe(info => {
  this.info = info;
})

不再允许订阅。

是否有任何其他方式可以取消订阅所有 firebase 订阅，而无需手动将所有订阅推送到数组并在用户注销之前循环它？

【问题讨论】：

标签： firebase google-cloud-firestore rxjs observable angularfire2

【解决方案1】：

也许将takeUntil 操作符传递给他们每个人？

destroyed = new Subject();
destroyed$ = this.destroyed.asObservable();

constructor() {
  resource1$.pipe(takeUntil(this.destroyed$)).subscribe(...);
  resource2$.pipe(takeUntil(this.destroyed$)).subscribe(...);
  resource3$.pipe(takeUntil(this.destroyed$)).subscribe(...);
}


ngOnDestroy() {
  this.destroyed.next();
  this.destroyed.complete();
}

takeUntil：发出值直到提供的 observable 发出。

【讨论】：

【解决方案2】：

首先，在rxjs中不推荐使用嵌套订阅，这样的逻辑使用switchMap操作符。

要管理组件中的订阅，您可以创建Subscription() 并在其上使用add 函数。请参见下面的示例：

import { Subscription } from 'rxjs';

@Component({...})
export class TestComponent implements OnInit, OnDestroy {

private readonly subscriptions = new Subscription()

ngOnInit() {
    const firstSubscription = stream1$.subscribe(...);
    const secondSubscription = stream2$.subscribe(...);
    this.subscriptions.add(firstSubscription);
    this.subscriptions.add(secondSubscription);
}

ngOnDestroy() {
    this.subscriptions.unsubscribe();
}

这里还有一个更复杂的流管理示例：

import { Subscription, of } from 'rxjs';
import { switchMap, tap } from 'rxjs/operators';

@Component({...})
export class TestComponent implements OnInit, OnDestroy {

ngOnInit() {
    const userSubscription = this.fb.user$.pipe(
        switchMap(data => {
            if (data) {
                return this.fb.afs.collection('col').doc(data.info).valueChanges();
            } else if (1 !== 2) { // for example
                return this.anotherService.getDataById(data.id).pipe(
                    tap((cat) => {
                        // here you can also do some intermediate logic
                        this.cats = [cat];
                    })
                )
            }
            // other things to be returned

            // don't forget to olways pass something. 
            // if nothing matched, return default value
            return of(null);
        })
    ).subscribe(info => {
        this.info = info;
    });
    this.subscriptions.add(userSubscription);
    const anotherSubscription = this.anotherService.mockMethod.subscribe(() => /** 
    something */);
    this.subscriptions.add(anotherSubscription);
}

ngOnDestroy() {
    this.subscriptions.unsubscribe();
}

}

所以这里你应该注意两点：

与您的问题相关：获取对订阅userSubscription 的引用（由.subscribe() 方法返回并将其添加到组件subscriptions。然后在ngOnDestroy 中取消订阅所有订阅组件。您可以添加任意数量的组件。
不要使用嵌套订阅。使用管道将允许您控制您的流并制作许多由 rxjs 提供的很酷的功能。这是延迟、过滤、映射和许多其他的。我建议您了解更多关于扁平化策略（flatMap、switchMap、concatMap 和 exhaustMap）的信息。查看这篇文章https://medium.com/@shairez/a-super-ninja-trick-to-learn-rxjss-switchmap-mergemap-concatmap-and-exhaustmap-forever-88e178a75f1b 并在那里观看视频。在我看来，这是对这个话题的最好解释。

【讨论】：