ECAP_Server/module/Local_Server/Data_Feed/Data_Feed.cpp

#include "Local_Server/Data_Feed/Data_Feed.h"
#include "../server/Global.h"
void Data_Feed_UDP_Server::handle_in_loop()  {
    svr.tick();
}
bool Data_Feed_UDP_Server::_open() {
    svr.set_bind_address("0.0.0.0", port);
    svr.create();

    server_logger->c_debug({}, {}, to_string() + "开启！");
    return true;
}
JSON Data_feed_Config::get_feed(const std::string& key) {
    auto ret = map.get(key);
    if (ret.has_value()) {
        auto r = ret.value().get();
        return r->to_json();
    }
    return {nullptr};
}

void Data_feed_Config::server(Global* g) {
    auto& svr = g->svr;
    auto& api = g->api;
    std::string name = "data_feed";


    svr.Post(api + "get_data_feed_state", [this](HTTP_Param) {
         CHECK_JSON_PARAM
         HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
         auto t = map.get(key);
         JSON ret{nullptr};
         if (t.has_value()) {
             ret = t.value()->get_state_json();
         }
         res->setBody(warp(ret).to_json_string());
     });



    svr.Post(api + "update_data_feed_config", [this](HTTP_Param) {
        CHECK_JSON_PARAM
        Get_J(packet_byte_size)
        Get_J(empty_wait_milliseconds)
        res->setBody(warp(to_json()).to_json_string());
        Global::save();
    });

    svr.Post(api + "get_data_feed_config", [this](HTTP_Param) {
        res->setBody(warp(to_json()).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 t = map.get(key);
        JSON ret{nullptr};
        if (t.has_value()) {
            auto dfs = dynamic_cast<Data_Feed_TCP_Server*>(t.value().get());
            if (dfs) {
                ret = dfs->get_clients_json();
            }
            auto dfs2 = dynamic_cast<Data_Feed_UDP_Server*>(t.value().get());
            if (dfs2) {
                ret = dfs2->get_clients_json();
            }
        }
        res->setBody(warp(ret).to_json_string());
    });

    svr.Post(api + svr.search + name, [this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(type, params.try_get_string("key"))
        res->setBody(warp(get_feed(type)).to_json_string());
    });



    svr.Post(api + svr.update + name, [g, this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
        HTTP_REQUIRE_VALUE(t, map.get(key))
        // 直接关闭
        t->close();
        t->from_json(&params);
        if (t->enable) {
            t->check_and_open();
        }
        Global::save();
        g->mode_acs.source_feed_relation_config.set_need_refresh();
    });

    svr.Post(api + svr.insert + name, [g, this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
        HTTP_REQUIRE_PTR(data, params.get("data"))
        HTTP_REQUIRE_VALUE(t, data->try_get_string("type"))
        auto df = create_data_feed_from_type(t);
        df->from_json(data);
        HTTP_REQUIRE_VALUE(enable, data->try_get_bool("enable"))
        if (enable)
        {
            df->check_and_open();
        }
        HTTP_REQUIRE_TRUE(map.insert(index, df), "index")
        res->setBody(warp(to_json()).to_json_string());
        Global::save();
        g->mode_acs.source_feed_relation_config.set_need_refresh();
    });

    svr.Post(api + svr.remove + name, [g, this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
        HTTP_REQUIRE_VALUE(df, map.try_get(index))
        df->close();
        HTTP_REQUIRE_TRUE(map.remove(index), "index")
        g->save();
        res->setBody(warp(to_json()).to_json_string());
        g->mode_acs.source_feed_relation_config.set_need_refresh();
    });

    svr.Post(api + svr.rise + name, [g, this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
        HTTP_REQUIRE_TRUE(map.swap(index, index - 1), "index")
        g->save();
        res->setBody(warp(to_json()).to_json_string());
        g->mode_acs.source_feed_relation_config.set_need_refresh();
    });
    svr.Post(api + svr.fall + name, [g, this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
        HTTP_REQUIRE_TRUE(map.swap(index, index + 1), "index")
        g->save();
        res->setBody(warp(to_json()).to_json_string());
        g->mode_acs.source_feed_relation_config.set_need_refresh();
    });
}

void BIN_Msg_Buffer::push(const std::string& msg)    {
    auto output_packet_size = Global::instance()->mode_acs.data_feed_config.packet_byte_size.load();
    std::lock_guard<std::mutex> g(mtx);
    auto& pool = Global::instance()->mode_acs.data_feed_config.pool_;
    if (msg_list_cache.empty() || msg_list_cache.back()->size() > output_packet_size) {
        auto t = pool.get();
        t->append(msg);
        msg_list_cache.push_back(t);
    } else {
        msg_list_cache.back()->append(msg);
    }
}
const std::vector<std::string *> &BIN_Msg_Buffer::get_all()  {
    std::lock_guard<std::mutex> g(mtx);
    std::swap(msg_list_cache, msg_list);
    msg_list_cache.clear();
    mode_s_msg_num = 0;
    mode_other_msg_num = 0;
    return msg_list;
}
BIN_Msg_Buffer::BIN_Msg_Buffer()  {
    msg_list_cache.reserve(8 * 1024);
    msg_list.reserve(8* 1024);
}
Psc::JSON BIN_Msg_Buffer::state_json()  {

    auto output_packet_size = Global::instance()->mode_acs.data_feed_config.packet_byte_size.load();
    size_t size, cache_size, pool_capacity, pool_free_count;
    {
        std::lock_guard<std::mutex> g(mtx);
        auto& pool = Global::instance()->mode_acs.data_feed_config.pool_;
        size = msg_list.size();
        cache_size = msg_list_cache.size();
        pool_capacity = pool.capacity();
        pool_free_count = pool.free_count();
    }
    return VAR_JSON_7(mode_s_msg_num, mode_other_msg_num, output_packet_size, cache_size, size, pool_capacity, pool_free_count);
}