Wie können Funktionsaufrufe mit RXJS dynamisch verkettet werden?

Question

Ich muss das Ergebnis einer Anfrage beim ersten Anruf zwischenspeichern und dann den zwischengespeicherten Wert für nachfolgende Anrufe lesen.

Um dieses Ziel zu erreichen, verwende ich Versprechungen und ich verkette sie. Ich habe eine funktionierende Lösung, aber ich würde es gerne in RxJS Observables statt Promises konvertieren.

Hier ist meine Arbeitslösung:

private currentPromise: Promise<{ [key: string]: any }>; 
private cache: any; 
public getSomething(name: string): Promise<number>{ 
    return this.currentPromise = !this.currentPromise ? 
    this._getSomething(name) : 
    new Promise((r) => this.currentPromise.then(() => this._getSomething(name).then((res) => r(res)))); 
} 

private _getSomething(name: string): Promise<any> { 
    return new Promise((resolve) => { 
    if (this.cache[name]) { 
     this.messages.push("Resolved from cache"); 
     resolve(this.cache[name]); 
     } else { 
     // Fake http call. I would use Angular's Http class. 
     setTimeout(()=> {this.messages.push("Resolved from server"); this.cache[name] = name; resolve(this.cache[name]); }, 2000); 
     } 
    }); 
} 

this.getSomething("thing1").then((res)=>this.messages.push(res)); 
this.getSomething("thing1").then((res)=>this.messages.push(res)); 
this.getSomething("thing2").then((res)=>this.messages.push(res)); 
this.getSomething("thing2").then((res)=>this.messages.push(res)); 
this.getSomething("thing1").then((res)=>this.messages.push(res)); 
this.getSomething("thing2").then((res)=>this.messages.push(res)); 
this.getSomething("thing1").then((res)=>this.messages.push(res)); 
this.getSomething("thing2").then((res)=>this.messages.push(res)); 

Sie können es auf dieser plunkr testen: https://plnkr.co/edit/j1pm2GeQf6oZwRvbUsXJ?p=preview

Wie erreiche ich das Gleiche mit RxJS 5 Beta?

aktualisieren

Bergi Kommentare Nach ich meine plunkr und meinen Code aktualisiert es näher zu bringen meine realen Fall

Answer 1

AsyncSubjects sind die Rx-Analogon von Versprechungen. publishLast ist der beste Weg, um aus einem Observablen einen zu machen. So etwas sollte funktionieren:

private cache: {string: Rx.Observable<any>}; 

public getSomethings(names: string[]) : Rx.Observable<any> { 
    // call getSomething for each entry in names 
    // streams is an array of observables 
    const streams = names.map(name => this.getSomething(name)); 

    // transform streams into an observable with an array of results 
    return Observable.zip(streams); 
} 

public getSomething(name: string) : Rx.Observable<any> { 
    if (!this.cache[name]) { 
     // create the request observable 
     // const request = Rx.Observable.ajax(...); // http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#static-method-ajax 
     // for now, just wait 2 seconds and return name 
     const request = Rx.Obsevable.of(name).delay(2000); 

     // use "do" to log whenever this raw request produces data 
     const loggedRequest = request.do(v => this.messages.push("retrieved from server " + v)); 

     // create an observable that caches the result 
     // in an asyncSubject 
     const cachedRequest = loggedRequest.publishLast(); 

     // store this in our cache object 
     this.cache[name] = cachedRequest; 
    } 

    // return the cached async subject 
    return this.cache[name]; 
} 


// usage 
this.getSomething("thing1").subscribe(v => this.messages.push("received " + v)); 
this.getSomething("thing1").subscribe(v => this.messages.push("received " + v)); 
this.getSomething("thing1").subscribe(v => this.messages.push("received " + v)); 
this.getSomething("thing1").subscribe(v => this.messages.push("received " + v));