RXJS flatMap 到重复的 observable答案

【问题标题】：RXJS flatMap to repetitive observableRXJS flatMap 到重复的 observable
【发布时间】：2020-03-05 23:49:04
【问题描述】：

我正在尝试实现服务，如果应用程序连接到我的服务器，它会提供 observable，所以当浏览器在线时，我们会使用计时器 ping 服务器。下面是代码：

public get $connected(): Observable<boolean> {
    return this.hasInternetConnection
               .asObservable()
               .pipe(
                 distinctUntilChanged(),
                 flatMap((connected: boolean) => {
                   if (!connected) {
                     return of(connected);
                   } else {
                     return timer(5000)
                       .pipe(
                         map(() => {
                           var success = Math.random() > 0.5;
                           console.log('PING: ' + success);
                           return success;
                         })
                       );
                   }
                 })
               );
  }

hasInternetConnection 只是一个绑定到窗口 online 和 offline 事件的 BehaviorSubject，计时器模拟对我的 API 服务器的 ping。

问题是我的订阅 $connected 仅从可观察的计时器中捕获第一个值，然后不起作用。在hasInternetConnection 主题更改为false 并返回true 之后，我的订阅再次获得第一个价值，然后什么也没有。这是我在控制台中看到的：

PING: true
subscription tap
PING: true
PING: false
PING: true
...

我该如何解决这个问题？谢谢！

【问题讨论】：

示例 2：计时器在 1 秒后发出，然后每 2 秒发出一次； "const source = timer(1000, 2000);"

标签： rxjs observable ngrx rxjs-pipeable-operators rxjs-observables

【解决方案1】：

完整解决方案：

  private hasInternetConnection: BehaviorSubject<boolean> = new BehaviorSubject<boolean>(navigator.onLine);
  private connectedSubject: BehaviorSubject<boolean> = new BehaviorSubject<boolean>(true);
  private recheckConnectionSubject: Subject<void> = new Subject<void>();

  constructor(
    private readonly http: HttpClient,
  ) {
    fromEvent(window, 'online')
      .pipe(takeUntil(this.destroyed))
      .subscribe(() => {
        this.hasInternetConnection.next(true);
      });
    fromEvent(window, 'offline')
      .pipe(takeUntil(this.destroyed))
      .subscribe(() => {
        this.hasInternetConnection.next(false);
      });
    merge(
      this.hasInternetConnection,
      this.recheckConnectionSubject,
    )
      .pipe(
        mapTo(this.hasInternetConnection.value),
        switchMap((connected: boolean) => {
          if (!connected) {
            return of(connected);
          } else {
            return timer(0, 30000)
              .pipe(
                mergeMapTo(this.http.get(`${environment.apiRoot}/ping`, { responseType: 'text' })
                               .pipe(
                                 map((res) => {
                                   return true;
                                 }),
                                 catchError(() => {
                                   return of(false);
                                 })
                               )
                ),
              );
          }
        })
      )
      .subscribe(this.connectedSubject);
  }

  public get $connected(): Observable<boolean> {
    return this.connectedSubject.asObservable()
               .pipe(
                 distinctUntilChanged(),
               );
  }

  public resetTimer(): void {
      this.recheckConnectionSubject.next();
  }

【讨论】：