RxJs：当全部取消订阅时，中止一个延迟和共享的 observable

Question

我想创建一个运行长轮询操作的rxjs Observable。 每次迭代都会发出中间结果。 当isComplete 返回真时，Observable 完成。

这个函数的行为应该如下

1. 只有当有至少一个订阅者时才应该开始
2. 应该允许多个订阅者共享结果
3. 如果没有没有订阅者离开
，它应该中止轮询并取消呼叫

以下代码运行正常，满足条件（1）和（2）：

function longPollingAction(fetch: () => Promise<Response>, cancel: () => {}): Observable<Response> {
   return defer(() => { // defer to start running when there's a single subscriber
     return from(fetch()).pipe(
         expand(() => timer(1000).pipe(switchMap(fetch))),
         takeWhile<Response>(isComplete, false),
    );
   }).pipe(share()); // share to allow multiple subscribers
}

function isComplete(r: Response): boolean {
   // returns true if r is complete. 
}

如何修改此代码以满足 (3) 的要求？在当前实现中，轮询停止，但我如何调用cancel？

Answer 1

使用终结

您可以使用finalize 调用取消。这可能是这样的：

function longPollingAction(
  fetch: () => Promise<Response>,
  cancel: () => void
): Observable<Response> {
  // defer to turn eager promise into lazy observable
  return defer(fetch).pipe( 
    expand(() => timer(1000).pipe(switchMap(fetch))),
    takeWhile<Response>(isComplete, false),
    finalize(cancel),
    share() // share to allow multiple subscribers
  );
}

function isComplete(r: Response): boolean {
   // returns true if r is complete. 
}

回调complete

tap 操作员可以访问next、error 和complete 排放。对于callback: () => void，这就足够了。

function longPollingAction(
  fetch: () => Promise<Response>,
  cancel: () => void
): Observable<Response> {
  // defer to turn eager promise into lazy observable
  return defer(fetch).pipe( 
    expand(() => timer(1000).pipe(switchMap(fetch))),
    takeWhile<Response>(isComplete, false),
    tap({
      complete: cancel
    }),
    share() // share to allow multiple subscribers
  );
}

function isComplete(r: Response): boolean {
   // returns true if r is complete. 
}

回调unsubscribe

我不认为这样的运算符存在，但我们可以很容易地制作一个。如果取消订阅，此运算符只会触发回调。它将忽略error 和complete。

function onUnsubscribe<T>(
  fn: () => void
): MonoTypeOperatorFunction<T> {
  return s => new Observable(observer => {
    const bindOn = name => observer[name].bind(observer);
    const sub = s.subscribe({
      next: bindOn("next"),
      error: bindOn("error"),
      complete: bindOn("complete")
    });
   
    return {
      unsubscribe: () => {
        fn();
        sub.unsubscribe()
      }
    };
  });
}

那么你可以这样使用它：

function longPollingAction(
  fetch: () => Promise<Response>,
  cancel: () => void
): Observable<Response> {
  // defer to turn eager promise into lazy observable
  return defer(fetch).pipe( 
    expand(() => timer(1000).pipe(switchMap(fetch))),
    takeWhile<Response>(isComplete, false),
    onUnsubscribe(cancel),
    share() // share to allow multiple subscribers
  );
}

function isComplete(r: Response): boolean {
   // returns true if r is complete. 
}

由于share 正在管理您的订阅并且共享只会取消订阅一次refCount < 1，因此在这种情况下调用取消的唯一方法是没有订阅者。

Answer 2

给猫剥皮的方法不止一种，但我会这样做：

const onUnsubscribe = (callback: () => void) => <T>(source$: Observable<T>) =>
  new Observable<T>(observer => {
    let isSourceDone = false;

    const subscription = source$.subscribe(
      val => {
        observer.next(val);
      },
      e => {
        isSourceDone = true;
        observer.error(e);
      },
      () => {
        isSourceDone = true;
        observer.complete();
      }
    );

    return () => {
      if (isSourceDone) return;
      callback();
      subscription.unsubscribe();
    };
  });

function longPollingAction(
  fetch: () => Promise<Response>,
  cancel: () => {}
): Observable<Response> {
  const lazyFetch$ = defer(() => fetch());
  return lazyFetch$.pipe(
    expand(() => timer(1000).pipe(mergeMapTo(lazyFetch$))),
    takeWhile<Response>(isComplete, false),
    onUnsubscribe(cancel),
    share()
  );
}

function isComplete(r: Response): boolean {
  // returns true if r is complete.
}