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

发布于12月28日

假设你有一个IObservable<uint>类型的可观测震源.我在找一个变换函数

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

它将source转换为以下形式:

当source产生一个值时，产生的可观测对象应该立即发布相同的值
假设最后发布的元素是N.如果source中没有任何新元素，则Result Observable应该在N个刻度之后发布值N-1；然后在N-1个刻度之后发布值N-2；依此类推，直到发布0
当source发布新值时，这种类似计时器的行为将根据新值进行重置

示例输入/输出:

Tick time	Source observable	Result observable
0	10	10
10		9
19		8
21	2	2
23		1
24		0
100	1	1
101		0
...	...	...

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

推荐答案

假设你的"滴答"是以十分之一秒为单位--这可能会满足你的需要:

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

    Countdown(subject).TimeInterval().Subscribe(x => Console.WriteLine(x));

    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-1,
        condition: i => i >= 0,
        iterate: i => i-1,
        resultSelector: i => i,
        timeSelector: _ => TimeSpan.FromMilliseconds(value*100)
        ).StartWith(value)).Switch();
}