Synchronously process an Observable dependent on another Observable using NGRX & Angular

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$;
  }

In the code above I parse an observable called patients$ that has an array of patients. I am then mapping those patients to an object array called this.rows that I show on my client.

My problem involves the hasAlert property which processes another observable itself in the method call to hasAlerts(). This method call to hasAlerts() does not happen synchronously so the console.log("this happens first"); occurs before my hasAlerts method can do the logic in the If statement to decide whether it should be set to true or false, instead it just uses the value it is initialized to in the first line of hasAlerts(). Confirmed by the console.log("this happens second"); being displayed second.

hasAlerts() could return a boolean not an observable, I was trying to see if that would solve my problem using an asycn pipe on the front end (it didnt).

I believe the way to solve this involves using mergemap however I am not sure how i would pass in the pat.id that the hasAlerts method needs? Or maybe this is not the correct approach to solve my current problem with the asynchronous execution of this code.

I am currently trying to use this this question about mergemap to solve my issue but passing the pat.id that the second observable in hasAlerts I haven't figured out yet. 1

updated code following Piccis idea.

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 returns an array of patients. I need to comment out the middle because there is a syntax error with the switchmap line saying " Argument of type 'void' is not assignable to parameter of type 'ArrayLike<{}>'"

Answer 1

You are looking in the right direction with mergeMap, but you need to rework a bit your code.

You start with this.patients$ which is an Observable.

From here you need to create a stream of Observable. This can be done with the static method from of Observable.

Now you are able to manager each single Patient and fetch his/her curObservation via the service fetchItems.

Eventually you recreate your array and then subscribe.

The final result could look like something along these lines

    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: false, //set the default value
                  settings: "Settings"
                };
              });
            })
            .do(data => consoel.log(data))  // check  whether data is an array of Patients as it is supposed to be
            .switchMap(p => Observable.from(p)) // creates a stream of Observable<Patient>
            .do(data => console.log(data))  // check whether data is a single Patient
            .mergeMap(patient => this.dataService.fetchItems<Observation>(
                "Observation",
                null,
                patient.id// patient should be your guy          
              )
              .map(curObservation => {
                if (curObservation.category.coding[0].code == "GlucoseEvent"){ 
                  shouldAlert$ = true;
                  console.log("should alert$", shouldAlert$);
                }
              })
            )
           .toArray()
           .subscribe(
              patients => {
                  this.table.recalculatePages();
                  console.log(this.rows);
              }
           )
        }

UPDATE - THE BASIC MECHANISM

If we remove all specifics of your case, the basic mechanism which is implemented in the snippet above is the following

import {Observable} from 'rxjs';

const obs1 = Observable.of([1, 2, 3, 4, 5, 6]);

obs1
.switchMap(n => Observable.from(n))
.mergeMap(n => Observable.of(n*2))
.toArray()
.subscribe(console.log)

Answer 2

The basic problem is combining two async calls, can be done with zip().

(Note, I originally posted a solution with forkJoin(), but this does not work with ngrx select(), since the select never completes - therefore forkJoin never fires).

Convert the Observable<Patient[]> returned from the first fetch to Observable<Patient>, as it is more convenient for the zip operator.

Next problem is 2nd async is dependent on result of 1st async (pat.id) - build that one with concatMap().

(Note, I originally suggested mergeMap(), however concatMap() guarantees the same ordering of hasAlert$ as in patient$. This is important, since this.dataService.fetchItems() may return individual fetches out of order).

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();
    }); 
}

---

Testing the Rx logic of the answer snippet.

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>

Answer 3

This was how I was able to get this logic to work for me, via a different approach. I need to go through the ultimate angular ngrx video and a few rxjs tutorials to better understand where I was going wrong in the other two answers because the examples provided worked well.

The approach that worked for me below

• use filter when processing the observations from the data service in the hasAlerts method and add any observations that meet the criteria to that and return it.

• set the hasAlerts property to false, and then call the hasAlerts() method for that given patient and modify that property on the row and then return the row.

  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
      var rowx= {
        username: pat.username,
        id: pat.id,
        hasAlert: false, //made this an observable for the async pipe in view
        settings: "Settings"
      };
  
      this.hasAlerts(pat).do(x => {
          observations++;
          if (observations>0)
          {
            rowX.hasAlert=true;
          }
        }).subscribe();
      return rowX;
    });
  }).subscribe(
  ()=>{
  
  },
  ()=>{
    this.table.recalculatePages();
  });
  
  }
   hasAlerts(pat: Patient): Observable<Observation> {
  
    var obs$ = this.dataService.fetchItems<Observation>(
      "Observation",
      null,
      pat.id
    ).filer(function(curObservation){
       if (curObservation.category.coding[0].code == "GlucoseEvent"){ 
         return true;
       }
       else{
         return false;
       }
    });
  
  return obs$;
  }