Chaining Subjects - RxJS/Angular2

Question

Ich bin sehr neu zu Observablen und ich versuche, meine Meinung über eine gute Möglichkeit, Themen zu ketten. Im Wesentlichen versuche ich eine Reihe von Observablen zu nehmen, die alle den gleichen Typ aussenden und sie zusammenketten, so dass, wenn ich next zum ersten Thema anrufe, jedes weitere Thema (vorausgesetzt, dass error dazwischen passiert) eine Chance bekommt Nimm den Wert, manipuliere ihn und gib ihn an das folgende Subjekt weiter, bis es das letzte Subjekt erreicht, das das Endergebnis ausstrahlen wird.

Ich habe meine eigene Klasse rolle mit diesem für den Umgang, aber es scheint, wie dieser Fall kommen würde die ganze Zeit mit Observablen so habe ich mich gefragt, ob jemand etwas weiß, das bereits in RxJS oder Angular2 errichtet wird, das tut Dies.

Auch versuche ich Subjekte zu zwingen, etwas zu tun, was sie nicht tun sollen? Gibt es einen besseren Weg, um Algorithmen so miteinander zu verketten, dass jede Funktion die Möglichkeit hat, die Eingabe der Reihe nach zu manipulieren, bevor sie schließlich zurückgegeben wird, mit einem möglichen Fehler, wenn es nötig ist? Was wird in diesem Fall als "Best Practice" angesehen?

EDIT

Um ein wenig mehr klar zu sein, was ich rede, das ist etwas wie das, was ich suche:

var wrapper = Subject.chain(subject1, subject2, subject3) 

// Subscriptions happen here 

/** 
* This calls subject1.next("HI"), 
* which then calls subject2.next() with the result of subject1's manipulation of "HI", 
* which then calls subject3.next() with the result of subject2's manipulation of the subject1's manipulation of "HI", 
* which then emits the result of subject3's manipulation of subject2's manipulation of subject1's manipulation of "HI" 
*/ 
wrapper.next("HI"); 

Abhilfe

In Wenn jemand dies in der Zukunft findet, ist dies die Problemumgehung, die ich mit Hilfe der Array.reduce Funktion verwendet habe. Es ist nicht perfekt, aber es wird für mich tun:

chain<T>(source: Observable<T>, destination: Subject<T>): Observable<T> 
{ 
    let processed = false; 
    return source.catch(
     err => { 
      let ret = new ReplaySubject<T>(1); 
      destination.first().subscribe(ret); 
      processed = true; 
      destination.error(err); 
      return ret; 
     } 
    ).finally(
     () => { 
      // TODO: Allow sources to not propagate complete status 
      !processed && destination.complete(); 
      processed = true; 
     } 
    ).flatMap(
     (next: T) => { 
      let ret = destination; 
      if(!processed) 
      { 
       ret = new ReplaySubject<T>(1); 
       destination.first().subscribe(ret); 
       destination.next(next); 
      } 
      processed = false; 
      return ret; 
     } 
    ); 
} 

Answer 1

Zur Kette Observablen, die Sie berücksichtigen sollten Operatoren wie flatMap oder switchMap verwenden. Hier

ist eine Probe mit HTTP:

this.http.get('...') 
    .map(res => res.json()) 
    .flatMap(data => { 
     // receive the result of the first request 
     // use this result to execute another one 
     return this.http.get('...') 
       .map(res => res.json()); 
    }).subscribe(data => { 
     // receive the result of the second request 
    }); 

Wenn Sie in ein gutes Tutorial zu diesem Thema interessiert sind, können Sie an diesem einen Blick:

• https://gist.github.com/staltz/868e7e9bc2a7b8c1f754

Answer 2

können Sie es verwenden, abonnieren Sie die Themen zu ketten.

subject1 
    .subscribe(subject2); 

subject2 
    .subscribe(subject3); 

subject3 
    .subscribe(x=> { 
     // do you want to do 
    });