Angular/RXJS - 如何合并基于不同操作多次触发的相同 HTTP 端点调用的响应？

Question

我实现了无限滚动，作为其中的一部分，我触发了一个 API 端点，以根据元素位置（使用的 intersectionObserver）批量获取结果。所以在这种情况下，我会以不同的时间间隔多次访问相同的 API。有时快速滚动时，我看到调用触发得非常快，并且查看未按预期呈现。在 RXJS/Angular 中有没有办法可以合并这些响应或等待第一次调用完成才能触发第二次调用？或任何想法/建议我该如何处理？

        export class {
          @Output() eventscroll = new EventEmitter();
          @ViewChild('iscroll')
            iscroll: ElementRef<HTMLElement>;

          @Component({
            selector: 'myinfinitescroll',
            template: `
    <div #iscroll></div>
              <ng-content></ng-content>

            `,

          });
          ngAfterViewInit() {
           this.zone.runOutsideAngular(() => {
             this.iob.observe(this.iscroll.nativeElement);
          })
          }

          this.iob = new IntersectionObserver(this.handleScroll(),
                    { root: null, rootMargin: '0px',threshold: 0 });

          handleScroll([entry]){
            if (entry.isIntersecting && entry.intersectionRatio >= 0) {
                this.eventscroll.emit();   ///api call happens here, currently it is just a simple HTTP post call  
              }
          }

****Edit1****


myservicemethod(a, b, c): void {
    const reqdata = {
        p1 : a
        p2 : b
        p3 : c
    };
    const debounceWindow = 1000;
    this._callAPI$.pipe(
      concatMap((data): Observable<any> => this.myAPI.getData(reqdata)),
      bufferTime(debounceWindow),
      map((resp: Array<Array<any>>) => resp.reduce((acc, item) => acc = [...acc, ...item], []))
    ).subscribe((resp)=>{
      if (resp) {
        this.updateRecords();
      }
  })
  }

我在组件构造函数上调用 this.myservice.myservicemethod，在组件 HTML 上调用 myservice._callAPI$.next($event)

Answer 1

好吧，我认为解决方案与您发布的代码无关，因为您应该以任何方式发射（您不能对发射进行去抖动，否则某些数据将丢失，并且滚动会出现一些奇怪的丢失孔数据）。

所以，有人会像这样使用你的组件：

<myinfinitescroll (eventscroll)="_callAPI$.next($event)">
  <my-panel [data]="_data$ | async"></my-panel>
</myinfinitescroll>

因此，假设您的数据以块的形式出现，并且每个块包含一个数据数组，并且您希望使用您的服务方法在将响应发送到单个数组之前缓冲一段时间：

grabSomeData(params: any): Observable<any[]>;

因此，在使用无限滚动的组件打字稿中，您必须做一些事情：

1. 按顺序进行调用并从 API 接收响应
2. 缓冲一段时间，比如说 1 秒。
3. 将所有接收到的数据数组重新发送为单个数组

Stackblitz demo

import {BehaviorSubject, Observable, Subject} from 'rxjs';
import {bufferTime, concatAll, concatMap} from 'rxjs/operators';

_data$ = new BehaviorSubject<any[]>([]);

_callAPI$ = new Subject<any>();

constructor(private myService: MyService) {
  this.scrollObservablesInit();
}

scrollObservablesInit() {
  // time window to buffer arriving data before reemits it
  const debounceWindow = 1000;

  this._callAPI$.pipe(
    // make the calls and receive the answers to the API
    concatMap((params) => this.myService.grabSomeData(params)),
    // buffer the arriving data for 1 second before reemit everything
    // in the buffer
    bufferTime(debounceWindow),
    // from above, now you get an array of arrays emitted 
    // by bufferWindow and we need to flatten it
    // obs.: there's an Array.prototype.flat operator
    //   on the recent versions on the ecma script, but
    //   let's not count on it here 
    map((data: Array<Array<any>>) => data
        .reduce((acc, item) => acc = [...acc, ...item], []))
  ).subscribe(this._data$)
}