javascript - How to dynamically chain function calls using RXJS? -

i need cache result of request on first call read cached value subsequent calls.

to achieve goal, using promises , chaining them. have working solution convert rxjs's observables instead of promises.

here working solution:

private currentpromise: promise<{ [key: string]: }>; 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 cache");         resolve(this.cache[name]);       } else {         // fake http call. use angular's http class.         settimeout(()=> {this.messages.push("resolved 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)); 

you can test on plunkr: https://plnkr.co/edit/j1pm2geqf6ozwrvbusxj?p=preview

how achieve same thing rxjs 5 beta?

update

following bergi's comments updated plunkr , code bring closer real case

asyncsubjects rx analog of promises. publishlast best way turn observable one. should work:

private cache: {string: rx.observable<any>};  public getsomethings(names: string[]) : rx.observable<any> {     // call getsomething each entry in names     // streams array of observables     const streams = names.map(name => this.getsomething(name));      // transform streams observable array of results     return observable.zip(streams); }  public getsomething(name: string) : rx.observable<any> {     if (!this.cache[name]) {         // create request observable         // const request = rx.observable.ajax(...); // http://reactivex.io/rxjs/class/es6/observable.js~observable.html#static-method-ajax         // now, wait 2 seconds , return name         const request = rx.obsevable.of(name).delay(2000);          // use "do" log whenever raw request produces data         const loggedrequest = request.do(v => this.messages.push("retrieved server " + v));          // create observable caches result         // in asyncsubject         const cachedrequest = loggedrequest.publishlast();          // store in our cache object         this.cache[name] = cachedrequest;     }      // return 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));