#define RXCPP_SUBJECT_TEST_ASYNC 1

#include "../test.h"

#include <rxcpp/operators/rx-finally.hpp>

#include <future>


const int static_onnextcalls = 10000000;
static int aliased = 0;

SCENARIO("for loop locks mutex", "[hide][for][mutex][long][perf]"){
    const int& onnextcalls = static_onnextcalls;
    GIVEN("a for loop"){
        WHEN("locking mutex 100 million times"){
            using namespace std::chrono;
            typedef steady_clock clock;

            int c = 0;
            int n = 1;
            auto start = clock::now();
            std::mutex m;
            for (int i = 0; i < onnextcalls; i++) {
                std::unique_lock<std::mutex> guard(m);
                ++c;
            }
            auto finish = clock::now();
            auto msElapsed = duration_cast<milliseconds>(finish-start);
            std::cout << "loop mutex          : " << n << " subscribed, " << c << " on_next calls, " << msElapsed.count() << "ms elapsed " << c / (msElapsed.count() / 1000.0) << " ops/sec" << std::endl;

        }
    }
}

namespace syncwithvoid {
template<class T, class OnNext>
class sync_subscriber
{
public:
    OnNext onnext;
    bool issubscribed;
    explicit sync_subscriber(OnNext on)
        : onnext(on)
        , issubscribed(true)
    {
    }
    bool is_subscribed() {return issubscribed;}
    void unsubscribe() {issubscribed = false;}
    void on_next(T v) {
        onnext(v);
    }
};
}
SCENARIO("for loop calls void on_next(int)", "[hide][for][asyncobserver][baseline][perf]"){
    const int& onnextcalls = static_onnextcalls;
    GIVEN("a for loop"){
        WHEN("calling on_next 100 million times"){
            using namespace std::chrono;
            typedef steady_clock clock;

            auto c = std::addressof(aliased);
            *c = 0;
            int n = 1;
            auto start = clock::now();
            auto onnext = [c](int){++*c;};
            syncwithvoid::sync_subscriber<int, decltype(onnext)> scbr(onnext);
            for (int i = 0; i < onnextcalls && scbr.is_subscribed(); i++) {
                scbr.on_next(i);
            }
            auto finish = clock::now();
            auto msElapsed = duration_cast<milliseconds>(finish-start);
            std::cout << "loop void           : " << n << " subscribed, " << *c << " on_next calls, " << msElapsed.count() << "ms elapsed " << *c / (msElapsed.count() / 1000.0) << " ops/sec" << std::endl;

        }
    }
}

namespace asyncwithready {
// ready is an immutable class.
class ready
{
public:
    typedef std::function<void()> onthen_type;
private:
    std::function<void(onthen_type)> setthen;
public:
    ready() {}
    ready(std::function<void(onthen_type)> st) : setthen(st) {}
    bool is_ready() {return !setthen;}
    void then(onthen_type ot) {
        if (is_ready()) {
            abort();
        }
        setthen(ot);
    }
};
template<class T, class OnNext>
class async_subscriber
{
public:
    OnNext onnext;
    bool issubscribed;
    int count;
    explicit async_subscriber(OnNext on)
        : onnext(on)
        , issubscribed(true)
        , count(0)
    {
    }
    bool is_subscribed() {return issubscribed;}
    void unsubscribe() {issubscribed = false;}
    ready on_next(T v) {
        // push v onto queue

        // under some condition pop v off of queue and pass it on
        onnext(v);

        // for demo purposes
        // simulate queue full every 100000 items
        if (count == 100000) {
            // 'queue is full'
            ready no([this](ready::onthen_type ot){
                // full version will sync producer and consumer (in producer push and consumer pop)
                // and decide when to restart the producer
                if (!this->count) {
                    ot();
                }
            });
            // set queue empty since the demo has no separate consumer thread
            count = 0;
            // 'queue is empty'
            return no;
        }
        static const ready yes;
        return yes;
    }
};
}
SCENARIO("for loop calls ready on_next(int)", "[hide][for][asyncobserver][ready][perf]"){
    static const int& onnextcalls = static_onnextcalls;
    GIVEN("a for loop"){
        WHEN("calling on_next 100 million times"){
            using namespace std::chrono;
            typedef steady_clock clock;

            auto c = std::addressof(aliased);
            *c = 0;
            int n = 1;
            auto start = clock::now();
            auto onnext = [&c](int){++*c;};
            asyncwithready::async_subscriber<int, decltype(onnext)> scbr(onnext);
            asyncwithready::ready::onthen_type chunk;
            int i = 0;
            chunk = [&chunk, scbr, i]() mutable {
                for (; i < onnextcalls && scbr.is_subscribed(); i++) {
                    auto controller = scbr.on_next(i);
                    if (!controller.is_ready()) {
                        controller.then(chunk);
                        return;
                    }
                }
            };
            chunk();
            auto finish = clock::now();
            auto msElapsed = duration_cast<milliseconds>(finish-start);
            std::cout << "loop ready          : " << n << " subscribed, " << *c << " on_next calls, " << msElapsed.count() << "ms elapsed " << *c / (msElapsed.count() / 1000.0) << " ops/sec" << std::endl;

        }
    }
}

namespace asyncwithfuture {
class unit {};
template<class T, class OnNext>
class async_subscriber
{
public:
    OnNext onnext;
    bool issubscribed;
    explicit async_subscriber(OnNext on)
        : onnext(on)
        , issubscribed(true)
    {
    }
    bool is_subscribed() {return issubscribed;}
    void unsubscribe() {issubscribed = false;}
    std::future<unit> on_next(T v) {
        std::promise<unit> ready;
        ready.set_value(unit());
        onnext(v); return ready.get_future();}
};
}
SCENARIO("for loop calls std::future<unit> on_next(int)", "[hide][for][asyncobserver][future][long][perf]"){
    const int& onnextcalls = static_onnextcalls;
    GIVEN("a for loop"){
        WHEN("calling on_next 100 million times"){
            using namespace std::chrono;
            typedef steady_clock clock;

            auto c = std::addressof(aliased);
            *c = 0;
            int n = 1;
            auto start = clock::now();
            auto onnext = [&c](int){++*c;};
            asyncwithfuture::async_subscriber<int, decltype(onnext)> scbr(onnext);
            for (int i = 0; i < onnextcalls && scbr.is_subscribed(); i++) {
                auto isready = scbr.on_next(i);
                if (isready.wait_for(std::chrono::milliseconds(0)) == std::future_status::timeout) {
                    isready.wait();
                }
            }
            auto finish = clock::now();
            auto msElapsed = duration_cast<milliseconds>(finish-start);
            std::cout << "loop future<unit>   : " << n << " subscribed, " << *c << " on_next calls, " << msElapsed.count() << "ms elapsed " << *c / (msElapsed.count() / 1000.0) << " ops/sec" << std::endl;

        }
    }
}

SCENARIO("for loop calls observer", "[hide][for][observer][perf]"){
    const int& onnextcalls = static_onnextcalls;
    GIVEN("a for loop"){
        WHEN("observing 100 million ints"){
            using namespace std::chrono;
            typedef steady_clock clock;

            static int& c = aliased;
            int n = 1;

            c = 0;
            auto start = clock::now();
            auto o = rx::make_observer<int>(
                [](int){++c;},
                [](rxu::error_ptr){abort();});
            for (int i = 0; i < onnextcalls; i++) {
                o.on_next(i);
            }
            o.on_completed();
            auto finish = clock::now();
            auto msElapsed = duration_cast<milliseconds>(finish-start);
            std::cout << "loop -> observer    : " << n << " subscribed, " << c << " on_next calls, " << msElapsed.count() << "ms elapsed " << c / (msElapsed.count() / 1000.0) << " ops/sec"<< std::endl;
        }
    }
}

SCENARIO("for loop calls subscriber", "[hide][for][subscriber][perf]"){
    const int& onnextcalls = static_onnextcalls;
    GIVEN("a for loop"){
        WHEN("observing 100 million ints"){
            using namespace std::chrono;
            typedef steady_clock clock;

            static int& c = aliased;
            int n = 1;

            c = 0;
            auto start = clock::now();
            auto o = rx::make_subscriber<int>(
                [](int){++c;},
                [](rxu::error_ptr){abort();});
            for (int i = 0; i < onnextcalls && o.is_subscribed(); i++) {
                o.on_next(i);
            }
            o.on_completed();
            auto finish = clock::now();
            auto msElapsed = duration_cast<milliseconds>(finish-start);
            std::cout << "loop -> subscriber  : " << n << " subscribed, " << c << " on_next calls, " << msElapsed.count() << "ms elapsed " << c / (msElapsed.count() / 1000.0) << " ops/sec" << std::endl;
        }
    }
}

SCENARIO("range calls subscriber", "[hide][range][subscriber][perf]"){
    const int& onnextcalls = static_onnextcalls;
    GIVEN("a range"){
        WHEN("observing 100 million ints"){
            using namespace std::chrono;
            typedef steady_clock clock;

            static int& c = aliased;
            int n = 1;

            c = 0;
            auto start = clock::now();

            rxs::range<int>(1, onnextcalls).subscribe(
                [](int){
                    ++c;
                },
                [](rxu::error_ptr){abort();});

            auto finish = clock::now();
            auto msElapsed = duration_cast<milliseconds>(finish-start);
            std::cout << "range -> subscriber : " << n << " subscribed, " << c << " on_next calls, " << msElapsed.count() << "ms elapsed " << c / (msElapsed.count() / 1000.0) << " ops/sec" << std::endl;
        }
    }
}

SCENARIO("for loop calls subject", "[hide][for][subject][subjects][long][perf]"){
    static const int& onnextcalls = static_onnextcalls;
    GIVEN("a for loop and a subject"){
        WHEN("multicasting a million ints"){
            using namespace std::chrono;
            typedef steady_clock clock;

            for (int n = 0; n < 10; n++)
            {
                auto p = std::make_shared<int>(0);
                auto c = std::make_shared<int>(0);
                rxsub::subject<int> sub;

#if RXCPP_SUBJECT_TEST_ASYNC
                std::vector<std::future<int>> f(n);
                std::atomic<int> asyncUnsubscriptions{0};
#endif

                auto o = sub.get_subscriber();

                o.add(rx::make_subscription([c, n](){
                    auto expected = n * onnextcalls;
                    REQUIRE(*c == expected);
                }));

                for (int i = 0; i < n; i++) {
#if RXCPP_SUBJECT_TEST_ASYNC
                    f[i] = std::async([sub, o, &asyncUnsubscriptions]() {
                        auto source = sub.get_observable();
                        while(o.is_subscribed()) {
                            std::this_thread::sleep_for(std::chrono::milliseconds(100));
                            rx::composite_subscription cs;
                            source
                            .finally([&asyncUnsubscriptions](){
                                ++asyncUnsubscriptions;})
                            .subscribe(
                                rx::make_subscriber<int>(
                                cs,
                                [cs](int){
                                    cs.unsubscribe();
                                },
                                [](rxu::error_ptr){abort();}));
                        }
                        return 0;
                    });
#endif
                    sub.get_observable().subscribe(
                        [c, p](int){
                            ++(*c);
                        },
                        [](rxu::error_ptr){abort();});
                }

                auto start = clock::now();
                for (int i = 0; i < onnextcalls && o.is_subscribed(); i++) {
#if RXCPP_DEBUG_SUBJECT_RACE
                    if (*p != *c) abort();
                    (*p) += n;
#endif
                    o.on_next(i);
                }
                o.on_completed();
                auto finish = clock::now();
                auto msElapsed = duration_cast<milliseconds>(finish-start);
                std::cout << "loop -> subject     : " << n << " subscribed, " << std::setw(9) << (*c) << " on_next calls, ";
#if RXCPP_SUBJECT_TEST_ASYNC
                std::cout << std::setw(4) << asyncUnsubscriptions << " async, ";
#endif
                std::cout << msElapsed.count() << "ms elapsed " << *c / (msElapsed.count() / 1000.0) << " ops/sec" << std::endl;
            }
        }
    }
}

SCENARIO("range calls subject", "[hide][range][subject][subjects][long][perf]"){
    static const int& onnextcalls = static_onnextcalls;
    GIVEN("a range and a subject"){
        WHEN("multicasting a million ints"){
            using namespace std::chrono;
            typedef steady_clock clock;
            for (int n = 0; n < 10; n++)
            {
                auto p = std::make_shared<int>(0);
                auto c = std::make_shared<int>(0);
                rxsub::subject<int> sub;

#if RXCPP_SUBJECT_TEST_ASYNC
                std::vector<std::future<int>> f(n);
                std::atomic<int> asyncUnsubscriptions{0};
#endif

                auto o = sub.get_subscriber();

                o.add(rx::make_subscription([c, n](){
                    auto expected = n * onnextcalls;
                    REQUIRE(*c == expected);
                }));

                for (int i = 0; i < n; i++) {
#if RXCPP_SUBJECT_TEST_ASYNC
                    f[i] = std::async([sub, o, &asyncUnsubscriptions]() {
                        while(o.is_subscribed()) {
                            std::this_thread::sleep_for(std::chrono::milliseconds(100));
                            rx::composite_subscription cs;
                            sub.get_observable()
                            .finally([&asyncUnsubscriptions](){
                                ++asyncUnsubscriptions;})
                            .subscribe(cs,
                                [cs](int){
                                    cs.unsubscribe();
                                },
                                [](rxu::error_ptr){abort();});
                        }
                        return 0;
                    });
#endif
                    sub.get_observable()
                        .subscribe(
                            [c, p](int){
                               ++(*c);
                            },
                            [](rxu::error_ptr){abort();}
                        );
                }

                auto start = clock::now();
                rxs::range<int>(1, onnextcalls)
#if RXCPP_DEBUG_SUBJECT_RACE
                    .filter([c, p, n](int){
                        if (*p != *c) abort();
                        (*p) += n;
                        return true;
                    })
#endif
                    .subscribe(o);
                auto finish = clock::now();
                auto msElapsed = duration_cast<milliseconds>(finish-start);
                std::cout << "range -> subject    : " << n << " subscribed, " << std::setw(9) << (*c) << " on_next calls, ";
#if RXCPP_SUBJECT_TEST_ASYNC
                std::cout << std::setw(4) << asyncUnsubscriptions << " async, ";
#endif
                std::cout << msElapsed.count() << "ms elapsed " << *c / (msElapsed.count() / 1000.0) << " ops/sec"<< std::endl;
            }
        }
    }
}


SCENARIO("subject - infinite source", "[subject][subjects]"){
    GIVEN("a subject and an infinite source"){

        auto sc = rxsc::make_test();
        auto w = sc.create_worker();
        const rxsc::test::messages<int> on;
        const rxsc::test::messages<bool> check;

        auto xs = sc.make_hot_observable({
            on.next(70, 1),
            on.next(110, 2),
            on.next(220, 3),
            on.next(270, 4),
            on.next(340, 5),
            on.next(410, 6),
            on.next(520, 7),
            on.next(630, 8),
            on.next(710, 9),
            on.next(870, 10),
            on.next(940, 11),
            on.next(1020, 12)
        });

        rxsub::subject<int> s;

        auto results1 = w.make_subscriber<int>();

        auto results2 = w.make_subscriber<int>();

        auto results3 = w.make_subscriber<int>();

        WHEN("multicasting an infinite source"){

            auto checks = rxu::to_vector({
                check.next(0, false)
            });

            auto record = [&s, &check, &checks](long at) -> void {
                checks.push_back(check.next(at, s.has_observers()));
            };

            auto o = s.get_subscriber();

            w.schedule_absolute(100, [&s, &o, &checks, &record](const rxsc::schedulable&){
                s = rxsub::subject<int>(); o = s.get_subscriber(); checks.clear(); record(100);});
            w.schedule_absolute(200, [&xs, &o, &record](const rxsc::schedulable&){
                xs.subscribe(o); record(200);});
            w.schedule_absolute(1000, [&o, &record](const rxsc::schedulable&){
                o.unsubscribe(); record(1000);});

            w.schedule_absolute(300, [&s, &results1, &record](const rxsc::schedulable&){
                s.get_observable().subscribe(results1); record(300);});
            w.schedule_absolute(400, [&s, &results2, &record](const rxsc::schedulable&){
                s.get_observable().subscribe(results2); record(400);});
            w.schedule_absolute(900, [&s, &results3, &record](const rxsc::schedulable&){
                s.get_observable().subscribe(results3); record(900);});

            w.schedule_absolute(600, [&results1, &record](const rxsc::schedulable&){
                results1.unsubscribe(); record(600);});
            w.schedule_absolute(700, [&results2, &record](const rxsc::schedulable&){
                results2.unsubscribe(); record(700);});
            w.schedule_absolute(800, [&results1, &record](const rxsc::schedulable&){
                results1.unsubscribe(); record(800);});
            w.schedule_absolute(950, [&results3, &record](const rxsc::schedulable&){
                results3.unsubscribe(); record(950);});

            w.start();

            THEN("result1 contains expected messages"){
                auto required = rxu::to_vector({
                    on.next(340, 5),
                    on.next(410, 6),
                    on.next(520, 7)
                });
                auto actual = results1.get_observer().messages();
                REQUIRE(required == actual);
            }

            THEN("result2 contains expected messages"){
                auto required = rxu::to_vector({
                    on.next(410, 6),
                    on.next(520, 7),
                    on.next(630, 8)
                });
                auto actual = results2.get_observer().messages();
                REQUIRE(required == actual);
            }

            THEN("result3 contains expected messages"){
                auto required = rxu::to_vector({
                    on.next(940, 11)
                });
                auto actual = results3.get_observer().messages();
                REQUIRE(required == actual);
            }

            THEN("checks contains expected messages"){
                auto required = rxu::to_vector({
                    check.next(100, false),
                    check.next(200, false),
                    check.next(300, true),
                    check.next(400, true),
                    check.next(600, true),
                    check.next(700, false),
                    check.next(800, false),
                    check.next(900, true),
                    check.next(950, false),
                    check.next(1000, false)
                });
                auto actual = checks;
                REQUIRE(required == actual);
            }

        }
    }
}

SCENARIO("subject - finite source", "[subject][subjects]"){
    GIVEN("a subject and an finite source"){

        auto sc = rxsc::make_test();
        auto w = sc.create_worker();
        const rxsc::test::messages<int> on;

        auto xs = sc.make_hot_observable({
            on.next(70, 1),
            on.next(110, 2),
            on.next(220, 3),
            on.next(270, 4),
            on.next(340, 5),
            on.next(410, 6),
            on.next(520, 7),
            on.completed(630),
            on.next(640, 9),
            on.completed(650),
            on.error(660, std::runtime_error("error on unsubscribed stream"))
        });

        rxsub::subject<int> s;

        auto results1 = w.make_subscriber<int>();

        auto results2 = w.make_subscriber<int>();

        auto results3 = w.make_subscriber<int>();

        WHEN("multicasting an infinite source"){

            auto o = s.get_subscriber();

            w.schedule_absolute(100, [&s, &o](const rxsc::schedulable&){
                s = rxsub::subject<int>(); o = s.get_subscriber();});
            w.schedule_absolute(200, [&xs, &o](const rxsc::schedulable&){
                xs.subscribe(o);});
            w.schedule_absolute(1000, [&o](const rxsc::schedulable&){
                o.unsubscribe();});

            w.schedule_absolute(300, [&s, &results1](const rxsc::schedulable&){
                s.get_observable().subscribe(results1);});
            w.schedule_absolute(400, [&s, &results2](const rxsc::schedulable&){
                s.get_observable().subscribe(results2);});
            w.schedule_absolute(900, [&s, &results3](const rxsc::schedulable&){
                s.get_observable().subscribe(results3);});

            w.schedule_absolute(600, [&results1](const rxsc::schedulable&){
                results1.unsubscribe();});
            w.schedule_absolute(700, [&results2](const rxsc::schedulable&){
                results2.unsubscribe();});
            w.schedule_absolute(800, [&results1](const rxsc::schedulable&){
                results1.unsubscribe();});
            w.schedule_absolute(950, [&results3](const rxsc::schedulable&){
                results3.unsubscribe();});

            w.start();

            THEN("result1 contains expected messages"){
                auto required = rxu::to_vector({
                    on.next(340, 5),
                    on.next(410, 6),
                    on.next(520, 7)
                });
                auto actual = results1.get_observer().messages();
                REQUIRE(required == actual);
            }

            THEN("result2 contains expected messages"){
                auto required = rxu::to_vector({
                    on.next(410, 6),
                    on.next(520, 7),
                    on.completed(630)
                });
                auto actual = results2.get_observer().messages();
                REQUIRE(required == actual);
            }

            THEN("result3 contains expected messages"){
                auto required = rxu::to_vector({
                    on.completed(900)
                });
                auto actual = results3.get_observer().messages();
                REQUIRE(required == actual);
            }

        }
    }
}


SCENARIO("subject - on_error in source", "[subject][subjects]"){
    GIVEN("a subject and a source with an error"){

        auto sc = rxsc::make_test();
        auto w = sc.create_worker();
        const rxsc::test::messages<int> on;

        std::runtime_error ex("subject on_error in stream");

        auto xs = sc.make_hot_observable({
            on.next(70, 1),
            on.next(110, 2),
            on.next(220, 3),
            on.next(270, 4),
            on.next(340, 5),
            on.next(410, 6),
            on.next(520, 7),
            on.error(630, ex),
            on.next(640, 9),
            on.completed(650),
            on.error(660, std::runtime_error("error on unsubscribed stream"))
        });

        rxsub::subject<int> s;

        auto results1 = w.make_subscriber<int>();

        auto results2 = w.make_subscriber<int>();

        auto results3 = w.make_subscriber<int>();

        WHEN("multicasting an infinite source"){

            auto o = s.get_subscriber();

            w.schedule_absolute(100, [&s, &o](const rxsc::schedulable&){
                s = rxsub::subject<int>(); o = s.get_subscriber();});
            w.schedule_absolute(200, [&xs, &o](const rxsc::schedulable&){
                xs.subscribe(o);});
            w.schedule_absolute(1000, [&o](const rxsc::schedulable&){
                o.unsubscribe();});

            w.schedule_absolute(300, [&s, &results1](const rxsc::schedulable&){
                s.get_observable().subscribe(results1);});
            w.schedule_absolute(400, [&s, &results2](const rxsc::schedulable&){
                s.get_observable().subscribe(results2);});
            w.schedule_absolute(900, [&s, &results3](const rxsc::schedulable&){
                s.get_observable().subscribe(results3);});

            w.schedule_absolute(600, [&results1](const rxsc::schedulable&){
                results1.unsubscribe();});
            w.schedule_absolute(700, [&results2](const rxsc::schedulable&){
                results2.unsubscribe();});
            w.schedule_absolute(800, [&results1](const rxsc::schedulable&){
                results1.unsubscribe();});
            w.schedule_absolute(950, [&results3](const rxsc::schedulable&){
                results3.unsubscribe();});

            w.start();

            THEN("result1 contains expected messages"){
                auto required = rxu::to_vector({
                    on.next(340, 5),
                    on.next(410, 6),
                    on.next(520, 7)
                });
                auto actual = results1.get_observer().messages();
                REQUIRE(required == actual);
            }

            THEN("result2 contains expected messages"){
                auto required = rxu::to_vector({
                    on.next(410, 6),
                    on.next(520, 7),
                    on.error(630, ex)
                });
                auto actual = results2.get_observer().messages();
                REQUIRE(required == actual);
            }

            THEN("result3 contains expected messages"){
                auto required = rxu::to_vector({
                    on.error(900, ex)
                });
                auto actual = results3.get_observer().messages();
                REQUIRE(required == actual);
            }

        }
    }
}