为什么 forkJoin 不适用于观察者？

Question

我尝试模拟服务器请求：

const _blocks$ = new BehaviorSubject<any[]>([]);
const _fields$ = new BehaviorSubject<any[]>([]);

然后我运行两个观察者：

 of(true)
      .pipe(delay(1000))
      .subscribe(() => {
        _blocks$.next(["blocks"]);
      });

    of(false)
      .pipe(delay(5000))
      .subscribe(() => {
        _fields$.next(["fields"]);
      });

在应用程序的某个地方，我尝试倾听并加入观察员：

forkJoin(_blocks$, _fields$).subscribe(
  (data) => {
    console.log(data);
  }
);

为什么执行后没有得到console.log(data);？

我的问题是监听对服务器的所有请求，并在它们完成后使用 forkJoin 加入结果。

Answer 1

因为forkJoin 等到所有流完成，而你的主题都没有。

您可以改用 combineLatest。

Answer 2

我的解决方案是：

let _blocks$ = new Observable<number>();
let _fields$ = new Observable<number>();

   _blocks$ = of(1).pipe(delay(1000));
    _fields$ = of(2).pipe(delay(5000));

    _blocks$.subscribe(() => {
      console.log("blocks");
    });

    _fields$.subscribe(() => {
      console.log("fields");
    });

    forkJoin(_blocks$, _fields$).subscribe(
      (data) => {
        console.log(data);
      }
    );

消费者可以在 _blocks$ 和 _fields$

上的应用程序中的任何位置订阅

Answer 3

您可以使用 take(1) 完成第一次发射后的 Observable，然后 forkJoin 将成功发射。

const _blocks$ = new BehaviorSubject<any[]>([]);
const _fields$ = new BehaviorSubject<any[]>([]);

const blocks$ = this._blocks$.asObservable().pipe(take(1));
const fields$ = this._fields$.asObservable().pipe(take(1));

forkJoin(blocks$, fields$).subscribe(console.log);

为什么 forkJoin 不适用于观察者？

Why forkJoin is not working for observers?

rxjs

angular

angular5

angular8