RX 操作员使用先前值进行扫描，信号源自值本身

Question

假设您有一个类型为

IObservable<uint>

的可观察源。我正在寻找一个转换函数

IObservable<uint> Countdown(IObservable<uint> source)
{
  // how? Something like ...
  return source.Select(value => ProduceInnerObservable(value)).Switch()
}

将

source

变换如下：

• 当

  source

  产生一个值时，生成的可观察值应立即发布相同的值
• 假设最后发布的元素是

  N

  。在

  source

  中没有任何新元素的情况下，可观察结果应在

  N-1

  刻度后发布值

  N

  ；然后

  N-2

  在

  N-1

  刻度之后；以此类推，直到发布0个
• 当

  source

  发布新值时，这种类似计时器的行为会根据新值重置

输入/输出示例：

滴答时间	可观察来源	结果可观察
0	10	10
10		9
19		8
21	2	2
23		1
24		0
100	1	1
101		0
...	...	...

最优雅的实现方式是什么？

Answer 1

假设您的“滴答声”是秒 - 也许这会满足您的需求：

static void Main()
{
    var subject = new Subject<int>();

    Countdown(subject).Subscribe(Console.WriteLine);

    subject.OnNext(10);

    Thread.Sleep(4000);

    subject.OnNext(2);

    Thread.Sleep(4000);

    subject.OnNext(1);

    Console.ReadLine();

}

static IObservable<int> Countdown(IObservable<int> source)
{
    return source.Select(value => Observable.Generate(
        initialState: value,
        condition: i => i >= 0,
        iterate: i => i-1,
        resultSelector: i => i,
        timeSelector: _ => TimeSpan.FromSeconds(1)
        )).Switch();
}