这是我在RX论坛上获得的一些帮助:
这个想法是发行一系列“票”以激发原始顺序。这些“票证”因超时而被延迟,但不包括第一个票证,该票证将立即添加到票证序列中。当事件进入且有票证等待时,该事件会立即触发,否则会等到票证再触发。当它触发时,发出下一张票,依此类推…
为了将票证和原始事件结合起来,我们需要一个组合器。不幸的是,此处无法使用“标准” .CombineLatest,因为它会在以前使用的票证和事件上触发。因此,我必须创建自己的组合器,该组合器基本上是经过过滤的.CombineLatest,它仅在组合中的两个元素均为“新鲜”时才触发- 之前从未返回过。我称之为.CombineVeryLatest又名.BrokenZip;)
使用.CombineVeryLatest,可以将上述想法实现为:
public static IObservable<T> SampleResponsive<T>(
this IObservable<T> source, TimeSpan delay)
{
return source.Publish(src =>
{
var fire = new Subject<T>();
var whenCanFire = fire
.Select(u => new Unit())
.Delay(delay)
.StartWith(new Unit());
var subscription = src
.CombineVeryLatest(whenCanFire, (x, flag) => x)
.Subscribe(fire);
return fire.Finally(subscription.Dispose);
});
}
public static IObservable<TResult> CombineVeryLatest
<TLeft, TRight, TResult>(this IObservable<TLeft> leftSource,
IObservable<TRight> rightSource, Func<TLeft, TRight, TResult> selector)
{
var ls = leftSource.Select(x => new Used<TLeft>(x));
var rs = rightSource.Select(x => new Used<TRight>(x));
var cmb = ls.CombineLatest(rs, (x, y) => new { x, y });
var fltCmb = cmb
.Where(a => !(a.x.IsUsed || a.y.IsUsed))
.Do(a => { a.x.IsUsed = true; a.y.IsUsed = true; });
return fltCmb.Select(a => selector(a.x.Value, a.y.Value));
}
private class Used<T>
{
internal T Value { get; private set; }
internal bool IsUsed { get; set; }
internal Used(T value)
{
Value = value;
}
}
编辑:这是AndreasKöpf在论坛上提出的CombineVeryLatest的另一个更紧凑的变体:
public static IObservable<TResult> CombineVeryLatest
<TLeft, TRight, TResult>(this IObservable<TLeft> leftSource,
IObservable<TRight> rightSource, Func<TLeft, TRight, TResult> selector)
{
return Observable.Defer(() =>
{
int l = -1, r = -1;
return Observable.CombineLatest(
leftSource.Select(Tuple.Create<TLeft, int>),
rightSource.Select(Tuple.Create<TRight, int>),
(x, y) => new { x, y })
.Where(t => t.x.Item2 != l && t.y.Item2 != r)
.Do(t => { l = t.x.Item2; r = t.y.Item2; })
.Select(t => selector(t.x.Item1, t.y.Item1));
});
}