Using expand for paging with recursive call

user_1248716 0
Viewed 162

I am getting data from a third party API that requires me to manage the paging manually by keeping track of the number of records retrieved and the total number of records.

I've tried to use the expand,reduce operators in RxJs as I've done something similar with other apis that provide me with a 'nextPage' link, but I am running into a infinite loop in this particular case.

I know the problem is that the nextPosition variable is not being updated when the output of expand is run through the expand again, but I am not sure if it is possible to solve this problem using this method.

Is it possible to use expand and reduce with a recursive function, and if so, what do I need to do to fix the below?

private async retreivePagedRecords<T>(companyURL: string, query: string, startPosition: number, totalCount: number, transformFn: (qbData: any) => T[]): Promise<T[]> {

        const headers = this.getHttpConfig();
        let pageQuery = `${query} STARTPOSITION ${startPosition} MAXRESULTS ${this.pagingSize}`;
        const nextPosition = startPosition + this.pagingSize;

        const records = await lastValueFrom(this.http.get(`${companyURL}/query?query=${pageQuery}`, headers)
            .pipe(
                map(x => x.data as any),
                map(x => {
                    //Trivial transformation to property names etc.
                    return transformFn(x);
                }),
                expand(x => nextPosition > totalCount ? [] : this.retreivePagedRecords<T>(companyURL, query, nextPosition, totalCount, transformFn)),
                reduce((acc: T[], x: T[]) => acc.concat(x ?? []), []),
                catchError(error => {
                    return of([]);
                })
            ));
        return records;
    }
rxjs typescript
3 Answers

for your EXPAND operator, you have to return EMPTY if condition to proceed if falsy

 import { EMPTY } from 'rxjs';

 expand(x => nextPosition > totalCount ? EMPTY : this.retreivePagedRecords<T>(companyURL, query, nextPosition, totalCount, transformFn)),

Check my post here about EXPAND operator https://dev.to/fatehmohamed14/recursive-http-calls-the-rxjs-way-d61

user_43992810

You're recursively calling your function forever. Instead, use expand to keep calling the API.

private pagingSize: number;

private async retreivePagedRecords<T>(
  companyURL: string, 
  query: string,
  totalCount: number, 
  transformFn: (qbData: any) => T[]
): Promise<T[]> {

  const headers = this.getHttpConfig();

  let queryString = startPos => `${companyURL}/query?query=${query} STARTPOSITION ${startPos} MAXRESULTS ${this.pagingSize}`;

  let queryStream = startPos => this.http.get(queryString(startPos), headers).pipe(
    map(payload => ({
      payload,
      currentPos: startPos
    })),
  );

  const queryResult$ = queryStream(0).pipe(
    
    expand(({currentPos}) => 
      currentPos > totalCount ? 
      EMPTY : 
      queryStream(currentPos + this.pagingSize)
    ),

    // Trivial transformation to property names etc.
    map(({payload}) => transformFn(payload.data)),
    // Reduce Ect
    reduce((acc: T[], x: T[]) => acc.concat(x ?? []), []),
    catchError(error => {
        return of([]);
    })
  );

  return await lastValueFrom(queryResult$);
}
user_135009860

Seem there should not be a reference to the same function inside of expand. "Expand" is already recursion.

You can store retrieved data inside "of()" and pass there correct "pageID"

In my case if "pageId" is equal to null => I need to finish recursive and return EMPTY

You can follow my example

public loadAllAlarms(params: AlarmRequest, isActiveAPI: boolean,  maxCount = null): Observable<AlarmDefinition[]> {
    console.log(params);
    return of({page: undefined, alarms: [], counts: 0}).pipe(
        expand(data => data.page === null
            ? EMPTY
            : this.AMapi.loadAlarms({...params, ...(!data.page ? {} : {pageState: data.page})}, isActiveAPI)
                .pipe(
                    map(response => {
                    const counts = data.counts + response.results.length;
                    return ({
                        page: (maxCount && (counts >= maxCount)) ? null : response.pageState,
                        alarms: response.results,
                        counts
                    })
                }))
        ),
        reduce((acc, items) => ([...acc, ...items.alarms]), []),
        map(data => maxCount ? data.slice(0, maxCount) : data)
    )
}

some unit tests

it('should call API until pageState === null is faced', () => {
  const countExistedPages = 3;
  let countCalls = 0;
  const spy = spyOn(AMapi, 'loadAlarms').and.callFake(() => {
    countCalls += 1;
    return of({pageState: countCalls === countExistedPages ? null : 'id', results: new Array(10)} as AlarmPagedResponse);
  });

  service.loadAllAlarms({} as any, true).subscribe(data => expect(data.length).toEqual(30));
  expect(spy).toHaveBeenCalledTimes(3);
});

it('should call API until maxCount is fulfill', () => {
  const countExistedPages = 3;
  let countCalls = 0;
  const spy = spyOn(AMapi, 'loadAlarms').and.callFake(() => {
    countCalls += 1;
    return of({pageState: countCalls === countExistedPages ? null : 'id', results: new Array(10)} as AlarmPagedResponse);
  });

  service.loadAllAlarms({} as any, true, 15).subscribe(data => expect(data.length).toEqual(15));
  expect(spy).toHaveBeenCalledTimes(2);
});
user_92009410
Related