angular - 使用 NGRX 和 Angular 同步处理依赖于另一个 Observable 的 Observable

Question

 ngOnInit(): void {
    this.store.dispatch(new patients.Load([]));
    this.patients$ = this.store.select(fromPatients.getAll);

    this.patients$.map(p =>{ //  patients$: Observable<Patient[]>;
      this.rows = p.map(pat => { //I use this on the front end
        return {
          username: pat.username,
          id: pat.id,
          hasAlert: this.hasAlerts(pat), //made this an observable for the async pipe in view
          settings: "Settings"
        };
      });
      this.table.recalculatePages();
      console.log(this.rows);
      console.log("This happens first");
    }).subscribe();

  }
  hasAlerts(pat: Patient): Observable<boolean> {
    var shouldAlert$ = Observable.of(false);//this value is always taken

      this.observations2$ = this.dataService.fetchItems<Observation>(
        "Observation",
        null,
        pat.id// How would i pass this in with mergeMap()?  
      );

      this.observations2$.subscribe(curObservation => {
        if (curObservation.category.coding[0].code == "GlucoseEvent"){ 
          shouldAlert$ = Observable.of(true);
          console.log("should alert$", shouldAlert$);
        }
      });

    console.log("this happens second");
    return shouldAlert$;
  }

在上面的代码中，我解析了一个名为患者$的可观察量，它有一个患者数组。然后，我将这些患者映射到我在客户端上显示的名为 this.rows 的对象数组。

我的问题涉及 hasAlert 属性，该属性在对 hasAlerts() 的方法调用中处理另一个可观察值本身。对 hasAlerts() 的此方法调用不会同步发生，因此 console.log(“这首先发生”);发生在我的 hasAlerts 方法可以执行 If 语句中的逻辑来决定是否应将其设置为 true 或 false 之前，它只使用在 hasAlerts() 第一行中初始化的值。由 console.log 确认(“这发生在第二个”)；第二个显示。

hasAlerts() 可以返回一个 bool 值而不是一个可观察值，我试图看看这是否可以在前端使用 asycn 管道解决我的问题(它没有)。

我相信解决这个问题的方法涉及使用合并映射，但是我不确定如何传递 hasAlerts 方法所需的 pat.id ？或者，这可能不是解决我当前异步执行此代码问题的正确方法。

我目前正在尝试使用这个this question about mergemap解决我的问题，但传递了我还没有弄清楚的 hasAlerts 中第二个可观察的 pat.id 。 1

按照 Piccis 的想法更新了代码。

this.patients$.map(p =>{ //  patients$: Observable<Patient[]>;
      this.rows = p.map(pat => { //I use this on the front end
        return {
          username: pat.username,
          id: pat.id,
          hasAlert: false, //set the default value
          settings: "Settings"
        };
      })
    }).do(data => console.log("data1",data))
  //   .switchMap(p => Observable.from(p))
  //   .do(data => console.log("data2",data)) // creates a stream of Observable<Patient>
  //   .mergeMap(patient => this.dataService.fetchItems<Observation>(
  //       "Observation",
  //       null,
  //       "pat..frank"//patient[0].id//"pat..frank"//patient.id// patient should be your guy          
  //     )
  //     .map(curObservation => {
  //       console.log("currOBS",curObservation); 

  //       if (curObservation.category.coding[0].code == "GlucoseEvent"){
  //         var shouldAlert$ = true;
  //         console.log("should alert$", shouldAlert$);
  //       }
  //     })
  //   ).do(data => console.log(data))
  //  .toArray()
   .subscribe(
      patients => {
          this.table.recalculatePages();
          console.log(this.rows);
      }
   )

Data1 返回患者数组。我需要注释掉中间，因为 switchmap 行存在语法错误，显示“‘void’类型的参数不可分配给‘ArrayLike<{}>’类型的参数”

Answer 1

基本问题是组合两个异步调用，可以通过 zip() 来完成。

(注意，我最初发布了一个使用 forkJoin() 的解决方案，但这不适用于 ngrx select()，因为 select 永远不会完成 - 因此 forkJoin 永远不会触发)。

转换Observable<Patient[]>从第一次获取返回到 Observable<Patient> ，因为这对于 zip 运算符(operator)来说更方便。

下一个问题是第二个异步依赖于第一个异步的结果 (pat.id) - 使用 concatMap() 构建该异步。 。

(注意，我最初建议 mergeMap() ，但是 concatMap() 保证 hasAlert$ 的顺序与病人$中的顺序相同。这很重要，因为 this.dataService.fetchItems() 可能会返回来自命令)。

import { zip } from 'rxjs/observable/zip';
...

ngOnInit(): void {
  this.store.dispatch(new patients.Load([]));

  const patient$ = this.store.select(fromPatients.getAll)
    .mergeMap(value => value); // convert from Observable<patients[]> to Observable<patient>

  const hasAlert$ = patient$.concatMap(patient => {
    return this.dataService.fetchItems<Observation>('Observation' null, patient.id)
      .map(curObservation => curObservation.category.coding[0].code === 'GlucoseEvent')
    );
  })

  zip(patient$, hasAlert$)  // combine values from both asyncs
    .map(([patient, hasAlert]) => {
      return {
        username: patient.username,
        id: patient.id,
        hasAlert,
        settings: "Settings"
      };
    })
    .toArray()
    .subscribe(rows => {
      this.rows = rows;
      this.table.recalculatePages();
    }); 
}

---

测试答案片段的接收逻辑。

console.clear();
const { zip, from, of } = Rx.Observable;
/* in Angular with Rxjs v5.5, use 
  import { zip } from 'rxjs/observable/zip';
  import { from } from 'rxjs/observable/of';
  import { of } from 'rxjs/observable/from';
*/

// Simulate other observables
const storeSelectFromPatientsGetAll$ = () =>
  of([{id: 1, username: 'Fred'}, {id: 2, username: 'Joan'}]);
const dataServiceFetchItems$ = (type, something, id) =>
  of({ category: { coding: [{code: 'GlucoseEvent'}] }})

// Testing the ngOnInit code
const patient$ = storeSelectFromPatientsGetAll$()
  .mergeMap(value => value);

const hasAlert$ = patient$.concatMap(patient => {
  return dataServiceFetchItems$('Observation', null, patient.id)
    .map(curObservation => curObservation.category.coding[0].code === 'GlucoseEvent');
});

zip(patient$, hasAlert$)  // wait for resolution of both asyncs
  .map(([patient, hasAlert]) => {
    return {
      username: patient.username,
      id: patient.id,
      hasAlert,
      settings: 'Settings'
    };
  })
  .toArray()
  .subscribe(rows => {
    console.log(rows);
  });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.8/Rx.js"></script>