ECAP_Server/module/Local_Server/Data_Feed/Data_Feed.h

#pragma once
#include "Local_Server/server/With_Loop_Coro.h"
#include "global.h"
#include <iostream>
enum class Output_Data_Format { AVR, AVR_MLAT, BIN, SBS, BIN_ID };
enum class Mode_S_Output_Type { DF_11_17_18, NO_POS_Mode_S, ALL_Mode_S };
struct Output_Format {
	Output_Data_Format type = Output_Data_Format::BIN;
	Psc::Copyable_Atomic<bool> use_status = false;
	Mode_S_Output_Type mode_s_output_type = Mode_S_Output_Type::DF_11_17_18;
	Psc::Copyable_Atomic<bool> use_mode_ac = false;
	Psc::Copyable_Atomic<bool> sbs_only_pos = false;
	PSC_USE_JSON
};
class Data_Feed : public With_Loop_Coro {
public:
	BIN_Msg_Buffer msg_buffer{};
	Output_Format output_format;
	Frequency_Limit_Multi sbs_flm{};
	asio::awaitable<void> loop_coro() final;
	virtual void handle_in_loop() {}

	virtual asio::awaitable<void> handle_in_loop_coro() {
		handle_in_loop();
		co_return;
	}

	virtual asio::awaitable<void> send_coro(const std::string&) {
		co_return;
	}

	virtual Psc::JSON to_json() {
		Psc::JSON ret = With_Loop_Coro::to_base_json();
		ret.append(Psc::JSON("output_format", output_format.to_base_json()));
		return ret;
	}

	virtual void from_json(const Psc::JSON* that_json) {
		With_Loop_Coro::from_base_json(that_json);
		output_format.from_base_json(that_json->get("output_format"));
	}

	virtual Psc::JSON get_custom_state_json() {
		return Psc::JSON::object();
	}

	virtual Psc::JSON get_state_json() {
		Psc::JSON ret = msg_buffer.state_json();
		ret += get_custom_state_json();
		return ret;
	}

	std::string to_string() {
		return "Data_Feed{" + VAR_STR_3(key, enable, type) + "}";
	}

	bool registered();
protected:
	Data_Feed() = default;
	Frequency_Limit too_many_msg_limit;
};
class Data_Feed_TCP_Server_Data {
public:
	std::uint16_t port{};
	size_t connect_user_buffer_size{};
	size_t connect_system_buffer_size{};
	PSC_USE_JSON
};
class Data_Feed_TCP_Server : public Data_Feed, public Data_Feed_TCP_Server_Data {
public:
	Psc::asio_socket::TCP_Server_Coro svr;

	Psc::JSON get_custom_state_json() override {
		auto& state = svr.state;
		Psc::JSON ret = Psc::JSON::object();
		Ret_J(connect_system_buffer_size)
		Ret_J(connect_user_buffer_size)
		ret.key = "fixed";
		auto t = VAR_JSON_1(state);
		t.append(ret);
		return t;
	}

	~Data_Feed_TCP_Server() override = default;

	asio::awaitable<void> handle_in_loop_coro() override {
		co_await svr.flush_clients_coro();
		co_await svr.tick_coro();
		co_return;
	}

	asio::awaitable<void> send_coro(const std::string& data) override {
		co_await svr.write_to_all_clients_coro(data);
		co_return;
	}

	void from_json(const Psc::JSON* that_json) override {
		Data_Feed::from_json(that_json);
		Data_Feed_TCP_Server_Data::from_base_json(that_json);
	}

	Psc::JSON to_json() override {
		Psc::JSON ret = Data_Feed::to_json();
		ret += Data_Feed_TCP_Server_Data::to_base_json();
		return ret;
	}

	asio::awaitable<void> _close() override {
		co_await svr.close_coro();
		co_return;
	}

	asio::awaitable<void> _open() override {
		svr.set_connect_user_buffer_size(connect_user_buffer_size);
		svr.set_connect_system_buffer_size(connect_system_buffer_size);
		svr.set_tcp_no_delay(false);
		svr.create();
		co_await svr.listen_coro("0.0.0.0", port);
		co_return;
	}

	Psc::JSON get_clients_json() {
		Psc::JSON ret = Psc::JSON::array();
		for (const auto& conn : svr.get_all_clients()) {
			auto& info = conn->info;
			Psc::JSON cur = Psc::JSON::object();
			auto& fd = info.fd;
			auto& cur_ip = info.sockaddr.ip;
			auto& cur_port = info.sockaddr.port;
			cur.append({"base", VAR_STR_3(cur_ip, cur_port, fd)});
			cur.append({"state", conn->send_buffer.state_str()});
			{
				auto& ins = conn->push_speed.instant_speed;
				auto& avr = conn->push_speed.average_speed;
				cur.append({"send_speed(KB/s)", VAR_STR_2(ins, avr)});
			}
			{
				auto& ins = conn->lose_speed.instant_speed;
				auto& avr = conn->lose_speed.average_speed;
				cur.append({"lose_speed(KB/s)", VAR_STR_2(ins, avr)});
			}
			{
				auto& ins = conn->send_num.instant;
				auto& avr = conn->send_num.average;
				cur.append({"send_num(byte/count)", VAR_STR_2(ins, avr)});
			}
			ret.append(cur);
		}
		return ret;
	}
};
class Data_Feed_TCP_Client_Data {
public:
	std::string url;
	std::uint16_t port{};
	PSC_USE_JSON
};
class Data_Feed_TCP_Client : public Data_Feed, public Data_Feed_TCP_Client_Data {
public:
	asio::awaitable<void> handle_in_loop_coro() override {
		co_await cli.tick_coro();
		co_return;
	}

	asio::awaitable<void> send_coro(const std::string& data) override {
		co_await cli.send_coro(data);
		co_return;
	}

	std::string url;

	Psc::JSON get_custom_state_json() override {
		auto& state = cli.state;
		return VAR_JSON_1(state);
	}

	std::uint16_t port{};
	Psc::asio_socket::TCP_Client_Coro cli;
	~Data_Feed_TCP_Client() override = default;

	asio::awaitable<void> _close() override {
		co_await cli.close_coro();
		co_return;
	}

	asio::awaitable<void> _open() override {
		Psc::asio_socket::Sockaddr_In sockaddr_in;
		sockaddr_in.ip = url;
		sockaddr_in.port = port;
		cli.set_dest_address(sockaddr_in);
		cli.create();
		co_await cli.connect_coro();
		co_return;
	}

	void from_json(const Psc::JSON* that_json) override {
		Data_Feed::from_json(that_json);
		Data_Feed_TCP_Client_Data::from_base_json(that_json);
	}

	Psc::JSON to_json() override {
		return Data_Feed::to_json() += Data_Feed_TCP_Client_Data::to_base_json();
	}
};
class Data_Feed_UDP_Server_Data {
public:
	std::uint16_t port{};
	PSC_USE_JSON
};
class Data_Feed_UDP_Server : public Data_Feed, public Data_Feed_UDP_Server_Data {
public:
	Psc::JSON get_custom_state_json() override {
		auto& state = svr.state;
		return VAR_JSON_1(state);
	}

	Data_Feed_UDP_Server() = default;
	~Data_Feed_UDP_Server() override = default;
	asio::awaitable<void> handle_in_loop_coro() override;
	asio::awaitable<void> send_coro(const std::string& data) override;

	[[nodiscard]] Psc::JSON get_clients_json() const {
		Psc::JSON ret = Psc::JSON::array();
		for (const auto& i : svr.clients) {
			Psc::JSON cur = Psc::JSON::object();
			cur.append({"ip", i.ip});
			cur.append({"port", i.port});
			ret.append(cur);
		}
		return ret;
	}

	void from_json(const Psc::JSON* that_json) override {
		Data_Feed::from_json(that_json);
		Data_Feed_UDP_Server_Data::from_base_json(that_json);
	}

	Psc::JSON to_json() override {
		return Data_Feed::to_json() += Data_Feed_UDP_Server_Data::to_base_json();
	}

	asio::awaitable<void> _close() override {
		co_await svr.close_coro();
		co_return;
	}

	asio::awaitable<void> _open() override;
	Psc::asio_socket::UDP_Server_Coro svr;
};
class Data_Feed_UDP_Client_Data {
public:
	std::uint16_t port{};
	std::string url{};
	PSC_USE_JSON
};
class Data_Feed_UDP_Client : public Data_Feed, public Data_Feed_UDP_Client_Data {
public:
	Data_Feed_UDP_Client() {}
	~Data_Feed_UDP_Client() override {}

	Psc::JSON get_custom_state_json() override {
		auto& state = cli.state;
		return VAR_JSON_1(state);
	}

	asio::awaitable<void> handle_in_loop_coro() override {
		co_await cli.tick_coro();
		co_return;
	}

	asio::awaitable<void> send_coro(const std::string& data) override {
		co_await cli.send_coro(data);
		co_return;
	}

	Psc::asio_socket::UDP_Client_Coro cli;

	void from_json(const Psc::JSON* that_json) override {
		Data_Feed::from_json(that_json);
		Data_Feed_UDP_Client_Data::from_base_json(that_json);
	}

	Psc::JSON to_json() override {
		return Data_Feed::to_json() += Data_Feed_UDP_Client_Data::to_base_json();
	}

	asio::awaitable<void> _close() override {
		std::cout << "udp client target address:" << cli.dest_address.to_string()
		<< " closed" << std::endl;
		co_await cli.close_coro();
		co_return;
	}

	asio::awaitable<void> _open() override {
		cli.create();
		cli.set_dest_address(url, port);
		co_await cli.connect_coro();
		co_return;
	}
};

inline std::shared_ptr<Data_Feed> create_data_feed_from_type(const std::string& t) {
	std::shared_ptr<Data_Feed> ret{};
	if (t == "Data_Feed_TCP_Server")
		ret = std::make_shared<Data_Feed_TCP_Server>();
	else if (t == "Data_Feed_UDP_Server")
		ret = std::make_shared<Data_Feed_UDP_Server>();
	else if (t == "Data_Feed_TCP_Client")
		ret = std::make_shared<Data_Feed_TCP_Client>();
	else if (t == "Data_Feed_UDP_Client")
		ret = std::make_shared<Data_Feed_UDP_Client>();
	else {
		std::ostringstream oss;
		oss << "unknown Data_Feed type: " << t << "   " << LOG_POS_SIMPLE
		<< std::endl;
		std::cout << oss.str();
		throw std::invalid_argument(oss.str());
	}
	return ret;
}

class Data_feed_Config_Data {
public:
	Psc::Copyable_Atomic<std::uint16_t> empty_wait_milliseconds{};
	Psc::Copyable_Atomic<std::uint16_t> packet_byte_size{};
	size_t tcp_server_default_connect_user_buffer_size{};
	size_t tcp_server_default_connect_system_buffer_size{};
	PSC_USE_JSON
};
class Data_feed_Config : public Data_feed_Config_Data {
public:
	Psc::String_Pool pool_ =
	Psc::String_Pool(1600, 8 * 1024); // 8192 * 1600  12.5 MB
	Ordered_Map<std::string, std::shared_ptr<Data_Feed>> map;
	Data_feed_Config() = default;

	void init(const Psc::JSON* that_json) {
		auto connect = that_json->get("list");
		for (auto& it : connect->children) {
			auto type = it.get_string("type");
			std::shared_ptr<Data_Feed> t = create_data_feed_from_type(type);
			t->from_json(&it);
			map.push_back(t);
		}
		Data_feed_Config_Data::from_base_json(that_json);
	}

	Psc::JSON list() {
		auto connect_json = Psc::JSON::array();
		for (auto& it : map.list()) {
			connect_json.children.emplace_back(it->to_json());
		}
		return connect_json;
	}

	Psc::JSON to_json() {
		Psc::JSON ret = Data_feed_Config_Data::to_base_json();
		ret.append({"list", list()});
		return ret;
	}

	Psc::JSON get_feed(const std::string& key);
	void server(Global* g);
};