#include "../server/Global.h"
#include "../server/Mode_Msg_Buffer.h"
#include "io_coro.h"

concurrencpp::result<void> Coro::coro_thread() {
     co_await concurrencpp::resume_on(executor_);
        auto g = Global::instance();
        for (auto& feed : g->mode_acs.data_feed_config.map.list()) {
            if (feed->enable) {
                auto ret = feed->check_and_open();
                if (!ret) {
                    std::cout << "Data feed first open failed: " << feed->key << std::endl;
                }
            }
        }
        while (running.load(std::memory_order_acquire)) {
            auto sources = Global::instance()->mode_acs.data_source_config.map.list();
            for (auto& source : sources) {
                if (!source || !source->enable) {
                    continue;
                }
                if (task_is_running(source->loop_task)) {
                    continue;
                }
                if (!source->is_open() && !source->open()) {
                    continue;
                }
                source->loop_task = std::make_unique<concurrencpp::result<
                    void>>(data_source_loop_coro(source));
            }
            auto feeds = Global::instance()->mode_acs.data_feed_config.map.list();
            for (auto& feed : feeds) {
                if (!feed || !feed->enable) {
                    continue;
                }
                if (task_is_running(feed->loop_task)) {
                    continue;
                }
                auto ret = feed->check_and_open();
                if (!ret) {
                    continue;
                }
                feed->loop_task = std::make_unique<concurrencpp::result<void>>(data_feed_loop_coro(feed));
            }
            co_await concurrencpp::resume_on(executor_);
        }
        auto start = std::chrono::steady_clock::now();
        while (has_running_loop_tasks()) {
            if (std::chrono::steady_clock::now() - start > std::chrono::seconds(10)) {
                std::cerr << "data loop task stop timeout" << std::endl;
                break;
            }
            co_await concurrencpp::resume_on(executor_);
        }
        co_return;
}

void Coro::start()  {
    if (data_feed_thread_task) {
        return;
    }
    std::cout << "start_io_coro" << std::endl;
    running.store(true, std::memory_order_release);
    executor_ = runtime_.make_worker_thread_executor();
    data_feed_thread_task = std::make_unique<concurrencpp::result<void>>(coro_thread());
}


void Coro::stop()  {
    running.store(false, std::memory_order_release);
    if (!data_feed_thread_task) {
        return;
    }
    try {
        data_feed_thread_task->get();
    }
    catch (const std::exception& e) {
        std::cerr << "data_feed_thread_coro exception: " << e.what() << std::endl;
        Psc::fail_fast_core_dump("");
    }
    catch (...) {
        std::cerr << "data_feed_thread_coro unknown exception" << std::endl;
        Psc::fail_fast_core_dump("");
    }
    data_feed_thread_task.reset();
    executor_.reset();
}

bool Coro::task_is_running(std::unique_ptr<concurrencpp::result<void>>& task)  {
    if (!task) {
        return false;
    }
    if (task->status() == concurrencpp::result_status::idle) {
        return true;
    }
    task->get();
    task.reset();
    return false;
}
concurrencpp::result<void> Coro::data_source_loop_coro(std::shared_ptr<Data_Source> source)  {
    co_await concurrencpp::resume_on(executor_);
    while (running.load(std::memory_order_acquire)) {
        if (!source->enable || !source->registered()) {
            break;
        }
        if (!source->is_open() && !source->open()) {
            co_await concurrencpp::resume_on(executor_);
            continue;
        }
        co_await source->handle_in_loop_coro();
        auto mode_data = co_await source->read_coro();
        if (!running.load(std::memory_order_acquire) || !source->enable || !source->registered()) {
            break;
        }
        source->process_mode_acs_data(mode_data);
        co_await concurrencpp::resume_on(executor_);
    }
    co_return;
}
concurrencpp::result<void> Coro::data_feed_loop_coro(std::shared_ptr<Data_Feed> feed)  {
        co_await concurrencpp::resume_on(executor_);
        auto& mode_acs = Global::instance()->mode_acs;
        auto& cfg = mode_acs.data_feed_config;
        auto& pool = cfg.pool_;
        auto& report_data_feed_msg_mum = mode_acs.report_data_feed_msg_mum;
        while (running.load(std::memory_order_acquire)) {
            std::vector<std::string> messages;
            {
                if (!feed->enable || !feed->registered()) {
                    break;
                }
                co_await feed->handle_in_loop_coro();
                auto s_num = feed->msg_buffer.mode_s_msg_num.load();
                auto other_num = feed->msg_buffer.mode_other_msg_num.load();
                const std::vector<std::string*>& all = feed->msg_buffer.get_all();
                Pool_Guard pg(&pool, all);
                auto num = report_data_feed_msg_mum.load();
                auto monitor_msg_live = mode_acs.monitor_msg_live.load();
                if (monitor_msg_live && num != 0 && all.size() > num && too_many_msg_limit.test()) {
                    std::ostringstream oss;
                    oss << "feed_key:" << feed->key << "   ";
                    oss << "recv_s:" << s_num << "   ";
                    oss << "recv_other:" << other_num << "   ";
                    std::cout << oss.str() << std::endl;
                }
                messages.reserve(all.size());
                for (const auto* str : all) {
                    if (!str || str->empty()) {
                        std::cout << "empty feed message:" << feed->key << std::endl;
                        continue;
                    }
                    messages.emplace_back(*str);
                }
            }
            if (messages.empty()) {
                co_await concurrencpp::resume_on(executor_);
                continue;
            }
            if (!feed->enable || !feed->registered()) {
                break;
            }
            for (auto& msg : messages) {
                co_await feed->send_coro(msg);
            }
            co_await concurrencpp::resume_on(executor_);
        }
        co_return;
    }

bool Coro::has_running_loop_tasks()  {
    auto sources = Global::instance()->mode_acs.data_source_config.map.list();
    for (auto& source : sources) {
        if (source && task_is_running(source->loop_task)) {
            return true;
        }
    }
    auto feeds = Global::instance()->mode_acs.data_feed_config.map.list();
    for (auto& feed : feeds) {
        if (feed && task_is_running(feed->loop_task)) {
            return true;
        }
    }
    return false;
}


