RxJs attrape l'erreur et continue

Question

J'ai une liste d'éléments à analyser, mais l'analyse de l'un d'entre eux peut échouer.

Quel est le "Rx-Way" pour attraper l'erreur mais continuer à exécuter la séquence ?

Exemple de code :

var observable = Rx.Observable.from([0,1,2,3,4,5])
.map(
  function(value){
      if(value == 3){
        throw new Error("Value cannot be 3");
      }
    return value;
  });

observable.subscribe(
  function(value){
  console.log("onNext " + value);
  },
  function(error){
    console.log("Error: " + error.message);
  },
  function(){
    console.log("Completed!");
  });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.6/rx.all.js"></script>

Ce que je veux faire de manière non-Rx :

var items = [0,1,2,3,4,5];

for (var item in items){
  try{
    if(item == 3){
      throw new Error("Value cannot be 3");
    }
    console.log(item);
  }catch(error){
     console.log("Error: " + error.message);
  }
}

Merci d'avance

Answer 1

Je vous suggère d'utiliser flatMap (maintenant mergeMap dans la version 5 de rxjs) à la place, ce qui vous permettra de faire disparaître les erreurs si vous ne vous en souciez pas. En fait, vous allez créer un Observable interne qui peut être avalé si une erreur se produit. L'avantage de cette approche est que vous pouvez enchaîner les opérateurs et si une erreur se produit n'importe où dans le pipeline, elle sera automatiquement transmise au bloc catch.

const {from, iif, throwError, of, EMPTY} = rxjs;
const {map, flatMap, catchError} = rxjs.operators;

// A helper method to let us create arbitrary operators
const {pipe} = rxjs;

// Create an operator that will catch and squash errors
// This returns a function of the shape of Observable<T> => Observable<R>
const mapAndContinueOnError = pipe(
  //This will get skipped if upstream throws an error
  map(v => v * 2),
  catchError(err => {
    console.log("Caught Error, continuing")
    //Return an empty Observable which gets collapsed in the output
    return EMPTY;
  })
)

const observable = from([0, 1, 2, 3, 4, 5]).pipe(
  flatMap((value) => 
    iif(() => value != 3, 
      of(value), 
      throwError(new Error("Value cannot be 3"))
    ).pipe(mapAndContinueOnError)
  )
);

observable.subscribe(
  (value) => console.log("onNext " + value), (error) => console.log("Error: " + error.message), () => console.log("Completed!")
);

<script src="https://unpkg.com/rxjs@7.0.0/dist/bundles/rxjs.umd.min.js"></script>

Answer 2

Vous devez basculer vers un nouveau flux jetable, et si une erreur se produit à l'intérieur de celui-ci, il sera éliminé en toute sécurité, et le flux original restera en vie :

Rx.Observable.from([0,1,2,3,4,5])
    .switchMap(value => {

        // This is the disposable stream!
        // Errors can safely occur in here without killing the original stream

        return Rx.Observable.of(value)
            .map(value => {
                if (value === 3) {
                    throw new Error('Value cannot be 3');
                }
                return value;
            })
            .catch(error => {
                // You can do some fancy stuff here with errors if you like
                // Below we are just returning the error object to the outer stream
                return Rx.Observable.of(error);
            });

    })
    .map(value => {
        if (value instanceof Error) {
            // Maybe do some error handling here
            return `Error: ${value.message}`;
        }
        return value;
    })
    .subscribe(
      (x => console.log('Success', x)),
      (x => console.log('Error', x)),
      (() => console.log('Complete'))
    );

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.1/Rx.min.js"></script>

Plus d'informations sur cette technique dans cet article de blog : La quête des boulettes de viande : Poursuivre les flux RxJS en cas d'erreur

Answer 3

Pour garder votre endlessObservable$ de mourir, vous pouvez mettre votre failingObservable$ dans un opérateur de cartographie d'ordre supérieur (par ex. switchMap , concatMap , exhaustMap ...) et avaler l'erreur en terminant le flux interne avec un empty() observable ne renvoyant aucune valeur.

Utilisation de RxJS 6 :

endlessObservable$
    .pipe(
        switchMap(() => failingObservable$
            .pipe(
                catchError((err) => {
                    console.error(err);
                    return EMPTY;
                })
            )
        )
    );

Answer 4

Vous pouvez en fait utiliser essai/attente à l'intérieur de votre carte pour gérer l'erreur. Voici l'extrait de code

var source = Rx.Observable.from([0, 1, 2, 3, 4, 5])
    .map(
        function(value) {
            try {
                if (value === 3) {
                    throw new Error("Value cannot be 3");
                }
                return value;

            } catch (error) {
                console.log('I caught an error');
                return undefined;
            }
        })
    .filter(function(x) {
        return x !== undefined; });

source.subscribe(
    function(value) {
        console.log("onNext " + value);
    },
    function(error) {
        console.log("Error: " + error.message);
    },
    function() {
        console.log("Completed!");
    });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.6/rx.all.js"></script>