c# - 如何将缓慢移动的可观测值与快速移动的可观测值的最新值结合起来

Question

一个 friend 问我这个问题 - 我认为这是一个很好的问题，所以我在这里重新发布它和我的答案:

我有这两个流:

var slowSource = Observable.Interval(TimeSpan.FromSeconds(1));
var fastSource = Observable.Interval(TimeSpan.FromMilliseconds(100));

我想将它们组合起来，以便生成包含以下内容的输出对 -来自slowSource的下一个值 - fastSource 的最新值

我只想要来自 SlowSource 的每个值一个输出对。例如，前三个输出值可能如下所示:

0,8
1,18,
2,28

连接让我很接近，但最终每个 SlowSource 都会有多个输出(我猜是由于持续时间重叠的方式):

var qry = slowSource.Join(
          right: fastSource,
          leftDurationSelector: i => fastSource,
          rightDurationSelector: j => fastSource,
          resultSelector: (l, r) => {return new {L = l, R = r};})

.Subscribe(Console.WriteLine);

使用 GroupJoin 和 Select 会产生看起来正确的输出:

var qry2 = slowSource.GroupJoin(
              right: fastSource,
              leftDurationSelector: i => fastSource,
              rightDurationSelector: j => fastSource,
              resultSelector: (l, r) => {return new {L= l, R = r};}
              )
          .Select(async item => {
            return new {L = item.L, R = await item.R.FirstAsync()};})
          .Subscribe(Console.WriteLine);

但是，这感觉并不是一个很好的方法；必须有一种更好的方法，使用其他组合器以更简单的方式完成类似的事情。有吗？

Answer 1

将 IObservable 与 IEnumerable 组合在一起的 Zip 重载怎么样？它使用 MostRecent() 来获取流的最新值的样本作为可枚举值。

slowSource.Zip(fastSource.MostRecent(0), (l,r) => new {l,r})