#include "cyber/common/global_data.h"
#include "cyber/common/log.h"
#include "cyber/init.h"
#include "cyber/mainboard/module_argument.h"         // "mainboard.cc"引用
#include "cyber/mainboard/module_controller.h"       // "mainboard.cc"引用
#include "cyber/state.h"

#include "gflags/gflags.h"

int main(int argc, char** argv) {
  google::SetUsageMessage("we use this program to load dag and run user apps.");

  // 解析模块参数
  // parse the argument
  ModuleArgument module_args;
  module_args.ParseArgument(argc, argv);

  // 初始化cyber
  // initialize cyber
  apollo::cyber::Init(argv[0]);

  // 启动模块
  // start module
  ModuleController controller(module_args);
  if (!controller.Init()) {
    controller.Clear();
    AERROR << "module start error.";
    return -1;
  }

  // 等待cyber关闭
  apollo::cyber::WaitForShutdown();
  controller.Clear();
  AINFO << "exit mainboard.";

  return 0;
}

void ModuleArgument::ParseArgument(const int argc, char* const argv[]) {
  
  // 解析输入参数
  GetOptions(argc, argv);
  
  ...
  
  // 设置执行组，类似linux的cgroups
  GlobalData::Instance()->SetProcessGroup(process_group_);
  // 设置调度器名称
  GlobalData::Instance()->SetSchedName(sched_name_);
  // 打印模块的信息：名称，组，DAG配置
  AINFO << "binary_name_ is " << binary_name_ << ", process_group_ is "
        << process_group_ << ", has " << dag_conf_list_.size() << " dag conf";
  // 打印所有模块的依赖关系
  for (std::string& dag : dag_conf_list_) {
    AINFO << "dag_conf: " << dag;
  }
}

bool Init(const char* binary_name) {
  // 获取锁，为了改变state状态而获取锁
  std::lock_guard<std::mutex> lg(g_mutex);
  // 如果已经初始化，则返回失败
  if (GetState() != STATE_UNINITIALIZED) {
    return false;
  }

  // 初始化日志，并且把打印日志线程放入调度器
  InitLogger(binary_name);
  auto thread = const_cast<std::thread*>(async_logger->LogThread());
  scheduler::Instance()->SetInnerThreadAttr("async_log", thread);
  std::signal(SIGINT, OnShutdown);
  
  // 注册退出句柄ExitHandle，调用Clear()函数执行
  // Register exit handlers
  if (!g_atexit_registered) {
    if (std::atexit(ExitHandle) != 0) {
      AERROR << "Register exit handle failed";
      return false;
    }
    AINFO << "Register exit handle succ.";
    g_atexit_registered = true;
  }
  // 设置状态为已经初始化
  SetState(STATE_INITIALIZED);
  return true;
}

// 1. 构造ModuleController

// 2. ModuleController控制器初始化

  // 动态库是否已经加载
  bool IsLibraryLoaded();
  // 加载动态库
  bool LoadLibrary();
  // 卸载动态库
  int UnloadLibrary();
  // 获取动态库路径
  const std::string GetLibraryPath() const;
  // 获取有效的类名称 
  std::vector<std::string> GetValidClassNames();
  // 创建对象
  std::shared_ptr<Base> CreateClassObj(const std::string& class_name);
  // 类是否有效
  bool IsClassValid(const std::string& class_name);

template <typename Base>
std::vector<std::string> ClassLoader::GetValidClassNames() {
  return (utility::GetValidClassNames<Base>(this));
}

ClassLoader::ClassLoader(const std::string& library_path)
    : library_path_(library_path),
      loadlib_ref_count_(0),
      classobj_ref_count_(0) {
  LoadLibrary();
}

int ClassLoader::UnloadLibrary() {
  std::lock_guard<std::mutex> lckLib(loadlib_ref_count_mutex_);
  std::lock_guard<std::mutex> lckObj(classobj_ref_count_mutex_);

  if (classobj_ref_count_ > 0) {
    AINFO << "There are still classobjs have not been deleted, "
             "classobj_ref_count_: "
          << classobj_ref_count_;
  } else {
    --loadlib_ref_count_;
    // 卸载动态库
    if (loadlib_ref_count_ == 0) {
      utility::UnloadLibrary(library_path_, this);
    } else {
      if (loadlib_ref_count_ < 0) {
        loadlib_ref_count_ = 0;
      }
    }
  }
  return loadlib_ref_count_;
}

  switch (mode) {
    case OptionalMode::INTRA:
      transmitter = std::make_shared<IntraTransmitter<M>>(modified_attr);
      break;

    case OptionalMode::SHM:
      transmitter = std::make_shared<ShmTransmitter<M>>(modified_attr);
      break;

    case OptionalMode::RTPS:
      transmitter =
          std::make_shared<RtpsTransmitter<M>>(modified_attr, participant());
      break;

    default:
      transmitter =
          std::make_shared<HybridTransmitter<M>>(modified_attr, participant());
      break;
  }

message ShmMulticastLocator {
    optional string ip = 1;
    optional uint32 port = 2;
};

message ShmConf {
    optional string notifier_type = 1;
    optional ShmMulticastLocator shm_locator = 2;
};

// 1. 
void Participant::CreateFastRtpsParticipant(
    const std::string& name, int send_port,
    eprosima::fastrtps::ParticipantListener* listener) {
  uint32_t domain_id = 80;

  const char* val = ::getenv("CYBER_DOMAIN_ID");
  if (val != nullptr) {
    try {
      domain_id = std::stoi(val);
    } catch (const std::exception& e) {
      AERROR << "convert domain_id error " << e.what();
      return;
    }
  }

  auto part_attr_conf = std::make_shared<proto::RtpsParticipantAttr>();
  auto& global_conf = common::GlobalData::Instance()->Config();
  if (global_conf.has_transport_conf() &&
      global_conf.transport_conf().has_participant_attr()) {
    part_attr_conf->CopyFrom(global_conf.transport_conf().participant_attr());
  }

  eprosima::fastrtps::ParticipantAttributes attr;
  attr.rtps.defaultSendPort = send_port;
  attr.rtps.port.domainIDGain =
      static_cast<uint16_t>(part_attr_conf->domain_id_gain());
  attr.rtps.port.portBase = static_cast<uint16_t>(part_attr_conf->port_base());
  attr.rtps.use_IP6_to_send = false;
  attr.rtps.builtin.use_SIMPLE_RTPSParticipantDiscoveryProtocol = true;
  attr.rtps.builtin.use_SIMPLE_EndpointDiscoveryProtocol = true;
  attr.rtps.builtin.m_simpleEDP.use_PublicationReaderANDSubscriptionWriter =
      true;
  attr.rtps.builtin.m_simpleEDP.use_PublicationWriterANDSubscriptionReader =
      true;
  attr.rtps.builtin.domainId = domain_id;
  attr.rtps.builtin.leaseDuration.seconds = part_attr_conf->lease_duration();
  attr.rtps.builtin.leaseDuration_announcementperiod.seconds =
      part_attr_conf->announcement_period();

  attr.rtps.setName(name.c_str());

  std::string ip_env("127.0.0.1");
  const char* ip_val = ::getenv("CYBER_IP");
  if (ip_val != nullptr) {
    ip_env = ip_val;
    if (ip_env.size() == 0) {
      AERROR << "invalid CYBER_IP (an empty string)";
      return;
    }
  }
  ADEBUG << "cyber ip: " << ip_env;

  eprosima::fastrtps::rtps::Locator_t locator;
  locator.port = 0;
  RETURN_IF(!locator.set_IP4_address(ip_env));

  locator.kind = LOCATOR_KIND_UDPv4;

  attr.rtps.defaultUnicastLocatorList.push_back(locator);
  attr.rtps.defaultOutLocatorList.push_back(locator);
  attr.rtps.builtin.metatrafficUnicastLocatorList.push_back(locator);

  locator.set_IP4_address(239, 255, 0, 1);
  attr.rtps.builtin.metatrafficMulticastLocatorList.push_back(locator);

  fastrtps_participant_ =
      eprosima::fastrtps::Domain::createParticipant(attr, listener);
  RETURN_IF_NULL(fastrtps_participant_);
  eprosima::fastrtps::Domain::registerType(fastrtps_participant_, &type_);
}


// 2.

bool Manager::StartDiscovery(RtpsParticipant* participant) {
  if (participant == nullptr) {
    return false;
  }
  if (is_discovery_started_.exchange(true)) {
    return true;
  }
  if (!CreatePublisher(participant) || !CreateSubscriber(participant)) {
    AERROR << "create publisher or subscriber failed.";
    StopDiscovery();
    return false;
  }
  return true;
}

ParticipantImpl::createPublisher

  notify_fd_ = socket(AF_INET, SOCK_DGRAM, 0);
  ssize_t nbytes =
      sendto(notify_fd_, info_str.c_str(), info_str.size(), 0,
             (struct sockaddr*)&notify_addr_, sizeof(notify_addr_));  

  listen_fd_ = socket(AF_INET, SOCK_DGRAM, 0);
  bind(listen_fd_, (struct sockaddr*)&listen_addr_, sizeof(listen_addr_)
  ssize_t nbytes = recvfrom(listen_fd_, buf, 32, 0, nullptr, nullptr);