c# - Observable.Delay 在触发 OnNext 之前调用 Dispose

Question

我无法理解 Observable.Delay 的工作原理以及何时调用 Dispose()。熟悉 Rx 的人可以帮忙吗？

以下代码片段:

    static void Main(string[] args)
    {
        var oneNumberEveryFiveSeconds = new SomeObservable();
        // Instant echo
        oneNumberEveryFiveSeconds.SubscribeOn(Scheduler.ThreadPool).Subscribe(num => Console.WriteLine(num));
        // One second delay
        oneNumberEveryFiveSeconds.Delay(TimeSpan.FromSeconds(1)).SubscribeOn(Scheduler.ThreadPool).Subscribe(num => Console.WriteLine("...{0}...", num));
        // Two second delay
        oneNumberEveryFiveSeconds.Delay(TimeSpan.FromSeconds(2)).SubscribeOn(Scheduler.ThreadPool).Subscribe(num => Console.WriteLine("......{0}......", num));

        Console.ReadKey();
    }

    public class SomeObservable : IObservable<int>
    {
        public IDisposable Subscribe(IObserver<int> o)
        {
            for (var i = 0; i < 2; i++)
            {
                o.OnNext(i);
            }
            o.OnCompleted();

            return new DisposableAction(() => { Console.WriteLine("DISPOSED"); });
        }
    }

    public class DisposableAction : IDisposable
    {
        public DisposableAction(Action dispose)
        {
            this.dispose = dispose;
        }

        readonly Action dispose;

        public void Dispose()
        {
            dispose();
        }
    }

产生以下结果:

0
1
DISPOSED
DISPOSED
DISPOSED
...0...
...1...
......0......
......1......

我希望它更像:

0
1
DISPOSED
...0...
...1...
DISPOSED
......0......
......1......
DISPOSED

有什么想法吗？？

Answer 1

Rx 的标准功能是在序列完成时处理订阅(即使它的值仍在通过另一个序列传输)。

考虑到这一点，Delay 无法控制从源序列发出值的速度，它只能将值延迟到它自己的观察者。