使用RxJS如何触发另一个动作流的可观察对象

如何解决使用RxJS如何触发另一个动作流的可观察对象

因此，我试图通过angular和RxJS反应性地构建列表组件。 该列表具有很多功能，例如：

• 按词搜索
• 更改页码
• 更改页面大小
• 按字段过滤
• 排序

搜索服务代码：

private seachItemSubject = new BehaviorSubject<string>("");
private pageLimitSubject = new BehaviorSubject<number>(10);
private pageNumberSubject = new BehaviorSubject<number>(0);
private orderingSubject = new BehaviorSubject<ListOrder>({});
private filteringSubject = new BehaviorSubject<any>({});

searchItemAction$ = this.seachItemSubject.asObservable();
pageLimitAction$ = this.pageLimitSubject.asObservable();
pageNumberAction$ = this.pageNumberSubject.asObservable();
orderingAction$ = this.orderingSubject.asObservable();
filteringAction$ = this.filteringSubject.asObservable();

combination$ = Observable.combineLatest(
    this.searchItemAction$,this.pageLimitAction$,this.pageNumberAction$,this.orderingAction$,this.filteringAction$
)

changeSeachItem(searchItem: string) {
    this.seachItemSubject.next(searchItem);
}

changePageLimit(pageLimit: number) {
    this.pageLimitSubject.next(pageLimit);
}

changePageNumber(pageNumber: number) {
    this.pageNumberSubject.next(pageNumber);
}

changeOrdering(listOreder: ListOrder) {
    this.orderingSubject.next(listOreder);
}

changeFilters(filters: any) {
    this.filteringSubject.next(filters)
}

组件代码

results$: Observable<MyResult> = this.listSearchService.combination$
    .switchMap(([filterItem,pageLimit,pageNumber,ordering,filters]) => this.listService.query({
        offset: pageNumber,limit: pageLimit,filter: filterItem,order: ordering.order,descending: ordering.descending,...filters
    }));

我现在面临的问题是，如果用户使用过滤功能，我希望更改页码。

我想我想要的是某种使用者可以在用户使用过滤功能时调用的主题或事件。

所以会是这样。

applyFilters(filters: any) {
    this.listSearchService.changePageNumber(0);
    this.listSearchService.changeFilters(filters);
    this.listSearchService.commit.next() -----> so it would change both then start the combine latest
    console.log(filters)
}

我不确定这是否是正确的答案，所以请告诉我您是否有更好的答案。

谢谢

解决方法

您已经发现，您无法使用下面的原始解决方案，因为从combineLatest调用changePageNumber和changeFilters时，applyFilters都会触发。

另一种方法是使用通过scan运算符实现的单个状态对象，并一次更新其中的一个或多个属性。这意味着从该服务触发的服务在每个“操作”中只会被调用一次。

顺便说一句，您当前正在将“状态”存储在每个BehaviorSubject中，并将其组合起来。

class ListState {
  searchItem: any = "";
  pageLimit: any = 10;
  pageNumber: any = 0;
  ordering: any = "asc";
  filtering: any = "";
}

const listStateUpdatesSubject = new BehaviorSubject<Partial<ListState>>(null);
const listStateUpdates$ = listStateUpdatesSubject.asObservable();

// fires once per state update,which can consist of multiple properties
const currListState$ = listStateUpdates$.pipe(
  filter(update => !!update),scan<Partial<ListState>,ListState>(
    (acc,value) => ({ ...acc,...value }),new ListState)
);

const callListService$ = currListState$.pipe(map(state => listService(state)));

function changeOrdering(listOreder: string) {
    changeState({ ordering: listOreder });
}

function changeFilters(filters: any) {
    changeState({ pageNumber: 0,filtering: filters }); // set both page number and filtering
}

// etc

这是一个有效的演示：https://stackblitz.com/edit/so-rxjs-filtering-2?file=index.ts，请注意，对changeFilters的调用导致该服务仅被调用一次。

原始解决方案

创建一个新的可观察对象来监听filteringAction$，执行所需的操作并在combineLatest中使用该新的对象。

...
orderingAction$ = this.orderingSubject.asObservable();
filteringAction$ = this.filteringSubject.asObservable();

filterApplied$ = filteringAction$.pipe(map(filter => applyFilters(filter)));

applyFilters(filters: any) {
  this.listSearchService.changePageNumber(0);
  this.listSearchService.changeFilters(filters);
  return filters;
}

combination$ = Observable.combineLatest(
  this.searchItemAction$,this.pageLimitAction$,this.pageNumberAction$,this.orderingAction$,this.filterApplied$    // instead of filteringAction$
)