#include "MessageManager.h"

MessageManager::MessageManager() {
    m_thread = std::thread(&MessageManager::run, this);
}

MessageManager::~MessageManager() {
    m_exit = true;
    if (m_thread.joinable()) m_thread.join();
}

void MessageManager::run() {
    LOG(trace) << "message runner[" << std::this_thread::get_id() << "] start...";
    while (!m_exit) {
        if (m_params.empty()) {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            continue;
        }
        auto &[topic, args] = m_params.front();
        callExecutor(topic, std::move(args));
        m_params.pop();
    }
    LOG(trace) << "message runner exit...";
}

void MessageManager::callExecutor(const std::string &signature, std::any &&parameter) {
    auto executors = m_executors.equal_range(signature);
    for (auto executor = executors.first; executor != executors.second; ++executor) {
        try {
            if (!executor->second(parameter)) {
                LOG(warning) << "topic: " << signature << " execute failed, parameter cast failed.";
            }
        } catch (const std::exception &e) {
            LOG(error) << e.what();
        }
    }
}