Observable's doOnError location corretta

Question

Sono una specie di nuovo a Observers e sto ancora cercando di capirli. Ho il seguente pezzo di codice:

observableKafka.getRealTimeEvents() 
     .filter(this::isTrackedAccount) 
     .filter(e -> LedgerMapper.isDepositOrClosedTrade((Transaction) e.getPayload())) 
     .map(ledgerMapper::mapLedgerTransaction) 
     .map(offerCache::addTransaction) 
     .filter(offer -> offer != null) // Offer may have been removed from cache since last check 
     .filter(Offer::isReady) 
     .doOnError(throwable -> { 
       LOG.info("Exception thrown on realtime events"); 
      }) 
     .forEach(awardChecker::awardFailOrIgnore); 

getRealTimeEvents() restituisce un Observable<Event>.

La posizione di .doOnError è importante? Inoltre, qual è l'effetto dell'aggiunta di più di una chiamata in questo pezzo di codice? Mi sono reso conto che posso farlo e tutti vengono invocati, ma non sono sicuro di quale possa essere il suo scopo.

Answer 1

Sì, lo fa. doOnError agisce quando un errore passa attraverso il flusso in quel punto specifico, quindi se l'operatore (i) prima dei (i) lancio (i) doOnError, verrà chiamata l'azione. Tuttavia, se si posiziona lo doOnError più in alto, può essere chiamato o meno a seconda di quali operatori downstream sono presenti nella catena.

Dato

Observer<Object> ignore = new Observer<Object>() { 
    @Override public void onCompleted() { 
    } 
    @Override public void onError(Throwable e) { 
    } 
    @Override public void onNext(Object t) { 
    } 
}; 

Ad esempio, il seguente codice sarà sempre chiamare doOnError:

Observable.<Object>error(new Exception()).doOnError(e -> log(e)).subscribe(ignore); 

Tuttavia, questo codice non:

Observable.just(1).doOnError(e -> log(e)) 
.flatMap(v -> Observable.<Integer>error(new Exception())).subscribe(ignore); 

maggior parte degli operatori si riprenderà eccezioni che hanno origine a valle.

Aggiunta multipe doOnError è vitale se si trasformare un'eccezione tramite onErrorResumeNext o onExceptionResumeNext:

Observable.<Object>error(new RuntimeException()) 
.doOnError(e -> log(e)) 
.onErrorResumeNext(Observable.<Object>error(new IllegalStateException())) 
.doOnError(e -> log(e)).subscribe(ignore); 

in caso contrario, ci si accede la stessa eccezione in più posizioni della catena.

Answer 2

i metodi doOn??? sono disponibili per gli effetti collaterali, l'elaborazione che in realtà non è il valore del core business, diciamo. La registrazione è un uso perfetto per quello. Detto questo, a volte vuoi fare qualcosa di più significativo con un errore, come riprovare, o mostrare un messaggio a un utente, ecc ... In questi casi, il modo "rx" sarebbe elaborare l'errore in una chiamata subscribe .

doOnError (e gli altri metodi doOn) avvolge l'originale Observable in uno nuovo e aggiunge comportamento ad esso (attorno al suo metodo onError, ovviamente). Ecco perché puoi chiamarlo più volte. Un altro vantaggio di poterlo chiamare in qualsiasi punto della catena è che è possibile accedere agli errori che altrimenti verrebbero nascosti all'utente del flusso (il Subscriber), ad esempio perché c'è un nuovo tentativo nella catena ...