using NUnit.Framework;

using System;

using System.Collections.Generic;

using System.Threading;

using System.Threading.Tasks;

using System.Reactive;

using System.Reactive.Linq;

using System.Reactive.Disposables;

using System.Reactive.Concurrency;

using Microsoft.Reactive.Testing;

​

public class ObservableTests : ReactiveTest

{

    private static IObservable<string> CreateObservable(AsyncOperationWorker worker, IScheduler scheduler)

    {

        return Observable.Create<string>(observer =>

        {

            var schedulingIsCompleted = new TaskCompletionSource<bool>();

            // Synchronous schedule wrapping async operations

            scheduler.Schedule<object>(

                null,

                TimeSpan.FromSeconds(0),

                (_, state) =>

                {

                    try

                    {

                        var task = Task.Run(async () => await worker.ExecuteAsync());

                        task.Wait();

                    }

                    catch (AggregateException ex)

                    {

                        foreach (var innerException in ex.InnerExceptions)

                        {

                            if (innerException is OperationCanceledException)

                            {

                                observer.OnCompleted();

                                schedulingIsCompleted.TrySetResult(true);

                                break;

                            }

                            observer.OnError(innerException);

                            schedulingIsCompleted.TrySetResult(true);

                            break;

                        }

                    }

                    observer.OnNext("From async");

​

                    return Disposable.Empty;

                });

​

            scheduler.Schedule(

                TimeSpan.FromSeconds(3),

                () => observer.OnCompleted());

​

            return Disposable.Empty;

        });

    }

    private class AsyncOperationWorker

    {

        private readonly CancellationToken _token;

        private readonly Exception _simulatedException;

​

        public AsyncOperationWorker(CancellationToken token, Exception simulatedException = null)

        {

            _token = token;

            _simulatedException = simulatedException;

        }

​

        public async Task ExecuteAsync()

        {

            await Task.Delay(10);

            if (_simulatedException != null)

                throw _simulatedException;

            _token.ThrowIfCancellationRequested();

        }

    }

    [Test]

    public void Async_schedule_completes_after_sync()

    {

        // Arrange

        var testScheduler = new TestScheduler();

        var testObserver = testScheduler.CreateObserver<string>();

        var cts = new CancellationTokenSource();

        var worker = new AsyncOperationWorker(cts.Token);

​

        var source = CreateObservable(worker, testScheduler);

​

        source.Subscribe(testObserver);

​

        // Act

        cts.Cancel(); 

        testScheduler.Start();

​

        var expected = new[] { OnCompleted<string>(1) }; // Expected results based on virtual time scheduling

​

        // Assert

        Assert.That(testObserver.Messages, Is.EqualTo(expected).Using(new NotificationComparer()));

    }

    [Test]

    public void Should_contain_OnError_when_worker_throws_exception()

    {

        // Arrange

        var testScheduler = new TestScheduler();

        var testObserver = testScheduler.CreateObserver<string>();

        var exception = new InvalidOperationException("Test failure");

        var worker = new AsyncOperationWorker(CancellationToken.None, exception);

​

        var source = CreateObservable(worker, testScheduler);

        source.Subscribe(testObserver);

​

        // Act

        testScheduler.Start();

​

        var expected = new[] {

            OnError<string>(1, exception)

        };

        // Assert

        Assert.That(testObserver.Messages, Is.EqualTo(expected).Using(new NotificationComparer()));

    }

    [Test]

    public void Should_contain_OnNext_when_operation_succeeds()

    {

        // Arrange

        var testScheduler = new TestScheduler();

        var testObserver = testScheduler.CreateObserver<string>();

        var worker = new AsyncOperationWorker(CancellationToken.None);

​

        // Act

        var source = CreateObservable(worker, testScheduler);

        source.Subscribe(testObserver);

        testScheduler.Start();

​

        // Assert

        var expected = new[] {

            OnNext(1, "From async"),

            OnCompleted<string>(30_000_000)

        };

        Assert.That(testObserver.Messages, Is.EqualTo(expected).Using(new NotificationComparer()));

    }

}

​

class NotificationComparer : IEqualityComparer<Recorded<Notification<string>>>

{

    public bool Equals(Recorded<Notification<string>> x, Recorded<Notification<string>> y)

    {

        // Fail if notification kinds or timestamps differ

        if (x.Value.Kind != y.Value.Kind || x.Time != y.Time)

            return false;

​

        switch (x.Value.Kind)

        {

            case NotificationKind.OnNext:

                // Compare values for OnNext notifications

                return x.Value.Value.Equals(y.Value.Value);

            case NotificationKind.OnError:

                // Compare exception types and messages for OnError

                var xEx = x.Value.Exception;

                var yEx = y.Value.Exception;

                return xEx?.GetType() == yEx?.GetType() && 

                       xEx?.Message == yEx?.Message;

            case NotificationKind.OnCompleted:

                // All OnCompleted notifications are considered equal

                return true;

            default:

                return false;

        }

    }

​

    public int GetHashCode(Recorded<Notification<string>> obj)

    {

        return HashCode.Combine(obj.Time, obj.Value.Kind);

    }

}

​

public static class Neo

{

            public static IDisposable Schedule(this IScheduler scheduler, TimeSpan dueTime, Action<Action<TimeSpan>> action)

        {

            if (scheduler == null)

            {

                throw new ArgumentNullException(nameof(scheduler));

            }

​

            if (action == null)

            {

                throw new ArgumentNullException(nameof(action));

            }

​

            return scheduler.Schedule(action, dueTime,  (a, self) => a(ts => self(a, ts)));

            }

}

​

public class Program

{

    public static void Main()

    {

        new NUnitLite.AutoRun().Execute(["--noc" /*no colored output for .Net Fiddle*/]);

    }

​

}