Dependent and parallel requests through Observable

问题:

I am putting together an application that displays information about 'Modyules' (on a course) that it has to pull via two http.gets (I have no control over the apis unfortunately):

1. Gets the list of Modyule.ids - need to store the id
2. For each Modyule.id another http.get to get the Modyule.name

In my modyule.component.ts, I have:

this.modyuleService.getModyules()
     .subscribe(
          modyules => this.modyules = modyules,
          error =>  this.errorMessage = <any>error
     );

In my modyule.service.ts, I have:

getModyules (): Observable<Modyule[]> {
        return this.http.get(listOfModyuleIdsUrl + '.json')
            .map(this.getModyulesDetails)
            .catch(this.handleError);
    }

But the getModyuleDetails is where I am struggling. This is where I have got to so far, mainly based on:

http://www.syntaxsuccess.com/viewarticle/angular-2.0-and-http

and looking at, but not seeing how I can apply:

https://stackoverflow.com/a/35676917/2235210

private getModyulesDetails = (res: Response) => {  // instance method for defining function so that the this.http below refers to the class.
    let listOfModyules = res.json();
    let modyulesToReturn = [];

    for (let individualModyule of listOfModyules){
        let tempModyule = new Modyule;
        tempModyule.id = individualModyule.id;

        this.http.get(individualModyuleUrl + individualModyule.id + '.json')
            .map((res: Response) => res.json()))
            .subscribe(res => {
                tempModyule.name = res.name
                modyulesToReturn.push(tempModyule);
            });
        }
    return modyulesToReturn;      
}

Now this is working on my local machine where I have mocked up the .json responses as static .json files..but, when I'm dealing with the real api, I can't rely on the http.gets in the for loop to have completed before returning modyulesToReturn.

I've had a couple of attempts at forkJoin but don't see how I can combine this with the requirement to capture the Modyule.id from the first http.get.

I'd be very grateful for any pointers for how to do parallel requests which depend on an initial request and yet be able to combine data from both.

EDIT in reponse to discussion with @Matt and @batesiiic:

So, picking up on @batesiiic's suggested rewrite of getModyulesDetails to return a Subject, the problem I can't get my head around is how to populate the Modyule.id in the first call and then the Modyule.name in the second call (which doesn't return any reference to the Modyule.id) ie:

private getModyulesDetails = (res: Response) => {  // instance method for defining function so that the this.http below refers to the class.
    let listOfModyules = res.json();
    let modyulesToReturn = [];
    let calls = [];

    for (let individualModyule of listOfModyules){
        /* I REALLY NEED TO POPULATE Modyule.id HERE */
        calls.push(
            this.http.get(individualModyuleUrl + individualModyule.id + '.json')
                .map((res: Response) => res.json()))
        );
    }

    var subject = new Subject<Modyules[]>();

    Observable.zip.apply(null, calls).subscribe((modyules: Modyules[]) => {
        var modyulesToReturn = [];
        modyules.forEach(modyule => {
            let tempModyule = new Modyule;
            /*I COULD POPULATE Modyle.name HERE BUT IT WON'T MATCH THE Modyule.id SET ABOVE - EXCEPT BY LUCK*/
            modyulesToReturn.push(tempModyule);
        }
        subject.next(modyulesToReturn);
    });

    return subject;      
}

I think I need to make the calls one by one in the for loop but somehow wait until they have all responded before returning the modyulesToReturn?

回答1:

You could potentially use Observable.zip to handle the multiple calls that are happening at once. It would require making your getModyules call to return a Subject (of some kind) which you could then call next on once the zip is complete.

getModyules (): Observable<Modyule[]> {
    var subject = new Subject<Modyule[]>();
    this.http.get(listOfModyuleIdsUrl + '.json')
        .subscribe(result => {
            this.getModyulesDetails(result)
                .subscribe(modyules => {
                    subject.next(modyules);
                });
        }, error => subject.error(error));

    return subject;
}

And details call becomes something like (zip returns a collection of all the returns, which I believe would be Modyules[] in your case):

private getModyulesDetails = (res: Response) => {  // instance method for defining function so that the this.http below refers to the class.
    let listOfModyules = res.json();
    let modyulesToReturn = [];
    let calls = [];

    for (let individualModyule of listOfModyules){
        calls.push(
            this.http.get(individualModyuleUrl + individualModyule.id + '.json')
                .map((res: Response) => res.json()))
        );
    }

    var subject = new Subject<Modyules[]>();

    Observable.zip.apply(null, calls).subscribe((modyules: Modyules[]) => {
        var modyulesToReturn = [];
        modyules.forEach(modyule => {
            let tempModyule = new Modyule;
            //populate modyule data
            modyulesToReturn.push(tempModyule);
        }
        subject.next(modyulesToReturn);
    });

    return subject;      
}

I made these changes in the browser, so I apologize for syntax errors, but I think the general idea would solve what you're trying to do.

回答2:

If one request depends on data returned by another request (in your case you need the module.id before you can make a call to module.name or its details), you should start the dependent request in the callback of the first request.

this.modyuleService.getModyules()
 .subscribe(
      modyules => {
        // start your request(s) for the details here
      },
      error =>  ...
 );

Furthemore, you use the following construct:

getModyulesDetails() {
  // this is what you want to return
  let modyulesToReturn = [];

  for (let individualModyule of listOfModyules){
    //.. you are starting asynchronous tasks here,
    // and fill your modyulesToReturn in their callbacks
  }

  // instantly return the variable that is filled asynchrously
  return modyulesToReturn;  
}    

This cannot work the way you intend to. The http requests in the for loop are asynchronous, i.e. they get started but the current thread will not wait until they finish but proceed with the next iteration of the for loop. When you return from your method it is very likely that the callbacks in your subscription have not been called yet. When they do, you have already returned your (possibly) empty array of modyulesToReturn.

Thus, you must not return synchronously. Instead, you should offer a method that returns an Observable, too, which fires everytime when you receive another modyule detail. The components that need those details would subscribe themselves to that Observable and receive updates asynchronously, as well.

If you start using asynchronous tasks you need to push that "design" down to your components.

回答3:

Well, thanks to both @batesiiic, @Matt (I've upvoted both your answers) and a useful hour (a drop in the ocean compared to the two long evenings and the whole of a Saturday I've spent on this already!) spent watching Angular University, I have a working solution (so far anyway!). .switchMap to chain two requests together, passing the Observable from the first into the second, @batesiiic's brilliant Subject idea (and forkJoin - I couldn't get zip to work) and the realisation that I could get my Modyule.id in the dependent request by looking at the response object itself!

In modyule.component.ts in ngOnInit():

this.modyuleService.getModyules()
        .switchMap(modyules =>this.modyuleService.getModyulesDetails(modyules)) 
        .subscribe(
            modyules => {
                this.modyules = modyules  
                },
           error =>  this.errorMessage = <any>error);

and in module.service.ts:

getModyules (): Observable<Modyule[]> {
    return this.http.get(this.listOfModyuleIdsUrl)
        .map(this.initialiseModyules)
        .catch(this.handleError);            
}

private initialiseModyules(res: Response){
    let body = res.json();
    let modyulesToReturn = [];
    for (let individualModyule of listOfModyules){
        let tempModyule = new Modyule;
        tempModyule.id = individualModyule.id;
        modyulesToReturn.push(tempModyule);
        }
    }
    return modyulesToReturn;
}

getModyulesDetails (modyules:Modyule[]): Observable<Modyule[]> {
    let calls  = [];

    for (let modyule of modyules){
        calls.push(
            this.http.get(this.individualModyuleUrl + modyule.id + '.json')
            );
    }

    var subject = new Subject<Modyule[]>();   

    Observable.forkJoin(calls).subscribe((res: any) => {
        for (let response of res){
            //Note this is a really very awkward way of matching modyule with a id assigned in getModyules (above) with the correct response from forkJoin (could come back in any order), by looking at the requested url from the response object
            let foundModyule = modyules.find(modyule=> {
                let modyuleUrl = this.individualModyuleUrl + modyule.id + '.json';
                return modyuleUrl === response.url;
            });
            let bodyAsJson = JSON.parse(response._body);
            foundModyule.name = res.name;
            }
        subject.next(modyules);
    });

    return subject;
}

Hope that helps someone else. However, still can't help feeling that there ought to be some way to not return from multiple calls to get details until they have all finished, without having to resort to the free for all of .forkJoin or .zip where you no longer have a relationship with the calling modyule except by looking at the response.url...

Thanks again for all your help.

回答4:

I have had to do a similar thing, where I must make a request for an authentication token before I can make any of a variety of follow-on requests.

I wanted to expose an "ApiService.Get(url)" for this, with all of the work needed to get & use the token hidden from higher level callers, but still only perform the call to get the token once.

I ended up with this kind of thing...

export class ApiService {
  private apiToken: string;
  private baseHeaders: Headers;
  private tokenObserver: ReplaySubject<Response>;

  constructor(private http: Http) {
    this.tokenObserver = null;
    try {
      // set token if saved in local storage
      let apiData = JSON.parse(localStorage.getItem('apiData'));
      if (apiData) {
        this.apiToken = apiData.apiToken;
      }
      this.baseHeaders = new Headers({
        'Accept-encoding': 'gzip',
        'API-Key': 'deadbeef-need-some-guid-123456789012',
      });
    } catch(e) {
      console.log(e);
    }
  }

  private requestHeaders(): Headers {
    if (this.apiToken) {
      this.baseHeaders.set('Authorization', 'Bearer ' + this.apiToken);
    }
    return this.baseHeaders;
  }

  private getToken(): Observable<Response> {
    if (this.tokenObserver == null) {
      this.tokenObserver = new ReplaySubject<Response>();
      this.http.get('/newtoken', this.requestHeaders())
        .subscribe(
          this.tokenObserver.next,
          this.tokenObserver.error,
          this.tokenObserver.complete,
        );
    }
    return this.tokenObserver.asObservable();
  }

  private isApiTokenValid():boolean {
    console.log("Pretending to check token for validity...");
    return !!this.apiToken;
  }

  // Get is our publicly callable API service point
  //
  // We don't accept any options or headers because all
  // interface to our API calls (other than auth) is through
  // the URL.    
  public Get(url: string): Observable<Response> {
    let observer: Subject<Response> = new Subject<Response>();
    try {
      if (this.authTokenValid()) {
        this.getRealURL(observer, url);
      } else {
        this.GetToken().subscribe(
          (v) => this.getRealURL(observer, url),
          observer.error,
          // ignore complete phase
        );
      }
    } catch(e) {
      console.log("Get: error: " + e);
    }
    return observer.asObservable();
  }

  private getRealURL(observer: Subject<Response>, url: string): void {
    try {
      this.http.get(url, {headers: this.requestHeaders()})
        .subscribe(
          observer.next,
          observer.error,
          observer.complete,
        );
    } catch(e) {
      console.log("GetRealURL: error: " + e);
    }
  }
}

With this in place, calls to my API come down to:

this.api.Get('/api/someurl')
  .subscribe(this.handleSomeResponse)
  .catch(this.handleSomeError);