How to share HttpClient GET with BehaviorSubject?

Question

So, I was utilizing shareReplay(1) to cache the array of Items in memory that are returned from an HttpClient GET call.

if (!this.getItems$) {
    this.getItems$ = this.httpClient
            .get<Item[]>(url, options)
            .pipe(shareReplay(1));
}
return this.getItems$;

I'm doing this because I have many components on the page that need the array of Items and I don't want to make the http call for each one.

I am also creating an Item on the page. So, after I make a call to the service to call the API which creates the Item in the database and returns it, I would like to add it to the array of Items in memory and alert all the components that are subscribing to the updated array. So, I tried this:

private items = new BehaviorSubject<Item[]>(null);

...

if (!this.getItems$) {
this.getItems$ = this.httpClient
    .get<Item[]>(url, options)
    .pipe(
      tap({
        next: (items) => {
          this.items.next(items);
        },
      })
    )
    .pipe(multicast(() => this.items));
}
return this.getItems$;

Then in the method that calls the API:

return this.httpClient.post<Item>(url, item, options).pipe(
      tap({
        next: (item) => {
          this.items.next(this.items.value.push(item));
        },
      })
    );

The issue is, that anything that is subscribing to the getItems method, is always returning null. There are items in the database, so even on the first call, there should be items returned. There are, as I tested it with the shareReplay(1) and it works.

How can I share utilizing a BehaviorSubject instead of a ReplaySubject?

Answer 1

For this code:

if (!this.getItems$) {                 
    this.getItems$ = this.httpClient 
            .get<Item[]>(url, options)
            .pipe(shareReplay(1));
}
return this.getItems$;

This: this.httpClient.get() is asynchronous. So if getItems$ is not yet set, it won't yet be set by the return statement.

The code is executed in order as follows:

• Line 1.
• Line 2 & 3
• Line 5. (returning null)
• Line 4. (actually processing the returned response)

Also, adding the if is redundant with the shareReplay that will automatically get the data only if the Observable is not yet set.

Try something like this:

getItems$ = this.httpClient
            .get<Item[]>(url, options)
            .pipe(shareReplay(1));

EDIT: Where the above code is a property, not within a method.

For the second part of the question regarding creating new items, you could do something like this, which is from one of my examples. You can rename the properties/interfaces as needed for your scenario.

  // Action Stream
  private productInsertedSubject = new Subject<Product>();
  productInsertedAction$ = this.productInsertedSubject.asObservable();

  // Merge the streams
  productsWithAdd$ = merge(
    this.productsWithCategory$,  // would be your getItems$
    this.productInsertedAction$
  )
    .pipe(
      scan((acc: Product[], value: Product) => [...acc, value]),
      catchError(err => {
        console.error(err);
        return throwError(err);
      })
    );

You can find a complete example with the above code here: https://github.com/DeborahK/Angular-RxJS/tree/master/APM-Final

EDIT:

For reference, here is the key code from the stackblitz:

Service

import { HttpClient } from '@angular/common/http';
import { Injectable } from '@angular/core';
import { merge, Observable, Subject, throwError } from 'rxjs';
import { catchError, scan, shareReplay, tap } from 'rxjs/operators';
import { Item } from './item.model';

@Injectable({
  providedIn: 'root'
})
export class ItemsService {
  url = 'https://jsonplaceholder.typicode.com/users';

  // This is a property
  // It executes the http request when subscribed to the first time
  // And emits the returned array of items.
  // All other times, it replays (and emits) the items due to the shareReplay.
  getItems$ = this.httpClient.get<Item[]>(this.url).pipe(
    tap(x => console.log("I'm only getting the data once", JSON.stringify(x))),
    tap(x => {
      // You can write any other code here, inside the tap
      // to perform any other operations
    }),
    shareReplay(1)
  );
  
    // Action Stream
  private itemInsertedSubject = new Subject<Item>();
  productInsertedAction$ = this.itemInsertedSubject.asObservable();

  // Merge the action stream that emits every time an item is added
  // with the data stream
  allItems$ = merge(
    this.getItems$,
    this.productInsertedAction$
  )
    .pipe(
      scan((acc: Item[], value: Item) => [...acc, value]),
      catchError(err => {
        console.error(err);
        return throwError(err);
      })
    );

  constructor(private httpClient: HttpClient) {}

  addItem(item) {
    this.itemInsertedSubject.next(item);
  }
}

Component

import { Component, OnInit } from '@angular/core';
import { Item } from './item.model';
import { ItemsService } from './items.service';

@Component({
  selector: 'app-item',
  templateUrl: './item.component.html'
})
export class ItemComponent {

  // Recommended technique using the async pipe in the template
  // With this technique, no ngOnInit is required.
  items$ = this.itemService.allItems$;

  constructor(private itemService: ItemsService) {}

  addItem() {
    this.itemService.addItem({id: 42, name: 'Deborah Kurata'})
  }
}

Template

<h1>My Component</h1>
<button (click)="addItem()">Add Item</button>
<div *ngFor="let item of items$ | async">
  <div>{{item.name}}</div>
</div>