c# - 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??

score 2 · Accepted Answer

Rx の標準機能は、シーケンスが完了したときにサブスクリプションを破棄することです (サブスクリプションの値がまだ別のシーケンスを介してパイプ処理されている場合でも)。

そのことを念頭に置いてDelay、ソースシーケンスから放出される値の速度を制御することはできず、それ自体のオブザーバーに対して値を遅らせることしかできません。

score 0 · Accepted Answer

推測する必要がある場合、Delay は元のオブザーバブルからのアイテムをキューに入れ、指定された遅延に基づいて適切と思われるアイテムをディスパッチします。したがって、元のオブザーバブルが長い間破棄されていても、Delay メソッドによって作成されたオブザーバブルはまだ生きていて、キックしています。あなたが観察している動作は、この説明にうまく適合しています。

score -1 · Accepted Answer

ThreadPoolがない場合、動作は同じです。

01廃棄廃棄...0... ... 1 ... ...... 0 ...... ......1...。

c# - Observable.Delay calling Dispose before OnNext is fired

3 に答える 3

Related

Reference