Angular 8 rxjs 从多个可观察对象中获取值

Question

我正在使用 rxjs switchMap 从主题中获取最新的表单值，设置一个局部变量，然后根据该值执行查询。

长话短说，我正在根据表单的当前值填充选择框选项。

  // Get current value of Record Form
  this.recordFormService.recordForm$
    .pipe(
      switchMap(recordForm => {
        // Get current Product Owner on Record Form
        this.currentProductOwner = recordForm.value.ProductOwner;

        // Search People
        return this.recordsApiService.searchPeople$(this.currentProductOwner);
      }),
      catchError(err => {
        console.error(`get recordForm$ failed ${err}`);
        return throwError(err);
      })
    )
    .subscribe(results => {
      // Set Product Owners select box options
      this.productOwners = results.data;
    });

一切正常。

但是，当特定表单值发生变化时，我还需要执行相同的逻辑。

// Run the same logic when the ProductOwner value changes
this.parentFormGroup.controls['ProductOwner'].valueChanges.subscribe(val => {});

我尝试将 combineLatest 运算符放在顶部，但是只有在两个可观察对象都发出至少一个值时才能解决。

valueChanges observable 不一定会改变，但 recordFormService.recordForm$ 将在页面加载时至少运行一次。

注意：valueChanges observable 的结果是一个字符串，recordForm$ observable 的结果是一个 FormGroup 对象，它们需要完全不同的处理方式。

我可以使用什么运算符来实现这个要求？

// Get latest value of all observables
combineLatest(
  this.parentFormGroup.controls['ProductOwner'].valueChanges,
  // Get current value of Record Form
  this.recordFormService.recordForm$
)
.pipe(
  switchMap(recordForm => {
    // Get current Product Owner on Record Form
    this.currentProductOwner = recordForm.value.ProductOwner;

    // Search People
    return this.recordsApiService.searchPeople$(this.currentProductOwner);
  }),
  catchError(err => {
    console.error(`get recordForm$ failed ${err}`);
    return throwError(err);
  })
)
.subscribe(results => {
  // Set Product Owners select box options
  this.productOwners = results.data;
});

更新：

我也试过merge。然而，两个 observable 的结果完全不同（一个是 formGroup，一个是字符串），它们都需要以不同的方式处理。

// Get latest value of all observables
merge(
  this.parentFormGroup.controls['ProductOwner'].valueChanges,
  // Get current value of Record Form
  this.recordFormService.recordForm$
)

Answer 1

你会想要merge:

import { merge } from 'rxjs';

从你想要响应的 observable 中创建 2 个可发出相同类型的 observable。我使用了一个假类型RecordForm，它保存了您的表单值和产品所有者作为示例：

const ownerChanges$: Observable<RecordForm> = this.parentFormGroup.controls['ProductOwner'].valueChanges.pipe(
  withLatestFrom(this.recordFormService.recordForm$),
  map(([owner, recordForm]) => ({ ...recordForm, owner }))
);
const formChanges$: Observable<RecordForm> = this.recordFormService.recordForm$.pipe(
  withLatestFrom(this.parentFormGroup.controls['ProductOwner'].valueChanges),
  map(([recordForm, owner]) => ({ ...recordForm, owner }))
);

然后将它们传递给合并：

// Get latest value of all observables
merge(ownerChanges$, recordChanges$).pipe(
  ...
  // the type passed here is always RecordForm - react accordingly
)
.subscribe(results => {
  // Set Product Owners select box options
  this.productOwners = results.data;
});

但是您需要确保传递给merge 的两个可观察对象是相同的类型，因此可以以相同的方式处理它们。

Answer 2

我认为您想要做的是创建另一个下游可观察对象；用于您的表单输入的那个应该触发 API 调用以及原始 observable。为此，您可以在原始 Observable 上使用 mergeMap 管道。这将在内部创建另一个 Emitter，它可以从第一个并行发射。试试这个：

const formObs$ = this.parentFormGroup.controls['ProductOwner'].valueChanges;

this.recordFormService.recordForm$.pipe(
  // give precedence in the merge callback to the original stream Observable,
  // but if just the second Observable fires here, grab that value and use it to pass downstream
  mergeMap(() => formObs$, (originalValue, formObsValue) => originalValue || formObsValue),
  // now recordForm is either from the original observable, or from your form input
  switchMap(recordForm => {
    this.currentProductOwner = recordForm.value.ProductOwner;

    // Search People
    return this.recordsApiService.searchPeople$(this.currentProductOwner);
  }),
  catchError(err => {
    console.error(`get recordForm$ failed ${err}`);
    return throwError(err);
  })
).subscribe(results => {
  // Set Product Owners select box options
  this.productOwners = results.data;
});

我在这里假设 recordForm$ 正在返回一个表单值，就像您的第二个可观察到的 .valueChanges 一样。如果不是这种情况，您只需要在 mergeMap 2nd 回调函数中对其进行按摩，即可为您想要的 switchMap 生成合并值。