Angular2: Dynamic synchronous http requests

问题:

Goal: To make a series of synchronous http requests and be able to subscribe to them as one observable stream.

Sample (Not Working):

let query_arr = ['test1','test2','test3']

function make_request(query_arr){

    if (query_arr.length){

        let payload = JSON.stringify(query_arr[0]);
        let headers = new Headers();

        query_arr.splice(0,1);

        this.http.post('https://endpoint/post',payload,{headers:headers})
            .map((res:Response) => {make_request(query_arr)})

    }

}.subscribe(
    data => console.log('finished http request, moving on to next http request'),
    err => console.error(err),
    () => console.log('all http requests have been finished')
);

make_request(query_arr)

Goal Functionality:

• Need to know when each response was returned
• Must know when all responses have returned

回答1:

You need to leverage the flatMap operator to execute your requests in series (one after one). For this, you need to build your data processing chain recursively. The point here is to call the operator on the previous observable (the one returned by the previous request).

This way the request will wait for the previous one to be complete before executed itself. The callback provided when subscribing will be called when all requests were executed.

Here is a sample implementation of this approach:

makeRequest(queryArr, previousObservable){
  if (queryArr.length) {
    let payload = JSON.stringify(queryArr[0]);
    let headers = new Headers();
    (...)

    queryArr.splice(0,1);

    var observable = null;
    if (previousObservable) {
      observable = previousObservable.flatMap(() => {
        return this.http.post('https://testsoapi.apispark.net/v1/entities', payload,{
            headers:headers
          })
          .map((res:Response) => res.json())
          .do(() => {
            console.log('request finished');
          });
      });
    } else {
      observable = this.http.post('https://testsoapi.apispark.net/v1/entities', payload, {
        headers:headers
      })
        .map((res:Response) => res.json())
        .do(() => {
          console.log('request finished');
        });
    }

    return this.makeRequest(queryArr, observable);
  } else {
    return previousObservable;
  }
}

This method can be called initially like this:

test() {
  let queryArr = [
    { val: 'test1' },
    { val: 'test2' },
    { val: 'test3' }
  ];

  this.makeRequest(queryArr).subscribe(
    () => {
      console.log('all requests finished');
    });
}

See this plunkr: https://plnkr.co/edit/adtWwckvhwXJgPDgCurQ?p=preview.

回答2:

There were a couple syntactical errors in your code as well that would need to be addressed. But those aside you can simplify greatly by using concatMap + defer instead.

let query_arr = ['test1','test2','test3'];
let self = this;

Rx.Observable.from(query_arr).map(JSON.stringify)
  .concatMap(payload => {
    let headers = new Headers();
    return Rx.Observable.defer(() => {
      self.http.post('https://endpoint/post',payload,{headers:headers});
    });
  }, resp => resp.json())
  .subscribe(
    data => console.log('finished http request, moving on to next http request'),
    err => console.error(err),
    () => console.log('all http requests have been finished')
  );

The basic idea of this is that it will convert the query array into an Observable then it will eagerly create a series of lazy requests that will only be executed when they are subscribed to. However, by wrapping the post in a defer each request will only get dispatched when the previous one completes.

回答3:

Or a non recursive version in typescript where you give an array to forkjoin

in the return observableObj(res.json()) you know each response when it returns from the httpcall

in the subscribe you know when all responses returned and an array of values

const observableObj = (obj) => Observable.of(obj)

class Requests {

 private query_arr = ['test1','test2','test3']
 private url = 'https://testsoapi.apispark.net/v1/entities'

 public make() {
   this.processHttp().subscribe(
        (d) => {
          console.log(d)              
        },
        (e) => {
          console.log(e)
        },
        () => {
          console.log("http calls are done")
        })

 }

 private httpCall(options : RequestOptions) : Observable<Response> {
   let username : string = 'xxx'
   let password : string = 'yyy'
   let headers = new Headers()
   headers.append("Authorization", "Basic " + btoa(username + ":" + password))
   headers.append("Content-Type", "application/x-www-form-urlencoded")
   options.headers = headers
   return this.http.get(this.url,options)
 }

 private createRequestOptions(option1 : string) {
   let data = {'option1':option1}
   let params = new URLSearchParams()
   for(var key in data) {
     params.set(key, data[key])
   }
   let options = new RequestOptions({
     search: params
   })
   return options
 }

 private processHttp() {
    return Observable.forkJoin(
        this.query_arr.map(option => {
            return this.httpCall(createRequestOption(option)).flatMap((res: Response) => {
                return observableObj(res.json())
            })
        }))            
 }
}