可观察的：在源完成之前获取间隔的最新值答案

【问题标题】：Observable: Getting latest value in intervals until source finishes可观察的：在源完成之前获取间隔的最新值
【发布时间】：2021-03-19 15:12:28
【问题描述】：

我正在寻找一个具有类似签名的可观察选择器：

static IObservable<T> TakeLatest(this IObservable<T> input, TimeSpan interval)

应该：

输入发出第一项后立即发出第一项
从那时起，在之后的固定时间间隔内，发出输入生成的最新项目
只要输入完成（或失败）就完成（或失败）

就弹珠而言，类似于以下内容 - 假设间隔 = 2 个时间单位：

Time	1	2	3	4	5	6	7	8	9	10	11	12	13	14	15
Input		A		B	C	D			E				F (complete)
Output		A		B		D		D		E		E	complete (F not emitted anymore)

是否有任何开箱即用的方法，或者一个相当简单的选择器来产生这些结果？

【问题讨论】：

我想你想要debounce
@Liam 我认为这还不够（如果错了，请纠正我）。例如。如果 source 产生一个值，我认为 debounce 最多会发出这个项目一次；在我的场景中，可能需要多次发出同一个项目（如果没有其他更新的项目可用）
我想你想把它和repeat 结合起来。问题是我只知道 rxjs 而不是 rx.Net 所以我不能 100% 确定运营商是如何映射的。因此只有一条评论
您可能正在搜索Sample 运算符。如果您还希望能够动态更改采样间隔，请查看this 问题。

标签： c# system.reactive

【解决方案1】：

这应该完全符合您的要求。不过我还没有测试过。

/// <summary>Samples the source observable sequence at each interval,
/// allowing repeated emissions of the same element.</summary>
public static IObservable<T> SampleWithDuplicates<T>(this IObservable<T> source,
    TimeSpan interval, IScheduler scheduler = null)
{
    scheduler ??= DefaultScheduler.Instance;
    return source.Publish(published => Observable
        .Interval(interval, scheduler)
        .WithLatestFrom(published, (_, x) => x)
        .Merge(published.FirstAsync())
        .TakeUntil(published.LastOrDefaultAsync()));
}

【讨论】：

确实很好用！（它不会发出第一个项目，但这当然很容易添加）。谢谢
您能否发布正确答案，因为没有一个建议的解决方案不能按 OP 的预期工作？
@DmitryStepanov 好的，完成。但它仍然未经测试。
@DmitryStepanov 显然 OP 的弹珠图具有竞争条件，因为间隔滴答声与源序列发出项目的时间大致相同。所以不能保证输出总是 A B D D E E。
@TheodorZoulias - 一个非常好的查询！

【解决方案2】：

我现在已经完成了以下操作 - 我认为它有效，但我会打开它以防万一有人能想到更优雅的方式（或者能想到我当前的问题实施）

static IObservable<T> TakeLatest(this IObservable<T> input, TimeSpan interval, IScheduler scheduler) => input
  .FirstAsync()
  .Select(_ => Observable.Interval(interval, scheduler).StartWith(0))
  .Switch()
  .CombineLatest(input, (a,b) => (a,b))
  .DistinctUntilChanged(x => x.a)
  .Select(x => x.b)
  .TakeUntil(input.LastAsync());

【讨论】：

一个问题是input 序列被订阅了三次，所以它不能与冷序列一起正常工作。您可能需要Publish input 序列。