Javascript RxJs 捕获错误并继续

RxJs catch error and continue

提问by Cheborra

I have a list of items to parse, but the parsing of one of them can fail.

我有一个要解析的项目列表，但其中一个的解析可能会失败。

What is the "Rx-Way" to catch error but continue executing the sequence

捕获错误但继续执行序列的“Rx-Way”是什么

Code Sample:

代码示例：

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>

What I want to do in a non-Rx-Way:

我想以非 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);
  }
}

Thanks in advance

提前致谢

回答by paulpdaniels

I would suggest that you use flatMap(now mergeMapin rxjs version 5) instead, which will let you collapse errors if you don't care about them. Effectively, you will create an inner Observable that can be swallowed if an error occurs. The advantage of this approach is that you can chain together operators and if an error occurs anywhere in the pipeline it will automatically get forwarded to the catch block.

我建议您改用flatMap（现在mergeMap在 rxjs 版本 5 中），如果您不关心它们，它将让您折叠错误。实际上，您将创建一个内部 Observable，如果发生错误，它可以被吞下。这种方法的优点是你可以将操作符链接在一起，如果管道中的任何地方发生错误，它都会自动转发到 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/bundles/rxjs.umd.min.js"></script>

回答by iamturns

You need to switch to a new disposable stream, and if an error occurs within it will be disposed safely, and keep the original stream alive:

您需要切换到一个新的一次性流，如果其中发生错误将被安全处理，并保持原始流处于活动状态：

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>

More info on this technique in this blog post: The Quest for Meatballs: Continue RxJS Streams When Errors Occur

这篇博文中有关此技术的更多信息：The Quest for Meatballs: Continue RxJS Streams When Errors Occurs

回答by Vedran

To keep your endlessObservable$from dying you can put your failingObservable$in a higher-order mapping operator (e.g. switchMap, concatMap, exhaustMap...) and swallow the error there by terminating the inner stream with an empty()observable returning no values.

为了防止您endlessObservable$死亡，您可以将您的failingObservable$放入高阶映射运算符（例如switchMap, concatMap, exhaustMap...）并通过使用不empty()返回值的observable终止内部流来吞下错误。

Using RxJS 6:

使用 RxJS 6：

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

回答by Toan Nguyen

You can actually use try/catchinside your mapfunction to handle the error. Here is the code snippet

您实际上可以在map函数中使用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>