ECAP_Server/module/Local_Server/Data_Feed/Data_Feed.h

#pragma once

#include "Local_Server/server/With_Loop_Coro.h"
#include "global.h"

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;
  std::atomic<bool> use_status = false;
  Mode_S_Output_Type mode_s_output_type = Mode_S_Output_Type::DF_11_17_18;
  std::atomic<bool> use_mode_ac = false;
  std::atomic<bool> sbs_only_pos = false;
  std::atomic<bool> use_crc = false;

  [[nodiscard]] Psc::JSON to_Json() const {
    Psc::JSON ret = Psc::JSON::object();
    Ret_J(type) Ret_J(use_status) Ret_J(mode_s_output_type) Ret_J(use_mode_ac)
        Ret_J(sbs_only_pos) return ret;
  }

  void from_json(const Psc::JSON *that_json) {
    Get_J(type) Get_J(use_status) Get_J(mode_s_output_type) Get_J(use_mode_ac)
        Get_J(sbs_only_pos)
  }
};

class Data_Feed : public With_Loop_Coro {
public:
  BIN_Msg_Buffer msg_buffer{};
  std::string type;
  Output_Format output_format;
  Frequency_Limit_Multi sbs_flm{};
  concurrencpp::result<void> loop_coro(
      std::shared_ptr<concurrencpp::worker_thread_executor> executor) override;
  virtual void handle_in_loop() {}
  virtual concurrencpp::result<void> handle_in_loop_coro() {
    handle_in_loop();
    co_return;
  }
  virtual concurrencpp::result<void> send_coro(const std::string &) {
    co_return;
  }
  virtual Psc::JSON to_json() {
    Psc::JSON ret = Psc::JSON::object();
    Ret_J(key);
    Ret_J(enable);
    Ret_J(type);
    ret.append({"output_format", output_format.to_Json()});
    return ret;
  }
  virtual void from_json(const Psc::JSON *that_json) {
    Get_J(key) Get_J(enable) Get_J(type)
        output_format.from_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.append_list(get_custom_state_json().children);
    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 : public Data_Feed {
public:
  std::uint16_t port{};
  size_t connect_user_buffer_size{};
  size_t connect_system_buffer_size{};
  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 {}
  concurrencpp::result<void> handle_in_loop_coro() override {
    co_await svr.flush_clients_coro();
    co_await svr.tick_coro();
    co_return;
  }
  concurrencpp::result<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);
    Get_J(port);
    Get_J(connect_user_buffer_size) Get_J(connect_system_buffer_size)
  }
  Psc::JSON to_json() override {
    Psc::JSON ret = Data_Feed::to_json();
    Ret_J(port);
    Ret_J(connect_user_buffer_size)
        Ret_J(connect_system_buffer_size) return ret;
  }
  concurrencpp::result<void> _close() override {
    co_await svr.close_coro();
    co_return;
  }
  concurrencpp::result<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 : public Data_Feed {
public:
  concurrencpp::result<void> handle_in_loop_coro() override {
    co_await cli.tick_coro();
    co_return;
  }
  concurrencpp::result<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;
  concurrencpp::result<void> _close() override {
    co_await cli.close_coro();
    co_return;
  }
  concurrencpp::result<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);
    Get_J(url) Get_J(port);
  }
  Psc::JSON to_json() override {
    Psc::JSON ret = Data_Feed::to_json();
    Ret_J(url);
    Ret_J(port);
    return ret;
  }
};

class Data_Feed_UDP_Server : public Data_Feed {
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;
  concurrencpp::result<void> handle_in_loop_coro() override;
  concurrencpp::result<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);
    Get_J(port);
  }
  Psc::JSON to_json() override {
    Psc::JSON ret = Data_Feed::to_json();
    Ret_J(port);
    return ret;
  }
  concurrencpp::result<void> _close() override {
    co_await svr.close_coro();
    co_return;
  }
  concurrencpp::result<void> _open() override;
  std::uint16_t port{};
  Psc::asio_socket::UDP_Server_Coro svr;
};

class Data_Feed_UDP_Client : public Data_Feed {
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);
  }
  concurrencpp::result<void> handle_in_loop_coro() override {
    co_await cli.tick_coro();
    co_return;
  }
  concurrencpp::result<void> send_coro(const std::string &data) override {
    co_await cli.send_coro(data);
    co_return;
  }
  Psc::asio_socket::UDP_Client_Coro cli;
  std::string url;
  std::uint16_t port{};
  void from_json(const Psc::JSON *that_json) override {
    Data_Feed::from_json(that_json);
    Get_J(url) Get_J(port);
  }
  Psc::JSON to_json() override {
    Psc::JSON ret = Data_Feed::to_json();
    Ret_J(url);
    Ret_J(port);
    return ret;
  }
  concurrencpp::result<void> _close() override {
    std::cout << "udp client target address:" << cli.dest_address.to_string()
              << " closed" << std::endl;
    co_await cli.close_coro();
    co_return;
  }
  concurrencpp::result<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;
}

struct Data_feed_Config {
  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;
  std::atomic<std::uint16_t> empty_wait_milliseconds{};
  std::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{};
  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);
    }
    Get_J(empty_wait_milliseconds)
        Get_J(tcp_server_default_connect_user_buffer_size)
            Get_J(tcp_server_default_connect_system_buffer_size)
  }

  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 = Psc::JSON::object();
    Ret_J(empty_wait_milliseconds) Ret_J(packet_byte_size)
        Ret_J(tcp_server_default_connect_user_buffer_size)
            Ret_J(tcp_server_default_connect_system_buffer_size)
                ret.append({"list", list()});
    return ret;
  }
  Psc::JSON get_feed(const std::string &key);
  void server(Global *g);
};
class Data_Source;