angular - RxJS - Dela oändlig ström som produceras av 'expand'

Jag har en paginerad tredjepartsresurs som finns i en webbtjänst. Vad jag vill göra är att omvandla den paginerade resursen till ett flöde av värden, och låta klienten bestämma hur många element som ska användas. Det vill säga, klienten bör inte veta att den ursprungliga resursen är paginerad.

Hittills har jag följande kod:

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();

Detta fungerar verkligen bra: jag får en lat oändlig ström av siffror, och klienten kan bara göra infinite$.pipe(take(n)) och ta de första n elementen, utan att veta att den underliggande resursen är paginerad.

Nu vill jag dela dessa värden när jag hanterar flera prenumeranter, det vill säga:

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);

Som du kan se kommer vi att ha flera prenumeranter på den oändliga strömmen, och jag vill återanvända de värden som redan har emitterats för att minska antalet fetchPage-anrop. I det här exemplet, när en klient efterfrågar 10 objekt (take(10)), då bör varje klient som begär färre än 10 objekt (t.ex. 5 objekt) resultera i inga anrop till fetchPage eftersom dessa objekt redan har emitterats.

Jag försökte följande, men jag kunde inte få önskat beteende:

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

Fungerar inte eftersom sena prenumeranter resulterar i flera anrop till 'fetchPage'.

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

Tvingar alla värden i strömmen även om de inte behövs (ingen klient har ännu efterfrågat alla objekt)

Alla tips skulle uppskattas. Om någon vill prova koden: https://stackblitz.com/edit/n4ywfw

Svar #1

Det verkar som att vi behöver hitta ett sätt att lagra något tillstånd, t.ex. det senaste lästa sidnumret och de hämtade objekten, och se till att detta tillstånd spelas upp när nya prenumerationer anländer.

En möjlighet skulle kunna vara att arbeta med stängningar på detta sätt.

Först skapar du en infiniteStreamFactoryGenerator-funktion som returnerar en funktion som, via closures, håller ett visst tillstånd, specifikt lastPage och itemsRead. Ett sådant tillstånd uppdateras inom den funktion som returneras av infiniteStreamFactoryGenerator-funktionen.

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))
      )
    );
  };
}

Därefter anropar vi infiniteStreamFactoryGenerator för att skapa den verkliga fabriksfunktionen och använder denna fabriksfunktion för att instansiera de olika strömmarna, så här

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);

Som du ser har jag också lagt till en fjärde prenumerant som kräver att ytterligare sidor laddas.

Enligt denna stackblitz verkar denna lösning fungera.

Ärligt talat antyder idén att behöva bygga en fabriksgeneratorfunktion för mig att det kan finnas ett enklare sätt att lösa detta intressanta problem, men jag har inte hittat det än.

Svar #2

Eftersom du vill ha cachningen av HTTP-anropet, måste du flytta shareReplay inuti din fetchPage-metod för att få det att fungera

return fetchPage().pipe(
 expand((res) => fetchPage(res.page + 1)),
 mergeMap((res) => from(res.items)),
 shareReplay()
);

https://stackblitz.com/edit/n4ywfw-xkajmp?file=index.ts

Här är ett annat exempel som jag har uppdaterat för dig, du kan märka att samtalet bara gjordes en gång

https://stackblitz.com/edit/angular-ivy-o9pjw6?file=src/app/app.component.ts