Angular 2/RXJS-需要一些帮助处理批处理请求 [英] Angular 2 / RXJS - need some help batching requests

问题描述

我一直在阅读rxjs文档，但是在所有运算符中迷路了.

I keep reading rxjs documentation but getting lost in all the operators..

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

this is what i got so far

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

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

so what i need this to do is take() some set amount from the array. use the results in a post request, when that comes out successful then i need to restart the process. I want to collect all the results, and keep progress as the process is going (for a progress bar)

我不需要所有这些代码.我真正需要知道的是如何使用rxjs将该数组拆分..发送它的一部分，然后重新开始该过程，直到没有剩余要发送的内容.

I don't need the code for all of that. what i really need to know is how to use rxjs to split this array up.. send part of it, and restart the process until theres nothing left to send.

最终解决方案

  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)
  )

父级中的

可链接请求方法

chainable request method in parent

  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())
  }

推荐答案

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

This largely depends on what you're trying to achieve.

如果要基于以前的某些Observable递归调用Observable，并且您不知道要调用多少次，请使用

If you want to recursively call an Observable based on some previous Observable and you don't know how many times you're going to call it then use expand() operator.

例如，此演示基于先前调用的响应(count属性)递归创建5个请求:

For example this demo recursively creates 5 requests based on the response from the previous call (count property):

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请求(

Or if you just want to call a bunch of HTTP request in order one after another (concatMap() operator) or call all of them at once and consume them as they arrive (mergeMap() operator):

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

这篇关于Angular 2/RXJS-需要一些帮助处理批处理请求的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！