#include "Local_Server/Data_Feed/Data_Feed.h"
#include <string_view>
#include "../server/Global.h"
bool Data_Feed::registered() {
    auto all_feed = Global::instance()->mode_acs.data_feed_config.map.list();
    for (auto& item : all_feed) {
        if (item.get() == this) {
            return true;
        }
    }
    return false;
}
void Data_feed_Config::set_connections_need_refresh() {
    for (const auto& source : Global::instance()->mode_acs.data_source_config.map.list())
        source->need_refresh_data_feed_key_list = true;
}
bool Data_feed_Config::source_has_connection(std::string_view source_key) const {
    for (const auto& feed : map.list()) {
        if (feed->source_key == source_key)
            return true;
    }
    return false;
}
asio::awaitable<void> Data_Feed_UDP_Server::handle_in_loop_coro() {
    co_await svr.tick_coro();
    co_return;
}
asio::awaitable<void> Data_Feed_UDP_Server::send_coro(std::string_view data) {
    co_await svr.write_to_all_clients_coro(data);
    co_return;
}
asio::awaitable<void> Data_Feed_UDP_Server::_open() {
    const auto value = specific.read([](const auto& value) { return value; });
    svr.set_bind_address("0.0.0.0", value.port);
    svr.create();
    co_await svr.bind_coro();
    server_logger->c_debug({}, {}, to_string() + "开启！");
    co_return;
}
void Data_feed_Config::server(Global* g) {
    auto& svr = g->svr;
    auto& api = g->api;
    svr.Post(api + "get_data_feed_state", [this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
        auto value = map.get(key);
        JSON ret{nullptr};
        if (value.has_value())
            ret = value.value()->get_state_json();
        res->setBody(warp(ret).to_json_string());
    });
    svr.Post(api + "get_data_feed_server_connect", [this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
        auto value = map.get(key);
        JSON ret{nullptr};
        if (value.has_value()) {
            if (auto tcp = dynamic_cast<Data_Feed_TCP_Server*>(value.value().get()))
                ret = tcp->get_clients_json();
            if (auto udp = dynamic_cast<Data_Feed_UDP_Server*>(value.value().get()))
                ret = udp->get_clients_json();
        }
        res->setBody(warp(ret).to_json_string());
    });
}
void BIN_Msg_Buffer::push(std::string_view msg) {
    auto output_packet_size = Global::instance()->mode_acs.data_feed_config.settings.member<&Data_feed_Config_Data::packet_byte_size>().read([](const auto& value) { return value; });
    std::lock_guard<std::mutex> g(mtx);
    if (msg_list_cache.empty() || msg_list_cache.back().size() > output_packet_size) {
        msg_list_cache.emplace_back();
        msg_list_cache.back().append(msg);
    }
    else {
        msg_list_cache.back().append(msg);
    }
    cache_bytes += msg.size();
    high_water_messages = std::max(high_water_messages, msg_list_cache.size());
    high_water_bytes = std::max(high_water_bytes, cache_bytes);
}
BIN_Msg_Buffer::Batch BIN_Msg_Buffer::take_all() {
    Batch result(msg_list_cache.get_allocator().resource());
    std::lock_guard<std::mutex> g(mtx);
    result.swap(msg_list_cache);
    dispatch_batches = result.size();
    dispatch_bytes = cache_bytes;
    cache_bytes = 0;
    mode_s_msg_num = 0;
    mode_other_msg_num = 0;
    return result;
}
BIN_Msg_Buffer::BIN_Msg_Buffer() {
    msg_list_cache.reserve(8 * 1024);
}
Psc::JSON BIN_Msg_Buffer::state_json() {
    auto output_packet_size = Global::instance()->mode_acs.data_feed_config.settings.member<&Data_feed_Config_Data::packet_byte_size>().read([](const auto& value) { return value; });
    size_t cache_size, current_cache_bytes, current_dispatch_batches, current_dispatch_bytes, current_high_water_messages, current_high_water_bytes;
    {
        std::lock_guard<std::mutex> g(mtx);
        cache_size = msg_list_cache.size();
        current_cache_bytes = cache_bytes;
        current_dispatch_batches = dispatch_batches;
        current_dispatch_bytes = dispatch_bytes;
        current_high_water_messages = high_water_messages;
        current_high_water_bytes = high_water_bytes;
    }
    Psc::JSON ret = Psc::JSON::object();
    ret.append({"mode_s_pending", mode_s_msg_num.load()});
    ret.append({"mode_other_pending", mode_other_msg_num.load()});
    ret.append({"batch_target_bytes", output_packet_size});
    ret.append({"pending_batches", cache_size});
    ret.append({"pending_bytes", current_cache_bytes});
    ret.append({"last_dispatch_batches", current_dispatch_batches});
    ret.append({"last_dispatch_bytes", current_dispatch_bytes});
    ret.append({"high_water_batches", current_high_water_messages});
    ret.append({"high_water_bytes", current_high_water_bytes});
    ret.append({"allocator", "std::pmr/mimalloc"});
    return ret;
}