Как поделиться HttpClient GET с BehaviorSubject?

Для этого кода:

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

Это: this.httpClient.get() асинхронно. Таким образом, если getItems$ еще не установлен, он еще не будет установлен оператором return.

Код выполняется в следующем порядке:

• Строка 1.
• Строки 2 и 3
• Строка 5. (возвращается null)
• Строка 4. (фактически обработка возвращенного ответа)

Кроме того, добавление if избыточно с shareReplay, которое автоматически получит данные, только если Observable еще не установлен.

Попробуйте что-то вроде этого:

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

РЕДАКТИРОВАТЬ: если приведенный выше код является свойством, а не методом.

Что касается второй части вопроса, касающейся создания новых элементов, вы можете сделать что-то вроде этого, что взято из одного из моих примеров. Вы можете переименовать свойства / интерфейсы в соответствии с вашим сценарием.

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

Вы можете найти полный пример с приведенным выше кодом здесь: https://github.com/DeborahK/Angular-RxJS/tree/master/APM-Final

РЕДАКТИРОВАТЬ:

Для справки, вот ключевой код из stackblitz:

Сервис

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

Компонент

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'})
  }
}

Шаблон

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

Чтобы ответить на свой вопрос, я смог найти решение.

Сначала мне нужно было получить последнюю версию rxjs:

npm install rxjs@latest

Затем я создал класс shareBehavior:

export function shareBehavior<T>(
  behaviorSubject: BehaviorSubject<T>
): MonoTypeOperatorFunction<T> {
  return share<T>({
    connector: () => behaviorSubject,
    resetOnError: true,
    resetOnComplete: false,
    resetOnRefCountZero: false,
  });
}

Затем в сервисе:

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

...

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

Это работает очень хорошо. Но проблема заключалась в том, что, поскольку поведение имеет значение по умолчанию, всем подписчикам отправляется начальное значение []. Они обновляются, как только завершается вызов API. Но я бы предпочел использовать ReplaySubject, чтобы подписчики получали значение только после возврата http. Сделайте это, см. https://stackoverflow.com/a/67508863/787958