javascript - RxJS 自定义运算符不调用完成

Question

我有以下自定义运算符:

export function test() {
  return function(source) {
    return new Observable(observer => {

      return source.subscribe({
        next(value) {
          observer.next(value);
        },
        error(error) {
          observer.error(error);
        },
        complete() {
          console.log('completed')
          observer.complete();
        }
      })
    });
  }
}

问题是，当我订阅一个完成的可观察对象时，我没有得到完整的回调，即我没有看到完成的日志。

interval(1000).pipe(
  test(),
  take(2)
).subscribe();

我错过了什么？

Answer 1

好问题，现在让我们深入解释一下。

首先，让我们看看解决方案

function test(limitter) {
  return function(source) {
    return new Observable(observer => {
      return source.subscribe({
        next(value) {
          observer.next(value);
        },
        error(error) {
          observer.error(error);
        },
        complete() {
          console.log("completed action");
          observer.complete("completed value");
        }
      });
    });
  };
}

interval(1000)
  .pipe(
    take(2),
    test()
  )
  .subscribe(
    x => {
      console.log("result: ", x);
    },
    err => {
      console.log(err);
    },
    end => {
      console.log("Observable has been completed");
    }
  );

那么有什么区别呢，在这个片段中，take 运算符先于自定义的 test() 运算符，这意味着每当我们达到所需的计数时(在我们的在 2 的情况下，take 运算符将返回 completed source，这将在后续订阅者中触发我们的 complete 方法(在我们的示例中，在自定义中test 运算符以及 subscribe 内)之后 source 将不会发出任何其他内容，因为它已经完成。

可以查看源码=> take() , 如果有一些模糊的部分，请随时询问更多信息。