Eine beliebige Anzahl von sequentiellen, abhängigen Anfragen mit `cycle-http` erstellen

Question

Gibt es irgendwelche Beispiele für die Erstellung einer beliebigen Anzahl von sequentiellen, abhängigen http-Anfragen mit cycle-http?

Ich möchte jede Seite von einer API holen, wo die nächste Anfrage nur mit Daten auf der aktuellen Seite gemacht werden kann.

Ich habe versucht, this answer anzupassen, die Observable.merge() verwendet, aber ich bin nicht sicher, wie die & sinken auf cycle-http Quellen bis haken.

Referenzen

• How to request data sequentially in Cycle.js?
• RxJs Observable Pagination

Answer 1

Hier ist eine weitere Ansicht auf eine beliebige Anzahl von sequentiellen, abhängigen Anfragen mit Cycle.js und der @cycle/fetch Treiber.

(mit GitHub Benutzern API. Die Benutzer Abfrage gibt 30 Benutzer pro Seite und der since URL-Parameter ist eine Benutzer-ID-Nummer und startet die Abfrage bei der nächsten Benutzer-ID.)

Zuerst wird der Hauptteil die main Funktion mit Kommentaren:.

const listResponse$ = sources.FETCH // response returned from FETCH driver 
    .mergeAll() 
    .flatMap(res => res.json()) 
    .scan(
    ((userstotals, users) => 
     [ 
     userstotals[0] + 1, // page count 
     users[29] && users[29].id, // last id on full page 
     userstotals[2].concat(users) // collect all users 
     ] 
    ), 
    [0, undefined, []] // default accumulator 
) 
    .share(); // allows stream split 


// Branch #1 - cycle again for more pages 
const listRequest$ = listResponse$ 
    .filter(users => 
    0 < users[1] && // full page id exists 
    maxpages > users[0]; // less than maxpages 
) 
    .startWith('initial') 
    .map(users => 
    `https:\/\/api.github.com/users?since=${ 
     (!isNaN(parseInt(users[1], 10)) && users[1]) || // last id full page 
     idstart // default id start 
    }` 
); 


// Branch #2 - display 
const dom$ = listResponse$ 
    .map(userstotals => div(JSON.stringify(userstotals[2]))); 

(Dies ist eine aktualisierte Antwort, die ich realisiert die scan s in einem kombiniert werden können.

)

ERKLÄRUNG: Ziehen Sie zuerst die Antwort des sources Eigenschaft FETCH, glätten Sie es und ziehen Sie die JSON, dann scan zu zählen, wie viele Seiten bisher abgefragt. Die Anzahl der abgefragten Seiten wird später mit maxpages verglichen, um die vorbestimmte Anzahl nicht zu überschreiten. Als Nächstes erhalten Sie die letzte id einer vollständigen Seite, falls vorhanden, und zuletzt concat die aktuelle Benutzerseite mit der Sammlung von bisher gesammelten Benutzerseiten. Nach dem Akkumulieren der Antwortinformationen share der Strom, so kann es in zwei Zweige aufgeteilt werden.

Die erste Verzweigung wird verwendet, um die Abfrage über den FETCH-Treiber erneut abzufragen, um weitere Seiten abzufragen. Aber zuerst filter, um nach der letzten Seite und Anzahl der abgefragten Seiten zu suchen. Wenn die ID keine Nummer ist, ist die letzte Seite erreicht. Fahren Sie nicht fort, wenn die letzte Seite bereits erreicht ist und daher keine weiteren Seiten abgefragt werden müssen. Fahren Sie auch nicht fort, wenn die Anzahl der abgefragten Seiten den Wert maxpages überschreitet.

Der zweite Zweig einfach in die akkumulierte Reaktion erreicht die vollständige Liste der Benutzer zu bekommen, dann JSON.stringify s das Objekt und wandelt es in ein virtuelles dom-Objekt (div Methode) zur Anzeige an die DOM-Treiber gesendet werden.

---

Und hier ist das komplette Skript:

import Cycle from '@cycle/rx-run'; 
import {div, makeDOMDriver} from '@cycle/dom'; 
import {makeFetchDriver} from '@cycle/fetch'; 

function main(sources) { // provides properties DOM and FETCH (evt. streams) 

    const acctok = ''; // put your token here, if necessary 
    const idstart = 19473200; // where do you want to start? 
    const maxpages = 10; 

    const listResponse$ = sources.FETCH 
    .mergeAll() 
    .flatMap(res => res.json()) 
    .scan(
     ((userstotals, users) => 
     [ 
      userstotals[0] + 1, // page count 
      users[29] && users[29].id, // last id on full page 
      userstotals[2].concat(users) // collect all users 
     ] 
    ), 
     [0, undefined, []] 
    ) 
    .share(); 

    const listRequest$ = listResponse$ 
    .filter(function (users) { 
     return 0 < users[1] && maxpages > users[0]; 
    }) 
    .startWith('initial') 
    .map(users => 
     `https:\/\/api.github.com/users?since=${ 
     (!isNaN(parseInt(users[1], 10)) && users[1]) || // last id full page 
     idstart // default id start 
     }` //&access_token=${acctok}` 
    ); 

    const dom$ = listResponse$ 
    .map(userstotals => div(JSON.stringify(userstotals[2]))); 

    return { 
    DOM: dom$, 
    FETCH: listRequest$ 
    }; 
} 

Cycle.run(main, { 
    DOM: makeDOMDriver('#main-container'), 
    FETCH: makeFetchDriver() 
}); 

---

(.. Meine erste Antwort, für die Nachwelt links Beachten Sie die beiden scan s)

const listResponse$ = sources.FETCH 
    .mergeAll() 
    .flatMap(res => res.json()) 
    .scan(((userscount, users) =>    // <-- scan #1 
    [userscount[0] + 1, users]), [0, []] 
) 
    .share(); 

const listRequest$ = listResponse$ 
    .filter(function (users) { 
    return users[1][29] && users[1][29].id && 
     maxpages > users[0]; 
    }) 
    .startWith('initial') 
    .map(users => 
    `https://api.github.com/users?since=${ 
     (users[1][users[1].length-1] && users[1][users[1].length-1].id) || 
     idstart 
     }` 
); 

const dom$ = listResponse$ 
    .scan(function (usersall, users) {   // <-- scan #2 
    usersall.push(users); 
    return usersall; 
    }, []) 
    .map(res => div(JSON.stringify(res))); 

Durch scan ing einmal , im Voraus musste ich dann die letzte ID der ganzen Seite abrufen, falls vorhanden, und diese im Akkumulator speichern.

Answer 2

es wird besser sein, wenn Sie einen Code Beispiel geben. jedoch grundlegende Logik kann so aussehen:

1. Karte den Antwortstrom auf eine Anfrage Strom
2. Startanforderung Strom mit einer anfänglichen Anfrage

-Code so sein würde:

function main (sources){ 
 
    const initialRequest = { 
 
    url: 'http://www.google.com' 
 
    }; 
 
    
 
    const request$ = sources.HTTP 
 
    .filter(response$ => /*FILTER LOGIC GOES HERE */) 
 
    .switch()//or you can use flatMap 
 
    .map(response =>/* MAP RESPONSE TO A NEW REQUEST */) 
 
    .startWith(initialRequest); 
 
    
 
    return { 
 
    HTTP: request$ 
 
    }; 
 
}

Answer 3

Also dies ist wahrscheinlich schrecklich kompliziert und ich sollte es Schrott richtig Erdal's answer, versuchen, aber hier ist das, was ich mit ...

habe kommen

Nutzung

export default function app({HTTP}) { 
    const { 
    allPagesRequest$: staffPagesReq$, 
    latestData$: staff$, 
    } = getAllPages({url: '/staff', HTTP}); 

    // staff$ is converted to vdom... 

    return /* sinks */ { 
    DOM: staffPagesReq$, 
    HTTP: staffVdom$, 
    } 
} 

Implementierung

const fetchNthPage = (optsIn) => { 
    const opts = merge(
    { 
     url: '', 
     page: 0, 
     HTTP: undefined, 
    }, optsIn 
); 

    const u = new URI(opts.url) 
    .setQuery({'_page': opts.page.toString()}); 

    const pageNResponse$ = opts.HTTP 
    .filter(
     res$ => res$.request.url === urlForEndpoint(u) 
    ) 
    .flatMap(
     res$ => res$.catch(
     err => Observable.of(
      { 
      body: {'error in response:': err.toString()} 
      } 
     ) 
    ) 
    ) 
    .map(res => res.body) 
    .take(1) 
    .shareReplay(1); 

    return Observable.of({ 
    pageNRequest$: Observable.of(basicRequestObject(u)), 
    pageNResponse$: pageNResponse$, 
    opts:   opts 
    }); 
}; 


const encapsulateAs = typeName => data => { 
    return {type: typeName, data} 
}; 


const fetchAllPagesIndividually = (optsIn) => { 
    const opts = merge(
    { 
     url: '', 
     page: 0, 
     HTTP: undefined, 
    }, 
    optsIn 
); 

    return Observable.defer(
    () => fetchNthPage(opts) 
     .flatMap(x => { 
     const mergedItems$ = Observable 
      .merge(
      x.pageNRequest$.map(encapsulateAs('request')), 
      x.pageNResponse$.map(encapsulateAs('response')) 
     ); 


     const optsForNextPage = merge(opts, { 
      page: opts.page + 1 
     }); 

     const next$ = Observable 
      .never() // `next$` shouldn't end when `pageNResponse$` does 
      .merge(x.pageNResponse$) 
      .shareReplay(1) 
      .takeWhile(res => { 
      //noinspection UnnecessaryLocalVariableJS 
      let isFullPage = path(['response', 'length'], res) === apiPageSize; 
      return isFullPage; 
      }) 
      .flatMap(() => { 
      return fetchAllPagesIndividually(optsForNextPage) 
      }); 

     //noinspection UnnecessaryLocalVariableJS 
     const concattedItem$ = Observable 
      .concat(
      mergedItems$, 
      next$ 
     ); 

     return concattedItem$ 
     }) 
     .shareReplay(1) 
); 
}; 


const concatPages = (acc, currentVal, index, source) => acc.concat(currentVal); 

const typeIs = typeStr => compose(
    equals(typeStr), 
    prop('type') 
); 

const typeNotIn = typesArray => compose(
    not, 
    unary(flip(contains)(typesArray)), 
    prop('type') 
); 

const getAllPages = (optsIn) => { 
    const f$ = fetchAllPagesIndividually(optsIn) 
    .shareReplay(1); 

    const allPagesRequest$ = f$ 
    .filter(typeIs('request')) 
    .map(prop('data')); 

    const allPagesResponse$ = f$ 
    .filter(typeIs('response')) 
    .map(prop('data')); 

    const theRest$ = f$ 
    .filter(typeNotIn(['request', 'response', 'data'])); 

    const latestData$ = allPagesResponse$ 
    .map(prop('response')) 
    .scan(concatPages); 

    return { 
    allPagesRequest$, 
    allPagesResponse$, 
    latestData$, 
    theRest$, 
    } 
}; 

compose() , , merge(), unary() usw. sind von Ramda.