RxJs捕獲錯誤並繼續

Question

我有一個要解析的項目列表，但其中一個解析可能會失敗。

什麼是 「RX-路」 搭上錯誤，但繼續執行序列

代碼示例：

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>

我想在非的Rx做什麼-Way：

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); 
 
    } 
 
}

在此先感謝

Answer 1

您可以實際使用的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>

Answer 2

我會建議你（在rxjs 5版現在mergeMap），而不是使用flatMap，這將讓你崩潰的錯誤，如果你不關心他們。實際上，您將創建一個內部Observable，如果發生錯誤，它可以被吞噬。這種方法的優點是可以將運算符鏈接在一起，並且如果在管道中的任何位置發生錯誤，它將自動轉發到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>

Answer 3

你需要切換到一個新的一次性流，如果發生錯誤，這將會被安全處置，並保留原始流活着：

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>

有關此技術的更多信息，請參閱此博客文章：The Quest for Meatballs: Continue RxJS Streams When Errors Occur