using System;
using System.Reactive.Linq;

namespace RX_Sample._03_ReactiveX
{
    public class DispatchObserver<T> : IObserver<T>
    {
        private readonly IObserver<T> observer;
        private readonly IDispatcher customDispatcher;

        public DispatchObserver(
            IObserver<T> observer,
            IDispatcher customDispatcher)
        {
            this.observer = observer;
            this.customDispatcher = customDispatcher;
        }

        public void OnCompleted()
        {
            customDispatcher.Dispatch(observer.OnCompleted);
        }

        public void OnError(Exception error)
        {
            customDispatcher.Dispatch(() => observer.OnError(error));
        }

        public void OnNext(T value)
        {
            customDispatcher.Dispatch(() => observer.OnNext(value));
        }
    }

    public class DispatcherObservable<T> : IObservable<T>
    {
        private readonly IObservable<T> innerObservable;
        private readonly IDispatcher customDispatcher;

        public DispatcherObservable(
            IObservable<T> innerObservable,
            IDispatcher customDispatcher)
        {
            this.innerObservable = innerObservable;
            this.customDispatcher = customDispatcher;
        }

        public IDisposable Subscribe(IObserver<T> observer)
        {
            return innerObservable.Subscribe(new DispatchObserver<T>(observer, customDispatcher));
        }
    }

    static class DispatchExtensions
    {
        public static IObservable<T> Dispatch<T>(this IObservable<T> observable, IDispatcher customDispatcher)
        {
            return new DispatcherObservable<T>(observable, customDispatcher);
        }
    }

    class Sample05
    {
        public Sample05(IDispatcher customDispatcher)
        {
            Observable.Range(0, 10).Dispatch(customDispatcher);
        }
    }
}
