using System;

using System.Reactive;

using System.Reactive.Concurrency;

using System.Reactive.Disposables;

using System.Reactive.Linq;

using System.Threading.Tasks;

using System.Threading;

public class Program

{

    public static void Main()

    {       

        Task.Run(async () =>

        {

            await RunMyCodeAsync();

        }).GetAwaiter().GetResult();

        Console.WriteLine($"{DateTime.UtcNow}: We are done.");

    }

    private static Random _rand = new Random();

    public static TimeSpan HeartbeatInterval => TimeSpan.FromMilliseconds(_rand.Next(100,1000));

    private static async Task RunMyCodeAsync()

    {

        var _cts = new CancellationTokenSource();

        Observable

            .Defer(() => Observable.FromAsync(_ => PublishHeartBeatAsync(_cts.Token)))

            .Delay(_ => Observable.Timer(HeartbeatInterval))

            .Repeat()

            .TakeUntil(Observable.Create<Unit>(obs => _cts.Token.Register(() => obs.OnNext(Unit.Default))))

            .Subscribe(

                _ => { }, // OnNext is not used

                () => Console.WriteLine($"{DateTime.UtcNow}: Observable completed."

            ));

        await Task.Delay(3000);

    }

    private static IObservable<Unit> GetHeartbeatObservable(CancellationToken ct) =>

    Observable.Create<Unit>(observer =>

    {

        async Task<IDisposable> RecursiveHeartbeatPublish(IScheduler scheduler, int _)

        {

            // Publish heartbeat.

            await PublishHeartBeatAsync(ct)

                .ContinueWith(task =>

                {

                    if (ct.IsCancellationRequested)

                    {

                        observer.OnCompleted();

                        return;

                    }

​

                    if (task.IsFaulted)

                    {

                        // Throw the exception to be handled by the observer, which should have retry and logging logic to handle retries sooner.

                        var ex =

                            task.Exception

                            ?? new Exception(

                                $"Unspecified error occurred in {nameof(PublishHeartBeatAsync)}."

                            );

                        //observer.OnError(new Exception("OnError test"));

                        Console.WriteLine($"{DateTime.UtcNow}: Throwing from RecursiveHeartbeatPublish...");

                        throw ex;

                    }

​

                    // Reschedule after the configured interval.

                    scheduler.Schedule(0, TimeSpan.FromSeconds(1), RecursiveHeartbeatPublish);

                });

​

            return Disposable.Empty;

        }

        // Start recursive scheduling.

        var scheduler = Scheduler.Default;

        return scheduler.Schedule(0, RecursiveHeartbeatPublish);

    });

    public static async Task PublishHeartBeatAsync(CancellationToken ct)

    {

        Console.WriteLine($"{DateTime.UtcNow.ToString("HH:mm:ss.fff")}: Hi from PublishHeartBeatAsync");

        //throw new Exception("Test exception.");

    }

}