如何与 BehaviorSubject 共享 HttpClient GET？

如何解决如何与 BehaviorSubject 共享 HttpClient GET？

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

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

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

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

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 进行共享？

解决方法

对于此代码：

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,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 get 返回时获得一个值。这样做，请参阅：https://stackoverflow.com/a/67508863/787958

如何与 BehaviorSubject 共享 HttpClient GET？

如何解决如何与 BehaviorSubject 共享 HttpClient GET？

解决方法

相关推荐