Ошибка RxJs и продолжается

У меня есть список элементов для синтаксического анализа, но разбор одного из них может завершиться неудачей.

Что такое "Rx-Way", чтобы поймать ошибку, но продолжить выполнение последовательности

Пример кода:

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>

Ответ 1

Я бы предположил, что вместо этого вы используете flatMap (теперь mergeMap в rxjs версии 5), что позволит вам свернуть ошибки, если вас это не волнует. Эффективно, вы создадите внутреннее Наблюдаемое, которое может быть проглочено, если произошла ошибка. Преимущество этого подхода состоит в том, что вы можете объединять операторов, и если ошибка происходит в любом месте конвейера, она автоматически пересылается в блок catch.

var observable = Rx.Observable.from([0, 1, 2, 3, 4, 5])
  .flatMap((value) =>          
    Rx.Observable.if(() => value != 3,
      Rx.Observable.just(value),
      Rx.Observable.throw(new Error("Value cannot be 3")))
    //This will get skipped if upstream throws an error
    .map(v => v * 2)
    .catch((err) => {
      console.log("Caught Error, continuing")
      //Return an empty Observable which gets collapsed in the output
      return Rx.Observable.empty();
    })
  );

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

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

Ответ 2

Вам нужно переключиться на новый одноразовый поток, и если в нем возникнет ошибка, он будет безопасно размещен и сохранит исходный поток:

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>

Ответ 3

Фактически вы можете использовать try/catch внутри вашей карты для обработки ошибки. Вот фрагмент кода

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>