指定大小的pauseableBuffer

Question

如何创建一个pausableBuffered可观察对象，它只缓冲最新的项目?

目前我有一个丑陋的解决方案:

source.pauseableBuffered(pauser).debounce(0)

Answer 1

可以有两个可观察序列1. source序列2. pauser序列IObservable<bool>

那么你可以把它们合并起来

Observable.CombineLatest(
    source,
    pauser,
    (s,p)=>Tuple.Create(s,p))
.Where(t=>!t.Item2)

这现在允许你忽略值，而pauser已经推了一个'true'值。

然后你可以简单地把replay(1)放在它的末尾。

Observable.CombineLatest(
    source,
    pauser,
    (s,p)=>Tuple.Create(s,p))
.Where(t=>!t.Item2)
.Replay(1)
//.Publish().RefCount(); //If required.

编辑Replay(1)只是噪声，并不关注op。

这段代码

var source = new Subject<int>();
var pauser = new Subject<bool>();
var query = Observable.CombineLatest(source, pauser, (s,p)=>Tuple.Create(s,p))
    .Where(t=>!t.Item2);
query.Dump();
pauser.OnNext(false);
source.OnNext(1);
source.OnNext(2);
source.OnNext(3);
pauser.OnNext(true);
source.OnNext(4);
source.OnNext(5);
source.OnNext(6);
pauser.OnNext(false);
source.OnNext(7);
source.OnNext(8);

产生如下结果

1 
2 
3 
6 
7 
8