javascript - 即使父源流仍然发出新值，具有合并功能的计时器也会停止发出值

Question

我尝试将应用程序的 3 个流状态合并到 1 个。循环是断开连接 <-> 连接 <-> 已验证。因此，只要应用程序仅连接到服务器，它就会尝试每 3 秒发送一次身份验证。除非已断开连接或已通过身份验证。

我正在使用 rxjs 计时器在要合并的流上发送身份验证，并使用 takeUntil 来停止发送身份验证(如果已通过身份验证)。但问题是，合并的 auth$ 没有完成 sendAuth 流，而是停止发出新的响应，即使待合并的 auth$ 仍在发出响应。代码如下:

this._response$ = fromEvent<string>(this._socket, "response")
  .pipe(map(data => JSON.parse(data)));

const disconnect$ = fromEvent(this._socket, "disconnect");
const connect$ = fromEvent(this._socket, "connect");
const auth$ = this._response$.pipe(
  // this stream still emitting even if takeUntil is declared on connect$.
  filter(res => res.action === "authentication" && res.type === "success"),
);

status$ = merge(
  disconnect$.pipe(
    map(() => {
      this._status = statusEnum.OFF;
      return this._status;
    })
  ),
  connect$.pipe(
    // will send auth each 3s after connected
    // until it is authenticated or disconnected
    switchMap(() => timer(0, 3000).pipe(
      tap(() => console.log('Sending auth right after connect')),
      map(() => this._sendAuth()),
      // completing the status$ stream instead of switchMap
      takeUntil(auth$),
    )),
    map(() => {
      this._status = statusEnum.ON;
      return this._status;
    }),
  ),
  auth$.pipe(
    // this stream stop emitting if takeUntil is declared on connect$.
    map(() => {
      this._status = statusEnum.AUTHENTICATED;
      return this._status;
    })
  ),
);

编辑:将共享运算符添加到要合并的身份验证管道后，合并的身份验证$再次开始发出值。

const auth$ = this._response$.pipe(
  // this stream still emitting even if takeUntil is declared on connect$.
  filter(res => res.action === "authentication" && res.type === "success"),
  share(),
);

Answer 1

// assume a isConnect stream which return a boolean indicate connection status
let isConnect$ = new BehaviorSubject(false);

// assume a behavior subject to store a auth status
let isAuth$ = new BehaviorSubject(false);

disconnect$.subscribe(() => this.isConnect$.next(false));

connect$.subscribe(() => this.isConnect$.next(true));

// combine isConnect and isAuth together
let status$ = combineLatest(isConnect$, isAuth$).pipe(
  map(() => (isAuth ? statusEnum.AUTHENTICATED : isConnect ? statusEnum.ON : statusEnum.OFF)),
);

// a timer for sending auth every 3 secs and only send if status is statusEnum.ON;
let auth$ = timer(0, 3000)
  .pipe(
    switchMapTo(status$),
    filter(status => status === statusEnum.ON),
  )
  .subscribe(() => this.sendAuth());

// in this.sendAuth, after successfully get auth, you have to make isAuth$.next(true);