RxJs：条件为真时缓冲事件，条件为假时传递事件

Question

我创建了下面的 Observable 构造函数，它的工作原理与描述的一样。有谁知道使用 RxJs 附带的运算符是否有更简洁的方法来实现相同的行为？我正在查看接近所需行为的bufferToggle，但我需要在缓冲区关闭时传递发出的值。

函数说明：缓冲器所发射的source值，如果condition发射true，并穿过出射source如果值condition发射false。如果条件在 befalse之后发出true，则缓冲区按照接收到的顺序释放每个值。缓冲区被初始化为传递发出的source值，直到condition发出true.

function bufferIf<T>(condition: Observable<boolean>, source: Observable<T>): Observable<T> {
  return new Observable<T>(subscriber => {
    const subscriptions: Subscription[] = [];
    const buffer = [];
    let isBufferOpen = false;

    subscriptions.push(
      // handle source events
      source.subscribe(value => {
        // if buffer is open, or closed but buffer is still being 
        // emptied from previously being closed.
        if (isBufferOpen || (!isBufferOpen && buffer.length > 0)) {
          buffer.push(value);

        } else {
          subscriber.next(value);
        }
      }),

      // handle condition events
      condition.do(value => isBufferOpen = value)
        .filter(value => !value)
        .subscribe(value => {
          while (buffer.length > 0 && !isBufferOpen) {
            subscriber.next(buffer.shift());
          }
        })
    );

    // on unsubscribe
    return () => {
      subscriptions.forEach(sub => sub.unsubscribe());
    };
  });
}

编辑

作为对评论的回应，以下是与上面相同的功能，但采用 RxJs 运算符的形式，并更新为使用 RxJx 6+ pipeabale 运算符：

 function bufferIf<T>(condition: Observable<boolean>): MonoTypeOperatorFunction<T> {
   return (source: Observable<T>) => {
     return new Observable<T>(subscriber => {
       const subscriptions: Subscription[] = [];
       const buffer: T[] = [];
       let isBufferOpen = false;
   
       subscriptions.push(
         // handle source events
         source.subscribe(value => {
           // if buffer is open, or closed but buffer is still being 
           // emptied from previously being closed.
           if (isBufferOpen || (!isBufferOpen && buffer.length > 0)) {
             buffer.push(value);
           } else {
             subscriber.next(value);
           }
         }),
   
         // handle condition events
         condition.pipe(
          tap(con => isBufferOpen = con),
          filter(() => !isBufferOpen)
         ).subscribe(() => {
          while (buffer.length > 0 && !isBufferOpen) {
            subscriber.next(buffer.shift());
          }
        })
       );
   
       // on unsubscribe
       return () => subscriptions.forEach(sub => sub.unsubscribe());
     });
   }
}

Answer 1

我找到了一个基于运算符而不是订阅的解决方案，但犹豫是否称其为更简洁。

请注意，如果可以保证缓冲区开/关流始终以关闭结束（即发出奇数次），则可以删除 endToken。

console.clear() 
const Observable = Rx.Observable

// Source and buffering observables
const source$ = Observable.timer(0, 200).take(15)
const bufferIt$ = Observable.timer(0, 500).map(x => x % 2 !== 0).take(6)  

// Function to switch buffering
const endToken = 'end'            
const bufferScanner = { buffering: false, value: null, buffer: [] }
const bufferSwitch = (scanner, [src, buffering]) => { 
  const onBufferClose = (scanner.buffering && !buffering) || (src === endToken)
  const buffer = (buffering || onBufferClose) ? scanner.buffer.concat(src) : []
  const value = onBufferClose ? buffer : buffering ? null : [src]
  return { buffering, value, buffer }
}
      
// Operator chain
const output = 
  source$
    .concat(Observable.of(endToken))     // signal last buffer to emit
    .withLatestFrom(bufferIt$)           // add buffering flag to stream
    .scan(bufferSwitch, bufferScanner)   // turn buffering on and off
    .map(x => x.value)                   // deconsruct bufferScanner
    .filter(x => x)                      // ignore null values
    .mergeAll()                          // deconstruct buffer array
    .filter(x => x !== endToken)         // ignore endToken

// Proof
const start = new Date()
const outputDisplay = output.timestamp()
  .map(x => 'value: ' + x.value + ', elapsed: ' + (x.timestamp - start) )
const bufferDisplay = bufferIt$.timestamp()
  .map(x => (x.value ? 'buffer on' : 'buffer off') + ', elapsed: ' + (x.timestamp - start) )
bufferDisplay.merge(outputDisplay)
  .subscribe(console.log)
  

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.2/Rx.js"></script>

脚注

我还找到了一个基于 的解决方案buffer()，但我不相信它在高频源下是否稳定。某些缓冲区配置似乎有些做作（即声明看起来合理，但测试显示偶尔会出现延迟，从而干扰缓冲区操作）。

无论如何，供参考，

/* 
  Alternate with buffered and unbuffered streams
*/

const buffered = 
   source$.withLatestFrom(bufferIt$)      
    .filter(([x, bufferIsOn]) => bufferIsOn)  
    .map(x => x[0])
    .buffer(bufferIt$.filter(x => !x))
    .filter(x => x.length)       // filter out empty buffers
    .mergeAll()                  // unwind the buffer

const unbuffered =
  source$.withLatestFrom(bufferIt$)      
    .filter(([x, bufferIsOn]) => !bufferIsOn)    
    .map(x => x[0])

const output = buffered.merge(unbuffered)