system.reactive - 组合最新，但只向左推

Question

我需要实现一个版本的CombineLatest (我将其称为 WithLatest 此处)为左侧的每个项目和右侧的最新项目调用选择器。它不应该只插入右侧的项目更改。

我想这是否是内置的Observable.Create或者现有扩展的组合不是特别重要；无论哪种方式，我都将使其成为“盒装”扩展方法。

示例

var left = new Subject<int>();
var right = new Subject<int>();

left.WithLatest(right, (l,r) => l + " " + r).Dump();

left.OnNext(1);   // <1>
left.OnNext(2);   // <2>
right.OnNext(1);  // <3>
right.OnNext(2);  // <4>
left.OnNext(3);   // <5>

应该屈服

2 1
3 2

编辑 :我的例子的逻辑是:

左边变成了 1。右边是空的，没有插入任何值。

Left 更新为 2(它忘记了以前的值)。右边仍然是空的，所以什么都没有推。

Right 填充为 1，因此 Left = 2(最新值)，Right = 1 被推送。到目前为止，WithLatest 之间没有区别。和 CombineLatest

权利被更新——没有任何东西被推送。这有什么不同

Left 更新为 3，所以 Left = 3, Right = 2(最新值)被推送。

有人建议我尝试:

var lr = right.ObserveOn(Scheduler.TaskPool).Latest();
left.Select(l => l + " " + lr.First()).Dump();

但这会阻塞当前线程以进行我的测试。

Answer 1

您可以使用现有运算符执行此操作。

Func<int, int, string> selector = (l, r) => l + " " + r;

var query = right.Publish(rs => left.Zip(rs.MostRecent(0), selector).SkipUntil(rs));

Publish确保我们只订阅 right一次并在所有订阅者之间共享订阅到rs .

MostRecent变成 IObservable<T>成IEnumerable<T>总是从源 observable 产生最近发出的值。

Zip之间IObservable<T>和 IEnumerable<U>每次 observable 发出一个值时都会发出一个值。

SkipUntil跳过对 (l, r)发生在 right 之前曾经发出一个值。