如何与 BehaviorSubject 共享 HttpClient GET？

Question

因此，我使用 shareReplay(1) 缓存从 HttpClient GET 调用返回的内存中的项目数组。

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

我这样做是因为页面上有很多组件需要项目数组，我不想为每个组件都调用 http。

我也在页面上创建项目。因此，在我调用服务以调用在数据库中创建项目的 API 和 returns 之后，我想将它添加到内存中的项目数组中并提醒所有订阅更新数组的组件。所以，我尝试了这个：

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

然后在调用API的方法中：

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

问题是，任何订阅 getItems 方法的东西总是返回 null。数据库中有项目，所以即使在第一次调用时，也应该返回项目。有，因为我用 shareReplay(1) 测试过它并且它有效。

如何使用 BehaviorSubject 而不是 ReplaySubject 进行分享？

Answer 1

对于此代码：

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 行。（return空）
第 4 行。（实际处理 returned 响应）

此外，添加 if 与 shareReplay 是多余的，只有当 Observable 尚未设置时才会自动获取数据。

尝试这样的事情：

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

编辑：上面的代码是属性，不在方法内。

关于创建新项目的问题的第二部分，你可以这样做，这来自我的一个例子。您可以根据场景的需要重命名 properties/interfaces。

  // 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>

Answer 2

为了回答我的问题，我找到了解决方案。

首先，我必须获得最新版本的 rxjs：

npm install rxjs@latest

然后，我创建了一个 shareBehavior class:

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 获取 returns 时获取值。这样做，请参阅：

如何与 BehaviorSubject 共享 HttpClient GET？

How to share HttpClient GET with BehaviorSubject?

rxjs

behaviorsubject

angular

angular-httpclient

rxjs-pipeable-operators