using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using nft.framework;
using System.IO;
using System.Diagnostics;
using System.Reflection;
using System.Drawing;
using System.Drawing.Imaging;
using System.Threading.Tasks;
using System.Threading;
using System.Reactive.Linq;
using System.Reactive.Concurrency;
using System.Windows.Forms;
using System.Reactive.Subjects;
using System.Reactive.Disposables;
using System.Reactive;

namespace nft.test.test.RxPractice
{
    static class RxTestMain
    {

        [TestEntry("RxTest#HeavyTaskModel同期処理テスト")]
        static private void DoSyncHeavyTask()
        {
            HeavyTaskModel model = new HeavyTaskModel();
            model.NotifyProgress.Subscribe(v => Debug.WriteLine(v,"progress0"));
            Debug.WriteLine("Start process!");
            model.Process();
        }

        [TestEntry("RxTest#HeavyTaskModel非同期処理テスト")]
        static private void DoAsyncHeavyTask()
        {
            HeavyTaskModel model = new HeavyTaskModel();
            // SynchronizationContext.Currentは Observable.Startから呼ぶとnullになるので使えない
            // メインスレッドから直接この関数を呼ぶ場合には有効
            model.NotifyProgress/*.ObserveOn(SynchronizationContext.Current)*/
                .Subscribe(v => Debug.WriteLine(v, "progress1"), ()=> Debug.WriteLine("Subscriber completed"));
            var o = Observable.Start(() =>
            {
                Debug.WriteLine("Start process!");
                model.Process();
            }).Finally(() => Debug.WriteLine("Process completed."));
            //o.Wait();   // 終了待機１、Finallyが呼ばれたあと戻る
            o.GetAwaiter().GetResult(); // 終了待機2、Finallyが呼ばれる前に戻る
        }

        [TestEntry("RxTest#HeavyTaskModel非同期処理テスト２")]
        static private async Task DoAsyncHeavyTask2()
        {
            int i = 0;
            var o = Observable.FromAsync(() => Task.Run(() => {
                HeavyTaskModel model = new HeavyTaskModel();
                model.NotifyProgress.Subscribe(v => Debug.WriteLine(v, "progress1"), () => Debug.WriteLine("Subscriber completed"));
                Debug.WriteLine("Start process!");
                model.Process();
                return ++i;
            }));
            o.Repeat(3).Subscribe(x => Debug.WriteLine(x,"Task result"), () => Debug.WriteLine("Task Complete"));
            await o.LastAsync(); // Repeatの一週目でくる
            Debug.WriteLine("Await1");
            await o.LastAsync(); // Repeatの二週目でくる
            Debug.WriteLine("Await2");
            await o.LastAsync(); // Repeatの三週目でくる
            Debug.WriteLine("Await3");
        }

        [TestEntry("RxTest#HeavyTaskModel非同期並行処理")]
        static private async Task DoAsyncHeavyTaskMulti()
        {
            var o = Observable.CombineLatest(
                Observable.Start(() => { HeavyTaskModel model = new HeavyTaskModel("A"); model.Process(); return "A End"; }),
                Observable.Start(() => { HeavyTaskModel model = new HeavyTaskModel("B"); model.Process(); return "B End"; }),
                Observable.Start(() => { HeavyTaskModel model = new HeavyTaskModel("C"); model.Process(); return "C End"; })
            ).Finally(() => Debug.WriteLine("All Process completed.")); ;
            foreach (string r in await o.FirstAsync())
            {
                Debug.WriteLine("Result = " + r);
            }
        }

        [TestEntry("RxTestMain#別スレッド処理＆キャンセル")]
        static private bool TestScedhulerAndCancel()
        {
            IObservable<int> ob =
                Observable.Create<int>(o => {
                    var cancel = new CancellationDisposable(); // internally creates a new CancellationTokenSource
                    NewThreadScheduler.Default.Schedule(() => {
                        int i = 0;
                        for (;;)
                        {
                            Thread.Sleep(100); 
                            if (cancel.Token.IsCancellationRequested)    // check cancel token periodically
                            {
                                Debug.WriteLine("Abort by cancel!");
                                o.OnCompleted(); // will not make it to the subscriber
                                return;
                            }
                            o.OnNext(i++);
                        }
                    });
                    return cancel;
                });
            IDisposable subscription = ob.Subscribe(i => Debug.WriteLine(i));
            Debug.WriteLine("Wait for 1 sec.");
            Thread.Sleep(1000);
            Debug.WriteLine("Dispose will cause cancel.");
            subscription.Dispose();
            return true;
        }

        static private AsyncCallback Callback;
        [TestEntry("RxTestMain#ストリーム操作サンプル")]
        static private async Task<bool> TestStream()
        {
            // Obsolated
            //var read = Observable.FromAsyncPattern<byte[], int, int, int>(inputStream.BeginRead, inputStream.EndRead);
            //IObservable<int> observable = read(someBytes, 0, 10);

            string file = "nftfw.resource.xml";
            byte[] buff = new byte[200];

            /****************** Old style before .NET Framework4.5 *******************/
            Subject<Unit> sub = new Subject<Unit>();
            FileStream fs = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.ReadWrite);
            Callback = result =>
            {
                int length = fs.EndRead(result);
                if (length > 0)
                {
                    Debug.WriteLine("Read Byte = " + length, "Old");
                    fs.BeginRead(buff, 0, 200, Callback, fs);
                }
                else
                {
                    Debug.WriteLine("Read Completed.", "Old");
                    fs.Close();
                    sub.OnCompleted();
                }
            };
            fs.BeginRead(buff, 0, 200, Callback, fs);
            await sub.FirstOrDefaultAsync(); // ここで待たないと同時並行で下の読み込みが始まる。
            sub.Dispose();

            /****************** .NET Framework4.5 or later *******************/
            FileStream fs2 = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.ReadWrite);
            StreamReader sr = new StreamReader(fs2);
            string s = await sr.ReadLineAsync();
            while (s != null)
            {
                Debug.WriteLine(s, "New");
                s = await sr.ReadLineAsync();
            }

            return true;
        }

        [TestEntry("RxTestMain#IEnumerable操作サンプル")]
        static private async Task<bool> TestEnumeration()
        {
            IEnumerable<int> someInts = new List<int> { 1, 2, 3, 4, 5 };

            // To convert a generic IEnumerable into an IObservable, use the ToObservable extension method.
            IObservable<int> observable = someInts.ToObservable();

            var o = Observable.Start(() => {
                observable.Subscribe<int>(i => Debug.WriteLine("Enumerated from list = " + i));
            });
            await o.Finally(() => Debug.WriteLine("Completed."));
            return true;
        }

        [TestEntry("RxTestMain#周期タイマー&スイッチ")]
        static private bool TestIntervalTimer()
        {
            var switches = new BehaviorSubject<bool>(false);
            var interval = Observable.Interval(TimeSpan.FromMilliseconds(500)).TakeUntil(switches.Where(on => !on));
            var query = switches.DistinctUntilChanged().Where(on => on).SelectMany(interval);

            using (query.TimeInterval().Subscribe(count => Debug.WriteLine(count, "timer")))
            {
                Debug.WriteLine("start");
                Debug.WriteLine("wait {0}s", 1);
                Thread.Sleep(TimeSpan.FromSeconds(1));
                Debug.WriteLine("ON","switch"); switches.OnNext(true);
                Thread.Sleep(TimeSpan.FromSeconds(2));
                Debug.WriteLine("OFF", "switch"); switches.OnNext(false);
                Debug.WriteLine("wait {0}s", 2);
                Thread.Sleep(TimeSpan.FromSeconds(2));
                Debug.WriteLine("ON", "switch"); switches.OnNext(true);
                Thread.Sleep(TimeSpan.FromSeconds(3));
                Debug.WriteLine("OFF", "switch"); switches.OnNext(false);
                Debug.WriteLine("wait {0}s", 1);
                Thread.Sleep(TimeSpan.FromSeconds(1));
                Debug.WriteLine("OFF", "switch"); switches.OnNext(false);
                Debug.WriteLine("wait {0}s", 1);
                Thread.Sleep(TimeSpan.FromSeconds(1));
                Debug.WriteLine("ON", "switch"); switches.OnNext(true);
                Thread.Sleep(TimeSpan.FromSeconds(1));
                Debug.WriteLine("ON", "switch"); switches.OnNext(true);
                Thread.Sleep(TimeSpan.FromSeconds(2));
            }
            return true;
        }

        [TestEntry("RxTestMain#ノーマルSubject")]
        static private bool TestSubject() {
            return TestSubjectMain(new Subject<int>());
        }

        [TestEntry("RxTestMain#全キャッシュ/ReplaySubject")]
        static private bool TestReplaySubject() {
            return TestSubjectMain(new ReplaySubject<int>());
        }

        [TestEntry("RxTestMain#最新キャッシュ/BehaviorSubject")]
        static private bool TestBehaviorSubject() {
            return TestSubjectMain(new BehaviorSubject<int>(0));
        }

        [TestEntry("RxTestMain#非同期/AsyncSubject")]
        static private bool TestAsyncSubject() {
            return TestSubjectMain(new AsyncSubject<int>());
        }

        static private bool TestSubjectMain(SubjectBase<int> subject) {
            //var subject = new ReplaySubject<int>();
            var disposerA = subject.DebugSubscribe("A");
            subject.OnNext(1);
            var disposerB = subject.DebugSubscribe("B");
            subject.OnNext(2);
            disposerA.DebugDispose("A");
            var disposerC = subject.DebugSubscribe("C");
            subject.OnNext(3);
            subject.OnCompleted();
            subject.OnNext(4);
            var disposerD = subject.DebugSubscribe("D");
            subject.OnNext(5);
            return true;
        }

        static IDisposable DebugSubscribe<T>(this IObservable<T> source, string name) {
            Debug.WriteLine("Enter Subscribe", name);
            var disposer = source.Subscribe
            (
                value => Debug.WriteLine("-> OnNext("+value+")", name),
                () => Debug.WriteLine("-> OnCompleted", name)
            );
            Debug.WriteLine("Exit Subscribe", name);
            return disposer;
        }

        static void DebugDispose(this IDisposable source, string name) {
            Debug.WriteLine("Enter Dispose", name);
            source.Dispose();
            Debug.WriteLine("Exit Dispose", name);
        }
    }


}
