buffer

缓存操作符有几种形式：

• bufferCount(countNum)： 缓存多少数量之后emit,但是如果流结束，没有到达缓存数量也会emit, 以数组的形式返回
• bufferTime(timer): 多长时间关闭一次缓存，单位为ms，例如 bufferTime(900), 表示900ms关闭一次buffer
• buffer(observable): 根据提供的observable来缓存
• bufferToggle: 暂时不讲
• bufferWhen: 暂时不讲

bufferCount

示例

var source = Rx.Observable.of('a', 'b', 'c', 'd', 'e')
  .zip(Rx.Observable.interval(1000).take(7))
  .bufferCount(2)
  .subscribe(console.log)

# marble图
/*
(abcde|) 
----0----1----2----3----4----5----6|
  zip
----a----b----c----d----e|
  bufferCount(2)
---------(['a', 'b'])----------(['c', 'd'])----(['e']|)
----此处'a'被缓存----此时a,b缓存数达到2个则emit---
打印结果
['a', 'b']
['c', 'd']
['e']

bufferTime

示例

var foo = Rx.Observable.of('a', 'b', 'c', 'd', 'e')
var bar = Rx.Observable.interval(1000).take(6);
        
/*
  (abcde|)                                                                  (foo)
  ---------0---------1---------2---------3---------4---------5|             (bar)
    zip
  ---------a---------b---------c---------d---------e|
    bufferTime(900)   每900ms关闭一个buffer
  -----------------a--------b--------c--------d--------e  
  --------!--------!--------!--------!--------!--------!     // 其中！的位置表示关闭buffer的时间点
*/

var result = foo.zip(bar, (x, _) => x).bufferTime(900).subscribe(console.log);

打印结果
[]
['a']
['b']
['c']
['d']
['e']

buffer

var bar = Rx.Observable.interval(600).take(6);

var goo = Rx.Observable.interval(900).take(3)
        
/*
  (abcde|)       (foo)
  ----0----1----2----3----4----5|             (bar)
    zip
  ----a----b----c----d----e|
    buffer(goo)
  --------0--------1--------2|  
  --------!--------a--------b--------c--------d----e|  
*/

var result = foo.zip(bar, (x, _) => x).buffer(goo);
result.subscribe(console.log)

// 打印结果 注意由于时间不精确可能导致结果的不准确性
['a']
['b', 'c']
['d', 'e']