#include "../server/Global.h"
#include "../server/Mode_Msg_Buffer.h"
#include <future>
#include <iostream>
long long get_current_milliseconds() {
    auto now = std::chrono::system_clock::now();
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch());
    return duration.count();
}
Frequency_Limit too_many_msg_limit;
Frequency_Limit flush_limit;



ucoro::awaitable<void> data_feed_thread_coro(std::atomic<bool>& running) {
    auto g = Global::instance();
    for (auto& source : g->mode_acs.data_source_config.map.list()) {
        if (source->enable) {
            source->open();
        }
    }
    for (auto& t : g->mode_acs.data_feed_config.map.list()) {
        if (t->enable) {
            auto ret = t->check_and_open();
            if (!ret) {
                std::cout << "[Data_feed_Config]  [" << t->key << "] 第一次打开失败！ 程序退出! " << std::endl;
            }
        }
    }
    auto& mode_acs = Global::instance()->mode_acs;
    auto& cfg = mode_acs.data_feed_config;
    auto& pool = cfg.pool_;
    auto t = cfg.empty_wait_milliseconds.load();
    auto& report_data_feed_msg_mum = mode_acs.report_data_feed_msg_mum;
    std::vector<std::shared_ptr<Data_Feed>> enable_feeds; // all_feed 保证声明周期仍然存在
    auto monitor_msg_live = mode_acs.monitor_msg_live.load();
    while (running.load(std::memory_order_acquire) == true) {
        bool all_empty = true;
        for (auto& source : mode_acs.data_source_config.map.list()) {
            if (!source->enable) continue;
            if (!source->is_open() && !source->open()) continue;


            co_await source->handle_in_loop_coro();
            auto mode_data = co_await source->read_coro();

            struct Process_Result {
                std::exception_ptr exception;
            };

            auto result = co_await ucoro::callback_awaitable<Process_Result>(
                [source, mode_data = std::move(mode_data)](auto done) mutable {
                    asio::post(
                        data_source_process_pool(),
                        [source, mode_data = std::move(mode_data), done = std::move(done)]() mutable {
                            try {
                                source->process_mode_acs_data(mode_data);
                                done(Process_Result{});
                            } catch (...) {
                                done(Process_Result{std::current_exception()});
                            }
                    });
                });

            if (result.exception) {
                std::rethrow_exception(result.exception);
            }

        }

        // 这两个每次循环都要重新计算 保证实时性
        // if (flush_limit.test()) {
        auto all_feed = cfg.map.list();
        enable_feeds.clear();
        for (auto& feed : all_feed) {
            if (!feed->enable) continue;
            enable_feeds.push_back(feed);
            // 执行特定的刷新逻辑  比如 tcp_server 检查自身的连接
            co_await feed->handle_in_loop_coro();
        }


        int m = static_cast<int>(enable_feeds.size());
        for (int j = 0; j < m; j++) {
            auto& feed = enable_feeds[j];
            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::shared_ptr<Msg>>& all = feed->msg_buffer.get_all();
            const std::vector<std::string*>& all = feed->msg_buffer.get_all();
            // 释放内存池


            Pool_Guard pg(&pool, all);

            auto num = report_data_feed_msg_mum.load();
            if (monitor_msg_live) {
                if (num != 0 && all.size() > num) {
                    if (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;
                    }
                }
            }
            if (all.empty()) continue;
            all_empty = false;
            auto n = all.size();



            for (auto i = 0; i < n; i++) {
                const auto& str = all[i];
                if (str->empty()) {
                    std::cout << "警告:未知原因:" << feed->key << "内有空的消息体" << std::endl;
                    continue;
                }
                co_await feed->send_coro(*str);
            }
        }
        if (t != 0) {
            if (all_empty) {
                std::this_thread::sleep_for(std::chrono::milliseconds(t * 2));
            } else {
                std::this_thread::sleep_for(std::chrono::milliseconds(t));
            }
        }
    }
    co_return;
}

void io_coro(std::atomic<bool>& running) {

    try {
        ucoro::sync_await(data_feed_thread_coro(running));
    } 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("");
    }

}