Rxjs schließt eine Observable sofort nach Fehler statt fortfahren

Question

Ich versuche, eine Reihe von Inhalten durch eine beobachtbare zu streamen und es stoppt nach dem ersten Fehler. Es ist einfach, dies als ein Array von Elementen zu betrachten, da es sich genauso verhält.

1. ich eine beobachtbare aus einem Array von Element zu schaffen
2. jedes Element die Zuordnung zu einer URL
3. die URL als Anforderung-promise Aufruf
4. eine Raste() durchführen, die eine observable.empty kehrt() im Falle eines Fehlers

mit RxJS 5:

rx.Observable.from(array) 
    .map(self.createUrl) 
    .flatMap(x => { 
      var options = { 
      uri: url, 
      headers: { 
       "Content-Type": "application/json" 
      }; 
      return rx.Observable.fromPromise(request-promise(options)); 
     }) 
    .catch(() => { 
    return rx.Observable.empty();}) 
    .subscribe(x => console.log('success:', x), 
      e => console.log('error'), 
      () => console.log('complete')); 

Wenn diese Sequenz ausgeführt wird, stoppt der Code nach dem ersten Fehler. Ich vermute, dass das leere Observable in # 4 das Observable beendet, aber ich bin mir nicht sicher warum.

Mein gewünschter Prozess besteht darin, dass alle Elemente in dem Array unabhängig von Fehlern verarbeitet werden - schließlich werden alle erfolgreichen Elemente verarbeitet und nach jedem Fehler fortgesetzt.

Answer 1

Sie müssen nur die catch() innerhalb der flatMap setzen:

rx.Observable.from(array) 
    .map(self.createUrl) 
    .flatMap(x => { 
    var options = { 
     uri: url, 
     headers: { 
     "Content-Type": "application/json" 
     } 
    }; 
    return rx.Observable 
     .fromPromise(request-promise(options)) 
     .catch(() => rx.Observable.empty()); 
    }) 
    .subscribe(x => console.log('success:', x), 
      e => console.log('error'), 
      () => console.log('complete')); 

Nun, wenn dann innerer beobachtbare sendet einen Fehler sofort gefangen würde und nicht an den Hauptstrom durch .flatMap() propagieren.