RxJS - Share infinite stream produced by 'expand'

user_11035317 0
Viewed 92

I have a paginated third-party resource living in a web service. What I want to do is turn that paginated resource into a stream of values, and let the client decide how many elements to use. That is, the client should not know that the original resource is paginated.

So far I got the following code:

import { from, Observable, of } from 'rxjs';
import { expand, mergeMap } from 'rxjs/operators';

interface Result {
  page: number;
  items: number[];
}

// Assume that this does a real HTTP request
function fetchPage(page: number = 0): Observable<Result> {
  let items = [...Array(3).keys()].map((e) => e + 3 * page);
  console.log('Fetching page ' + page);
  return of({ page, items });
}

// Turn a paginated request into an infinite stream of 'number'
function mkInfiniteStream(): Observable<number> {
  return fetchPage().pipe(
    expand((res) => fetchPage(res.page + 1)),
    mergeMap((res) => from(res.items))
  );
}

const infinite$ = mkInfiniteStream();

This works really well: I get a lazy infinite stream of numbers, and the client can just do infinite$.pipe(take(n)) and take the first n elements, without knowing that the underlying resource is paginated.

Now, what I want to do is to share those values when dealing with multiple subscribers, that is:

infinite$.pipe(take(10)).subscribe((v) => console.log('[1] got ', v));

// Assume that later we have new subscribers
setTimeout(() => {
  infinte$.pipe(take(5)).subscribe((v) => console.log('[2] got ', v));
}, 1000);

setTimeout(() => {
  infinte$.pipe(take(4)).subscribe((v) => console.log('[3] got ', v));
}, 1500);

As you can see, we'll have multiple subscribers to the infinite stream, and I want to reuse the values already emitted in order to reduce the number of fetchPage calls. In this example, once a client asked for 10 items (take(10)), then any client that requests less than 10 items (ex. 5 items) should result in no calls to fetchPage since those items were already emited.

I tried the following, but I could not get the desired behavior:

const infinite$ = mkInfiniteStream().pipe(share())

Does not work since late subscribers result in multiple calls to 'fetchPage'.

const infinite$ = mkInfiniteStream().pipe(shareReplay())

Forces all values in the stream even though they are not needed (no client asked for all items yet)

Any hint would be appreciated. In case anyone wants to try the code: https://stackblitz.com/edit/n4ywfw

observable rxjs angular typescript
3 Answers

It looks like we need to find a way to store some state, e.g. the last page number read and the items fetched, and make sure this state is replayed when new subscriptions arrive.

A possibility could be to work with closures this way.

First you create an infiniteStreamFactoryGenerator function which returns a function which, via closures, holds some state, specifically the lastPage and the itemsRead. Such state is updated within the function returned by the infiniteStreamFactoryGenerator function.

function infiniteStreamFactoryGenerator() {
  let lastPage = 0;
  let itemsRead = [];
  return () => {
    // if  there are items read we return them first and then we start
    // the infinite stream
    return concat(
      from(itemsRead),
      fetchPage(lastPage).pipe(
        expand((res) => {
          return res.page < 10 ? fetchPage(res.page + 1) : EMPTY;
        }),
        tap((res) => {
          // here we update the state
          lastPage++;
          itemsRead = [...itemsRead, ...res.items];
        }),
        mergeMap((res) => from(res.items))
      )
    );
  };
}

Then we invoke the infiniteStreamFactoryGenerator to create the real factory function and use such factory function to instantiate the various streams, like this

const infiniteStreamFactory$ = infiniteStreamFactoryGenerator();

infiniteStreamFactory$()
  .pipe(take(10))
  .subscribe((v) => console.log("[1] got ", v));

// Assume that later we have new subscribers
setTimeout(() => {
  infiniteStreamFactory$()
    .pipe(take(5))
    .subscribe((v) => console.log("[2] got ", v));
}, 1000);

setTimeout(() => {
  infiniteStreamFactory$()
    .pipe(take(4))
    .subscribe((v) => console.log("[3] got ", v));
}, 1500);

setTimeout(() => {
  infiniteStreamFactory$()
    .pipe(take(20))
    .subscribe((v) => console.log("[4] got ", v));
}, 2000);

As you see I have added also a fourth subscriber which requires to load additional pages.

According to this stackblitz this solution seems to work.

To be honest, the idea of having to build a factory generator function suggests me there may be a simpler way to solve this interesting problem, but I have not found it yet.

user_56999930
user_43992810

The solution I ended up with involved creating an stateful Observable using closures. We kind of make our own version of share which it's bolted in to the Observable.

The only change is to mkInfiniteStream from the original code:

function mkInfiniteStream(): Observable<number> {
  let lastPage = 0;
  let itemsRead = [];
  return defer(() =>
    from(itemsRead).pipe(
      concatWith(
        fetchPage(lastPage).pipe(
          expand((res) => fetchPage(res.page + 1)),
          tap((res) => {
            lastPage++;
            itemsRead = [...itemsRead, ...res.items];
          }),
          mergeMap((res) => from(res.items))
        )
      )
    )
  );
}

// Usage is still the same:
const infinite$ = mkInfiniteStream();

The idea is to keep track internally of all items and the last page fetched. When requesting an item that was not fetched yet, the Observable is extended with items from the next page. This in turn updates all the items available (itemsRead = itemsRead = [...itemsRead, ...res.items]), so new subscribers will use items already fetched stored in itemsRead before forcing the expansion of the Observable.

Using share does nothing on this Observable, since data is already shared by the implementation. Using shareReplay forces the infinite stream, so it's not advised!.

I wish there was a more "pure" implementation, so if you come up with one please share it!

Thanks to @Picci for the initial idea.

user_110353170
Related