我正在尝试制作我认为的 pausable buffer
我有人为此分享了他们的代码,但我不知道如何将其变成自定义操作(没有 typescript /只有 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)。
最佳答案
您可以像这样将其添加到原型(prototype)中:
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 - 如何使用 rxjs 5 创建一个 pausableBuffer,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/40557715/