using System;

using System.Collections.Generic;

using System.Reactive.Concurrency;

using System.Reactive.Disposables;

using System.Reactive.Linq;

using System.Threading;

using System.Threading.Tasks;

​

class Program

{

    static void Main()

    {

        CreateObservable(Scheduler.CurrentThread,TimeSpan.FromSeconds(0))

            .Catch<string, Exception>(ex =>

        {

            Console.WriteLine($"Caught first exception in stream: {ex.Message}");

            return Observable.Empty<string>();

        }).Subscribe();

        var observable = Observable.Create<string>(async (observer, cancellationToken) =>

        {

            //throw new OperationCanceledException();

            try{//

                observer.OnError( new OperationCanceledException());

            //Task.Run(async () =>

            //{throw new OperationCanceledException();

                int count = 0;

                while (++count < 6)

                {

                    //cancellationToken.ThrowIfCancellationRequested();

                    await Task.Delay(100); // delay supports cancellation

                }

                observer.OnCompleted();

           // });

            }                catch (AggregateException ex)

                {Console.WriteLine("fwe");

                    foreach (var innerException in ex.InnerExceptions)

                    {

                        if (innerException is OperationCanceledException)

                        {throw new OperationCanceledException();

                            observer.OnCompleted();

                            break;

                        }

                        observer.OnError(innerException);

                        break;

                    }

                }

​

            return Task.FromResult<string>("");

        })

        .Catch<string, Exception>(ex =>

        {

            Console.WriteLine($"Caught exception in stream: {ex.Message}");

            return Observable.Empty<string>();

        });

​

        var subscription = observable.Subscribe(

            onNext: v => Console.WriteLine($"Received: {v}"),

            onError: ex => Console.WriteLine($"Error: {ex.Message}"),

            onCompleted: () => Console.WriteLine("Completed"));

​

        Thread.Sleep(100);

        Console.WriteLine("Disposing subscription (cancelling)...");

        subscription.Dispose();

​

        Thread.Sleep(500);

    }

        private static IObservable<string> CreateObservable(IScheduler scheduler, TimeSpan delay)

    {

        return Observable.Create<string>(observer =>

        {

            var schedulingIsCompleted = new TaskCompletionSource<bool>();

​

            Func<IScheduler, IDisposable> recursiveSchedule = null;

            recursiveSchedule = (sched) =>

            {

                var hasValue = false;

                observer.OnError(new OperationCanceledException());

​

                return Disposable.Empty;

            };

​

            scheduler.Schedule(TimeSpan.FromSeconds(3), () => observer.OnCompleted());

​

            return scheduler.Schedule(

                TimeSpan.FromSeconds(0), 

                (sched, dueTime) => recursiveSchedule(sched) 

            );

        });

    }

}