Demonstrate leakiness of Generate and recursive Schedule

LeakyGenerate.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Threading.Tasks;
using System.Runtime.InteropServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using JetBrains.Profiler.Windows.Api;

namespace RxMemoryLeak
{
    class Program
    {
        static void Main(string[] args)
        {
            var mre = new ManualResetEvent(false);

            if (MemoryProfiler.IsActive && MemoryProfiler.CanControlAllocations)
                MemoryProfiler.EnableAllocations();

            MemoryProfiler.Dump();

            Demo2(mre);

            mre.WaitOne();

            GC.Collect();
            GC.WaitForPendingFinalizers();
            GC.Collect();

            MemoryProfiler.Dump();
        }

        static void Demo(ManualResetEvent mre)
        {



            TimeSpan timeSpan = TimeSpan.FromSeconds(1);
            IObservable<int> obs = Observable.Generate(initialState: 1,
                                                          condition: x => x < 20,
                                                          iterate: x => x + 1,
                                                          resultSelector: x => x,
                                                          timeSelector: x => timeSpan);

            obs.Subscribe(x =>
            {
                if (x == 10)
                {
                    GC.Collect();
                    GC.WaitForPendingFinalizers();
                    GC.Collect();
                    MemoryProfiler.Dump();
                }
                Console.WriteLine(x);
            }, () => mre.Set());


        }

        static void Demo2(ManualResetEvent mre)
        {
            IDisposable cancel;

            TimeSpan timeSpan = TimeSpan.FromSeconds(1);

            Func<IScheduler, int, IDisposable> recurse = null;
            recurse = (self, state) =>
            {
                Console.WriteLine(state);

                if (state == 10)
                {
                    /* At this point, an inspection reveals
                       the disposables are not getting disposed
                       providing the lines at the bottom remain
                       commented out and this is a release build */                    
                    GC.Collect();
                    GC.WaitForPendingFinalizers();
                    GC.Collect();
                    MemoryProfiler.Dump();
                }

                if (state == 20)
                {
                    mre.Set();
                    return Disposable.Empty;
                }

                return self.Schedule(state + 1, timeSpan, recurse);
            };

            cancel = Scheduler.Default.Schedule(1, timeSpan, recurse);

            mre.WaitOne();

            GC.Collect();
            GC.WaitForPendingFinalizers();
            GC.Collect();
            MemoryProfiler.Dump();

            // Refering to cancel in any way here will prevent it from
            // being GC'd and will cause a "leak" of the chain of disposables
            // it roots. Try uncommenting either of the lines below to see the
            // leak appear. Note the leak will appear regardless in a debug build
            // because in debug builds local variables are not GC'd until the end
            // of a method in order to aid debugging
            
            // if(cancel == null) Console.WriteLine("Hang on to cancel");
            // cancel.Dispose();

        }
    }
}