return mix of sync and async values from single Observable

user_905865 0
Viewed 94

I have a cache for already fetched objects. My function getListByIds should return Observable<Model>. Inside this function I need to check if object exists in cache (check by id) and if not then fetch it from backend. Code using async/await could be:

public async getListByIds(ids: number[]): Promise<Model> {
  return ids.map(id => {
    let cached = this.cacheService.get(id); // may return undefined
    if (!cached) cached = await this.http.get(baseUrl + '/get/' + id).toPromise();
    return cached;
  });
}

I need to convert it to RxJS - to Promise() is deprecated and I want to learn RxJs. I'd like not to use subscribe() inside.

Edit

Let's assume I want to store ready models in cache and from API there are pure JOSN data which need to be parsed.

public async getListByIds(ids: number[]): Promise<Model> {
  return ids.map(id => {
    let cached = this.cacheService.get(id); // may return undefined
    if (!cached) {
      const apiData = await this.http.get(baseUrl + '/get/' + id).toPromise();
      cached = this.parseApiData(apiData);
    }
    return cached;
  });
}

Where should I call parseApiData() ?

rxjs angular typescript
2 Answers

The function would look something like this:

import { forkJoin } from "rxjs/observable/forkJoin";
import { of } from "rxjs/observable/of";

public getListByIds(ids: number[]): Observable<Model> {
  return forkJoin(ids.map(
    (id) => {
      let cached = this.cacheService.get(id)
      if (!cached) return this.http.get(baseUrl + '/get/' + id).pipe(
        map(apiData => this.parseApiData(apiData))
      )
      return of(cached)
    }
  )
}

The forkJoin waits for all mapped Observables to complete.

user_86136300

Here are two different approaches depending on whether you'd like your observable to emit an array of a stream.

// A Stream of Models
public getListByIds(ids: number[]): Observable<Model> {
  return from(ids).pipe(
    map(id =>  [id, this.cacheService.get(id)]),
    mergeMap(([id, cached]) => !!cached ? 
      of(cached) : 
      this.http.get(baseUrl + '/get/' + id)
    )
  );
}

// An Array of Models
public getListByIds(ids: number[]): Observable<Model[]> {
  return forkJoin(ids
    .map(id =>  [id, this.cacheService.get(id)])
    .map(([id, cached]) => !!cached ? 
      of(cached) : 
      this.http.get(baseUrl + '/get/' + id)
    )
  );
}

For the sake of completeness, you can transform the first into the second with toArray.

// An Array of Models
public getListByIds(ids: number[]): Observable<Model[]> {
  return from(ids).pipe(
    map(id =>  [id, this.cacheService.get(id)]),
    mergeMap(([id, cached]) => !!cached ? 
      of(cached) : 
      this.http.get(baseUrl + '/get/' + id)
    ),
    toArray()
  );
}
user_135009860
Related