Observable.Delay calling Dispose before OnNext is fired

Question

I am having problem understanding how Observable.Delay works and when the Dispose() is meant to be called. Would anyone familiar with Rx be able to help please?

The following code snippet:

    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();
        }
    }

produces the below result:

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

I was expecting it to be more like:

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

Any idea??

Answer 1

Without the ThreadPool the behaviour is identical:

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

Answer 2

If I had to guess I'd say that Delay queues up the items coming from the original observable and then dispatches them as it sees fit based on the specified delay. Thus even though the original observable has long been disposed the observable created by Delay method is still alive and kicking. The behavior you are observing fits this explanation nicely.

Answer 3

The standard functionality of Rx is to dispose a subscription when the sequence completes (even if its values are still being piped through another sequence).

With that in mind, Delay cannot control the speed of values being emitted from the source sequence, it can only delay the values to its own observers.