RxJS: Übergabefehler in cycle.js benutzerdefinierten Treiber

Question

Ich habe einige Fehlerbehandlungscode in meiner main Funktion wie folgt implementiert. Es verwendet den Operator catch, um Fehler in einem Stream zu filtern und zu melden und sie in einem anderen zu ignorieren. Dadurch kann ich Fehler, die bei Anfragen auftreten, kennen und darüber berichten, während der gesamte Stream nicht fehlschlägt, so dass nachfolgende Anfragen fortgesetzt werden können.

Aus Gründen, die in den folgenden Codeschnipsel nicht offensichtlich sind, treibe ich einen benutzerdefinierten Treiber, um Daten anzufordern und zu verarbeiten. Ich verwende nicht den Zyklus-HTTP-Treiber.

Hier ist mein Code, der erfolgreich einen Fehler meldet:

function main(sources) { 

    // Catch driver errors so they can be logged 
    const error$ = sources.CustomDriver 
     .map(x => x.catch(e => Rx.Observable.just(e))) 
     .flatMap(p => p) 

    // Filter out the errors to deal with requests that did not fail 
    const data$ = sources.CustomDriver 
     .map(x => x.catch(e => Rx.Observable.empty())) 
     .flatMap(p => p) 

    return { 
     CustomDriver: Rx.Observable.just('initial event'), 
     Log: data$, 
     Error: error$ 
    } 
} 

Cycle.run(main, { 
    CustomDriver: makeCustomDriver(), 
    Log: msg$ => { msg$.subscribe(
     msg => console.log('LOG: ', msg), 
     err => console.log('problem with Log driver: ', err), 
     () => console.log('Log Completed') 
    ) }, 
    Error: msg$ => { msg$.subscribe(
     e => console.log('ERR: ', e), 
     err => console.log('problem with Error driver:', err), 
     () => console.log('Error Completed') 
    ) } 
}) 

function makeCustomDriver() { 
    return function customDriver(requests$) { 
     return requests$ 
      .map(request => Rx.Observable.fromPromise(makeFailedRequest())) 
    } 
} 

function makeFailedRequest() { 
    console.log('some API request') 
    return Promise.reject('error') 
} 

Der Ausgang ist wie folgt:

some API request 
some API request 
Log Completed 
ERR: error 
Error Completed 

Auf der positiven Seite der Fehler gemeldet wird. Die API-Anfrage wird jedoch tatsächlich zweimal gemacht, was ich ursprünglich nicht erwartet habe. Nachdem ich etwas mehr RxJS gelernt hatte und ein besseres Verständnis von heißen und kalten Observablen hatte, erkannte ich, dass ich zwei Subskriptionen für den CustomDriver-Stream erstellte (einen für Fehler $ und einen für Daten $) und weil CustomDriver Observable kalt war würde die Observable.just für jeden Teilnehmer wiederholen.

Also habe ich versucht, meine CustomDriver Observavble heiß mit share zu machen:

function makeCustomDriver() { 
    return function customDriver(requests$) { 
     return requests$ 
      .map(request => Rx.Observable.fromPromise(makeFailedRequest())) 
      .share() 
    } 
} 

Mit dieser Änderung wird die Ausgabe wie folgt:

some API request 
Error Completed 
Log Completed 

So konnte ich der doppelten Anforderung loszuwerden aber der Fehler wurde dabei verschluckt.

Was passiert mit share, dass die Fehler verloren gehen und wie kann ich doppelte Anfragen vermeiden, ohne Fehler zu verlieren?

Answer 1

.shareReplay(1) scheint das gewünschte Ergebnis zu geben.

Answer 2

Es gibt eine Fabrik für die Herstellung von benutzerdefinierten Treibern der Art, die Sie wollen (von Promises) https://github.com/whitecolor/cycle-async-driver es enthält, Helfer für den Umgang mit Fehlern (success und failure).

können Sie erstellt Treiber einfach so:

import {makeAsyncDriver} from 'cycle-async-driver' 

    customDriver = makeAsyncDriver(
    (request) => requestHanderThatReturnsPromise(reques) 
    )