Angular 2 / RXJS - 需要一些帮助来处理请求

我一直在阅读 rxjs 文档，但迷失在所有操作符中。

这就是我到目前为止所得到的

  let obs = Observable.from([1, 3, 5])   

所以我需要做的是take()数组中的一些设定数量。在发布请求中使用结果，当成功时，我需要重新启动该过程。我想收集所有结果，并随着过程的进行而保持进度（对于进度条）

我不需要所有这些的代码。我真正需要知道的是如何使用 rxjs 来分割这个数组..发送其中的一部分，然后重新启动该过程，直到没有什么可以发送。

最终解决方案

  var _this = this

  function productsRequest(arr) {
    return _this.chainableRequest('post', `reports/${clientId}/${retailerId}/`, loadedProductsReport, {
        'identifiers': arr,
        'realTime': true
      })    
  }

  let arrayCount = Math.ceil(identifiers.length/10)
  let obs = Observable.from(identifiers)            
    .bufferCount(10)
    .concatMap(arr => {
      arrayCount--
      return arrayCount > 0 ? productsRequest(arr) : Observable.empty()
    })


  let subscriber = obs.subscribe(
    value => console.log(value)
  )

父级中的可链接请求方法

  chainableRequest(method: string, endpoint: string, action: Function, data = {}, callback?: Function){
let body = (<any>Object).assign({}, {
  headers: this.headers
}, data)


return this._http[method.toLowerCase()](`${this.baseUri}/${endpoint}`, body, body)
          .map((res: Response) => res.json())
  }

这很大程度上取决于您想要实现的目标。

如果你想基于之前的某个 Observable 递归调用一个 Observable 并且你不知道要调用它多少次，那么使用expand() http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-expand操作员。

例如，此演示根据先前调用的响应递归创建 5 个请求（count财产）：

import { Observable } from 'rxjs/Observable';

function mockPostRequest(count) {
    return Observable.of(`{"count":${count},"data":"response"}`)
        .map(val => JSON.parse(val));
}

Observable.of({count: 0})
    .expand(response => {
        console.log('Response:', response.count);
        return response.count < 5 ? mockPostRequest(response.count + 1) : Observable.empty();
    })
    .subscribe(undefined, undefined, val => console.log('Completed'));

打印到控制台：

Response: 0
Response: 1
Response: 2
Response: 3
Response: 4
Response: 5
Completed

观看现场演示：http://plnkr.co/edit/lKNdR8oeOuB2mrnR3ahQ?p=preview http://plnkr.co/edit/lKNdR8oeOuB2mrnR3ahQ?p=preview

或者如果你只是想依次调用一堆 HTTP 请求（concatMap() http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-concatMap操作员）或立即调用所有这些并在它们到达时使用它们（mergeMap() http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-mergeMap操作员）：

Observable.from([
    'https://httpbin.org/get?1',
    'https://httpbin.org/get?2',
    'https://httpbin.org/get?3',
  ])
  .concatMap(url => Observable.of(url))
  .subscribe(response => console.log(response));

打印到控制台：

https://httpbin.org/get?1
https://httpbin.org/get?2
https://httpbin.org/get?3

观看现场演示：http://plnkr.co/edit/JwZ3rtkiSNB1cwX5gCA5?p=preview http://plnkr.co/edit/JwZ3rtkiSNB1cwX5gCA5?p=preview