forkJoin 不会等待所有可观察对象完成

Question

import { Component } from "@angular/core";
import { Observable, of, throwError, Subscription, forkJoin } from "rxjs";
import { mergeMap, map, delay, timeout, first, take } from "rxjs/operators";
import { ajax } from "rxjs/ajax";

class Test {
  id: number;
  firstObj: FirstObj;
  secondObj: SecondObj;
  thirdObj: string;
}

interface FirstObj {
  firstObs: boolean;
  result: string;
}

interface SecondObj {
  secondObs1: boolean;
  secondObs2: boolean;
}

@Component({
  selector: "my-app",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent {
  private workOrderSubscription: Subscription;
  name = "Angular";
  obsOne: Observable<any> = of("First Obs")
    .pipe(delay(6000))
    .pipe(
      map(res => {
        return {
          firstObs: true,
          result: res
        };
      })
    );

  dataTwo: SecondObj = { secondObs1: true, secondObs2: false };

  obsTwo: Observable<any> = of(this.dataTwo);
  obsThree: Observable<any> = of("error");

  private getId(): Observable<any> {
    return of("id" as string);
  }

  public retrieveWork(): Observable<Test> {
    const test: Test = new Test();
    this.getId().subscribe((id: number) => {
      test.id = id as number;
      forkJoin(this.obsOne, this.obsTwo, this.obsThree).subscribe(
        ([first, second, third]: [FirstObj, SecondObj, string]) => {
          // some appropriate checks here
          test.firstObj = first as FirstObj;
          test.secondObj = second as SecondObj;
          test.thirdObj = third as string;
          console.log("first is " + first.result);
          console.log("second is " + second.secondObs1);
          console.log("third is " + third);
          console.log(`inside ******************** ${JSON.stringify(test)}`);
          return of(test);
        },
        error => {
          //error here
          console.log("GOT ERROR");
        }
      );
    });

    console.log(`returning ******************** ${JSON.stringify(test)}`);
    return of(test);
  }

  ngOnInit() {
    console.log("data ************************");
    this.retrieveWork()
      .pipe(timeout(10000))
      .subscribe(data => {
        {
          console.log("printing data ***************************");
          console.log(`${JSON.stringify(data)}`);
        }
      });
  }
  ngOnDestroy() {}
}

我尝试了很多方法，基本上我正在尝试做的是在ngOnInit我订阅get()，但是我得到的数据并不完整。订阅不会等到forkJoin 完全返回。我不确定如何获得完整的数据。

在第一个订阅完成后它发出，我希望第一个和第二个订阅都完成，然后填写“测试”类，然后我的订阅者在ngOnInit 中获取数据。

我的输出是这样的

数据************************

返回 ******************** {"id":"id"}

打印数据***************************

{"id":"id"}

首先是第一个观察点

秒是真的

第三个是错误

内部 ******************** {"id":"id","firstObj":{"firstObs":true,"result":"First Obs"},"secondObj":{"secondObs1":true,"secondObs2":false},"thirdObj":"error"}

正如我们在上面的输出中看到的，只有 id 被打印（在第一次订阅之后）， 但是在第二次订阅后，ngOnInit 中的订阅者不再获取数据。

谢谢。

Answer 1

我用高阶映射运算符 (switchMap) 替换了嵌套的 Observable，它似乎可以工作。 （但我不太清楚这应该输出什么。）

这就是我所拥有的：

import { Component } from "@angular/core";
import { Observable, of, forkJoin } from "rxjs";
import { switchMap, map, delay, timeout} from "rxjs/operators";

class Test {
  id: number;
  firstObj: FirstObj;
  secondObj: SecondObj;
  thirdObj: string;
}

interface FirstObj {
  firstObs: boolean;
  result: string;
}

interface SecondObj {
  secondObs1: boolean;
  secondObs2: boolean;
}


@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent {
  title = "Angular";
  obsOne: Observable<any> = of("First Obs")
    .pipe(delay(6000))
    .pipe(
      map(res => {
        return {
          firstObs: true,
          result: res
        };
      })
    );

  dataTwo: SecondObj = { secondObs1: true, secondObs2: false };

  obsTwo: Observable<any> = of(this.dataTwo);
  obsThree: Observable<any> = of("error");


  // Not sure why this was sometimes a string and sometimes a number
  // Changed to a number.
  private getId(): Observable<number> {
    return of(7);
  }

  // Made this declarative so you could do further operations on the
  // stream as needed.
  myTest$ =
    this.getId()
      .pipe(
        // Added switchMap, which is a higher-order mapping operator
        // and automatically subscribes (and unsubscribes) to its inner Observables
        switchMap(id => {
          const test: Test = new Test();
          test.id = id;
          return forkJoin(this.obsOne, this.obsTwo, this.obsThree)
            .pipe(
              map(([first, second, third]: [FirstObj, SecondObj, string]) => {
                // some appropriate checks here
                test.firstObj = first as FirstObj;
                test.secondObj = second as SecondObj;
                test.thirdObj = third as string;
                console.log("first is " + first.result);
                console.log("second is " + second.secondObs1);
                console.log("third is " + third);
                console.log(`inside ******************** ${JSON.stringify(test)}`);
                return test;
              })
            )
        })
      );

  ngOnInit() {
    console.log("data ************************");
    this.myTest$
      .pipe(timeout(10000))
      .subscribe(data => {
        {
          console.log("printing data ***************************");
          console.log(`${JSON.stringify(data)}`);
        }
      });
  }

  ngOnDestroy() { }
}

这是否如您所愿？

Answer 2

最后这似乎奏效了。我是 RXJS 的新手。欢迎提出任何建议。

也处理了所有的错误情况。

import { Component } from "@angular/core";
import {
  Observable,
  of,
  throwError,
  Subscription,
  forkJoin,
  combineLatest
} from "rxjs";
import {
  mergeMap,
  map,
  delay,
  timeout,
  first,
  take,
  catchError
} from "rxjs/operators";
import { ajax } from "rxjs/ajax";

class Test {
  id: number;
  firstObj: FirstObj;
  secondObj: SecondObj;
}

interface FirstObj {
  firstObs: boolean;
  result: string;
}

interface SecondObj {
  secondObs1: boolean;
  secondObs2: boolean;
}

@Component({
  selector: "my-app",
  templateUrl: "./app.component.html",
  styleUrls: ["./app.component.css"]
})
export class AppComponent {
  private workOrderSubscription: Subscription;
  name = "Angular";

  dataone: FirstObj = { firstObs: true, result: "hai" };
  obsOne: Observable<any> = of(this.dataone).pipe(delay(1000));
  dataTwo: SecondObj = { secondObs1: true, secondObs2: false };
  obsTwo: Observable<any> = throwError(this.dataTwo);

  private getId(): Observable<any> {
    return of(1).pipe(delay(4000));
  }

  public get(): Observable<any> {
    let test: Test = new Test();
    return new Observable(subscriber =>
      this.getId().pipe(timeout(5000)).subscribe(
        (id: number) => {
          test.id = id;
          forkJoin(
            this.obsOne.pipe(timeout(1000)).pipe(catchError(error => of(null))),
            this.obsTwo.pipe(timeout(1000)).pipe(catchError(error => of(null))),
          ).subscribe(([first, second]) => {
            test.firstObj = first;
            test.secondObj = second;
            console.log("printing data ***************************2");
            console.log(`${JSON.stringify(test)}`);
            subscriber.next(test);
            subscriber.complete();
          });
        },
        err => {
          console.log("error while getting data");
          subscriber.error("error while getting data");
          subscriber.unsubscribe();
        }
      )
    );
  }

  ngOnInit() {
    let subscription: Subscription = this.get()
      .pipe(timeout(6000))
      .subscribe(
        (data: Test) => {
          console.log("in subscribe");
          console.log(`${JSON.stringify(data)}`);
        },
        err => {
          console.log("ERROR " + err);
          subscription.unsubscribe();
        }
      );
  }
  ngOnDestroy() {}
}