我如何创建一个带有 rxjs 5 的 pausableBuffer

Question

我正在尝试制作我认为的 pausable buffer

有人为此分享了他们的代码，但我不知道如何将其变成自定义操作（没有打字稿/只有 ES6。

const attach = Rx.Observable.timer(0 * 1000, 8 * 1000).mapTo('@');
const detach = Rx.Observable.timer(4 * 1000, 8 * 1000).mapTo('#');

const input = Rx.Observable.interval(1* 1000);
const pauser = attach.mapTo(true).merge(detach.mapTo(false));

input
  .publish(_input => _input
    .combineLatest(pauser, (v, b) => b)
    .filter(e => e)
    .publish(_switch => _input.bufferWhen(() => _switch.take(1)))
  )
  .flatMap(e => Rx.Observable.from(e))
  .concatMap(e => Rx.Observable.empty().delay(150).startWith(e))

有人可以帮我创建那个，这样我就可以做 input.pausableBuffer(pauser)（也许可以定义一个 startsWith）。

Answer 1

您可以像这样将其添加到原型中：

var pausableBuffer = function(pauser) {
  return this.publish(_input => _input
    .combineLatest(pauser, (v, b) => b)
    .filter(e => e)
    .publish(_switch => _input.bufferWhen(() => _switch.take(1)))
  )
  .flatMap(e => Rx.Observable.from(e));
}

Rx.Observable.prototype.pausableBuffer = pausableBuffer;

要记住的一件事是，这将在暂停状态下启动。要改为以活动状态启动它，请将 .startWith(true) 添加到 pauser。

var pausableBuffer = function(pauser) {
  return this.publish(_input => _input
    .combineLatest(pauser.startWith(true), (v, b) => b)
    .filter(e => e)
    .publish(_switch => _input.bufferWhen(() => _switch.take(1)))
  )
  .flatMap(e => Rx.Observable.from(e));
}

Rx.Observable.prototype.pausableBuffer = pausableBuffer;

2019 年更新：RxJs 6 风格：

var pausableBuffer = function(pauser) {
  return (source) => source.pipe(publish(_input => 
  combineLatest(_input, pauser.pipe(startWith(true))).pipe(
    map(([inp, pa]) => pa),
    filter(pa => pa),
    publish(_switch => _input.pipe(bufferWhen(() => _switch.pipe(take(1)))))
  )),
    mergeMap(e => from(e))
  );
}

Demo

我如何创建一个带有 rxjs 5 的 pausableBuffer

How can I create a pausableBuffer w/ rxjs 5

rxjs

rxjs5