rxjs - Lage en uendelig rullende liste i typescript med rxjs5

Jeg prøver å bygge en uendelig rullende liste ved hjelp av TypeScript og rxjs. Det vil si at jeg ønsker at applikasjonen skal hente noen sider med resultater fra backend, og deretter hente flere resultater hver gang brukeren ruller nær bunnen.

Jeg har en Observable, bygget med Observable.prototype.expand(), som vil gi meg alle resultater, til slutt henter alle sider fra serveren. På grunn av Observable sin natur kan jeg imidlertid ikke pause denne prosessen. Når jeg abonnerer, vil den uunngåelig hente alle resultater så raskt som mulig. Jeg trenger en annen løsning, hvor jeg kan hente fra strømmen av resultater i den hastigheten jeg trenger.

Ting blir mer komplisert av det faktum at jeg ikke kan hente den andre siden fra API-et direkte, hver side inneholder informasjonen jeg trenger for å hente den neste. Et svar ser slik ut:

interface GraphApiResponse {
    data?: any[];
    paging?: {
        cursors: {
            before: string,
            after: string
        },
        next?: string,
        previous?: string
    };
}

Tilstedeværelsen av paging.next indikerer at det finnes en annen side, og paging.cursors.after brukes til å faktisk hente den.

Jeg ser ut til å ikke klare å finne ut hvordan jeg skal implementere dette, uten å gjøre det rotete. Imidlertid virker en uendelig liste som et så vanlig problem at det er usannsynlig at det ikke finnes en god løsning for det. Hvordan bør jeg gå frem for å implementere dette, uten å gjøre det til en rot?

Ting jeg har prøvd

Iterabel av løfter

Min første tanke var å bruke en Iterable av Promises, men jeg vet ikke hvor mange resultater jeg vil få, noe som tvinger meg til å lage en uendelig Iterable<Promise<Response?>> hvor alle Promisene vil løse seg til undefined etter et visst punkt. Men siden den er uendelig kan jeg ikke iterere over den på vanlig måte (det ville fylle hele tilgjengelig minne med Promises), og å faktisk bruke resultatene når de er i den formen betyr å hente hver Promise i resolve-funksjonen til den forrige.

Denne løsningen ser ut til å kunne fungere, men med hver linje jeg skriver blir den mindre lesbar og mer komplisert.

Slår det sammen med et atferdsemne

Mens jeg googlet på problemet fant jeg et relatert SO-spørsmål samt en GitHub-issue om rxjs backpressure, begge inneholder kodeeksempler av Ben Lesh, som tydeligvis kan brukes til å legge til backpressure til en Observable, dessverre, uansett hva jeg prøver, kan jeg ikke få kilden Observable til å sende ut verdiene sine langsommere enn den genererer dem, de blir alltid bare bufret et sted, noe som betyr at nettverksforespørslene vil skje uansett.

Fra GitHub:

// this behavior subject is basically your "give me the next batch" mechanism.
// in this example, we're going to make 5 async requests back to back before requesting more.
const BATCH_SIZE = 5;
const requests = new BehaviorSubject(BATCH_SIZE); // start by requesting five items

// for every request, pump out a stream of events that represent how many you have left to fulfill
requests.flatMap((count) => Observable.range(0, count).map(n => count - n - 1))
  // then concat map that into an observable of what you want to control with backpressure
  // you might have some parameterization here you need to handle, this example is simplified
  // handle side effects with a `do` block
  .concatMap(() => getSomeObservableOfDataHere().do(stuffWithIt), (remaining) => remaining)
  // narrow it down to when there are no more left to request,
  // and pump another batch request into the BehaviorSubject
  .filter(remaining => remaining === 0)
  .mapTo(BATCH_SIZE)
  .subscribe(requests);

Fra StackOverflow:

// start with 5 values
const controller = new Rx.BehaviorSubject(5);

// some observable source, in this case, an interval.
const source = Rx.Observable.interval(100)

const controlled = controller.flatMap(
      // map your count into a set of values
      (count) => source.take(count), 
      // additional mapping for metadata about when the block is done
      (count, value, _, index) => {
        return { value: value, done: count - index === 1 }; 
      })
      // when the block is done, request 5 more.
      .do(({done}) => done && controller.next(5))
      // we only care about the value for output
      .map(({value}) => value);


// start our subscription
controlled.subscribe(x => {
  console.log(x)
});

Jeg kan ta feil om dette, men det virker for meg at når jeg abonnerer på en Observable, produserer den verdiene sine så raskt som mulig, uten noen måte å bremse den på, så dette er sannsynligvis ikke en løsning.

Bruke ixjs

Det virker som om ixjs er ment å være en løsning på problemet mitt, men det depotet har ikke blitt oppdatert på lenge. Det finnes tydeligvis en reimplementering i TypeScript, men den ser ut til å være tidlig i utviklingen og ikke godt dokumentert ennå.

Jeg vil helst ikke være avhengig av et rammeverk som brukes av så få mennesker for det som egentlig er et veldig enkelt problem.

Refaktorering av appen

Jeg har søkt på nettet etter implementeringer av en uendelig rullende liste i TypeScript (med Angular). Min nåværende tilnærming er å ha en tjeneste som leverer et objekt som kan brukes til å hente alle resultatene. Deretter har jeg en komponent som viser dem. Alternativer ser ut til å være å gjøre sjekken for rullingsposisjon direkte i tjenesten som forespør bakenden, eller å la komponenten hente en ny Observable fra backend-tjenesten når brukeren ruller.

Begge løsningene ville tvinge meg til å blande kode, som for øyeblikket er pent separert. Jeg vil foretrekke at tjenesten returnerer noe som jeg bare kan gi til komponenten, uten at komponenten trenger å vite om nettverksforespørsler, eller at tjenesten trenger å vite om rulleposisjonen.

Svar #1

Jeg vil foreslå at du ser på mergeScan-operatoren i stedet. Det virker som om det kan passe godt her.

MergeScan ligner på expand-operatoren ved at den returnerer dataene fra forrige forespørsel som en accumulator, men i motsetning til expand fortsetter den ikke å kjøre til evig tid.

I utgangspunktet antar at du har en funksjon makeRequest(params) som tar en forespørsel og returnerer et Observable som til slutt løses til et svar, og en strøm som representerer rullehendelser som vi vil kalle fetchMore$, kan du lage en hent-på-etterspørsel-tjeneste slik:

// This is abstracted as a simple "fetch" concept but in reality should
// be hooked up to your scroll handler, properly debounced etc.
this.fetchMore$
  .mergeScan(
    // Make the request
    (acc, _) => makeRequest(acc.paging.next ? acc.paging.cursors.after : ''),
    {paging: {}}, // Initial request body
    1 // Maximum concurrency, i.e. how many requests can be in flight at once 
  )
  .pluck('data')
  .subscribe(data => {/*Do something with the data*/});

Jeg satte samtidigheten til 1 fordi selv om du kunne ha flere forespørsler i gang samtidig, er det for øyeblikket ingen garanti for rekkefølge, så resultatet kan bli at akkumulatoren kommer ut av synk hvis brukeren scroller veldig raskt, mens med en samtidighet på 1 vil dataene alltid være ordnet.

Svar #2

Bruk rett og slett ikke expand, fordi da vil du bli tvunget til å fikle med en tilpasset planlegger, noe som er unødvendig komplisert.

Du kan ganske enkelt konstruere en Subject på hvilken du sender scroll-handlinger som du mapper asynkront til nye resultater. Her er en grov skisse av ideen (uten å anta et rammeverk eller engang kjøre dette gjennom TypeScript-kompilatoren):

class InfiniteScroll<T> {
   private subject = new Rx.Subject<{}>();
   readonly data = subject
     .asObservable()
     .flatmap(_ => this.getMoreData())      

   constructor(private getMoreData: () => Promise<T[]>) { }

   onScrollDown() {
     subject.next({});
   }
}

Du kan ha nytte av å lære litt om cycle.js.

Relaterte artikler