Rxjs: Observables 和 Subjects 排序

Question

我有一个“用户”组件及其相关服务：userService。我想将组件表单用于新建和编辑。

在订阅服务主题时，用户组件与其他几个组件（国家和州）及其服务相关。

export class CountrystateService {

  private countries: Country[] = [];
  private countriesSubject = new Subject<Country[]>();
  countriesSubject$ = this.countriesSubject.asObservable();

  private allStates: State[];
  private states: State[];
  private statesSubject = new Subject<State[]>();
  statesSubject$ = this.statesSubject.asObservable();

  constructor(private apiService: ApiService) { }

  getCountriesAllStatesFromRest() {
    forkJoin({
        countries: this.apiService.getEntriesFromRest('countries').pipe(
            tap((countries: Country[])=> {this.countries=countries;})),
        states: this.apiService.getEntriesFromRest('countries').pipe(
            tap((states: State[])=> {this.allStates=states;}))
        }).subscribe(
            (results) => {
                this.countriesSubject.next(this.countries.slice());
            },
            (error) => {
                this.countriesSubject.next(error);
                return throwError(error);
            },
            () =>{}
        )
    }

    filterStatesForCountry(countryId?) {
    if (countryId) {
        this.states=this.allStates.filter(item => +item.country.split('/').pop() === countryId);
        this.statesSubject.next(this.states);
    }

  }

用户组件 ngOnInit：

  ngOnInit() {

    this.stateSubscription = this.countrystateService.statesSubject$.subscribe(
      (states: State[]) => {
        if (states) {
            this.states = states;       
        this.FooFunction();
        }
      },
      (error) => {..}
    );

    this.countrySubscription = this.countrystateService.countriesSubject$.subscribe(
      (countries: Country[]) => {
        if (countries) {
            this.countries = countries;         
        }
      },
      (error) => {...}
    );

    this.userSubscription = this.userService.userSubject$.subscribe(
      (user: User=> {
        if (user) {
            this.user = user;
        }
      },
      (error) => {...}
    );

在编辑上下文（参数中的用户 ID）中，我需要订购请求： - 获取用户 - 获取有关 user.country.id 的州列表

然后我尝试了这样的事情：

this.userIdSubscription = this.route.paramMap.subscribe(
    (params: ParamMap) => {
        const userId = +params.get('id');
        if (userId) {
            ...
            this.userService.getUserFromRest(userId);
            this.countrystateService.getCountriesAllStatesFromRest();
            forkJoin({
                user: this.userService.userSubject$,
                states: this.countrystateService.statesSubject$,
            }).pipe(
                tap(results => console.log('FORKJOIN TRIGGERED')),
                tap(results => {
                    ...
                    this.countrystateService.filterStatesForCountry(results.user.country.id);
                }),
            );
        }
        else { this.countrystateService.getCountriesAllStatesFromRest(); }
    }
);

但它无法正常工作... 谢谢你的帮助，

Answer 1

更新

经过调查，我们发现 Subjects$ 没有完成，因此 forkJoin 不适合，解决方案是改用 combineLatest().pipe(first())。

原创

正确的做法是单一订阅和构建管道，尝试使用下一种方法遵循 rx 风格。

this.userIdSubscription = this.route.paramMap.pipe(
  map((params: ParamMap) => +params.get('id')),
  filter(userId => !!userId),
  tap(userId => {
    // ...
    this.userService.getUserFromRest(userId);
    this.countrystateService.getCountriesAllStatesFromRest();
  }),
  switchMap(userId => forkJoin({
    user: this.userService.userSubject$,
    states: this.countrystateService.statesSubject$,
  })),
  tap(results => {
    this.countrystateService.filterStatesForCountry(results.user.country.id);
  }),
).subscribe();

现在 IMO 更具可读性。

---

你好像忘记订阅了。

            forkJoin({
                user: this.userService.userSubject$,
                states: this.countrystateService.statesSubject$,
            }).pipe(
                tap(results => console.log('FORKJOIN TRIGGERED')),
                tap(results => {
                    ...
                    this.countrystateService.filterStatesForCountry(results.user.country.id);
                }),
            ).subscribe(); // <- this place.

---

如果不是答案，那么看起来服务还没有完成，所以你不能使用 forkJoin，它希望所有的 observables 都完成。

尝试使用 combineLatest，它会在第一次发出后发出。

            combineLatest([ // <- this change.
                user: this.userService.userSubject$,
                states: this.countrystateService.statesSubject$,
            ]).pipe(
                first(), // <- we need only the first result?
                tap(results => console.log('FORKJOIN TRIGGERED')),
                tap(([user]) => {
                    ...
                    this.countrystateService.filterStatesForCountry(user.country.id);
                }),
            ).subscribe(); // <- this change too.

Answer 2

使用来自ActivatedRoute 的params 而不是来自Router 的paramMap

constructor(private aroute:Activated, ...) { ... }

this.userIdSubscription = this.aroute.params.subscribe(
    (params) => {
      const userId = params.id;
      ...

}

希望对你有帮助！