Terminal_process_CT/dispatch/dispatch_communicator.cpp

//
// Created by zx on 2020/7/21.
//

#include "dispatch_communicator.h"
#include "uniq_key.h"

Dispatch_communicator::~Dispatch_communicator(){}

Error_manager Dispatch_communicator::dispatch_request(message::Dispatch_request_msg& request,
        message::Dispatch_response_msg& result,Thread_condition& cancel_condition)
{

    /*
      * 检查request合法性,以及模块状态
      */

    if(request.base_info().sender()!=message::eMain||request.base_info().receiver()!=message::eDispatch)
        return Error_manager(ERROR,MINOR_ERROR,"dispatch request invalid");

    if(m_response_table.find(request.command_key())==true)
        return Error_manager(ERROR,MAJOR_ERROR," dispatch request repeated");
    //设置超时,若没有设置,默认300000  五分钟
    int timeout=request.base_info().has_timeout_ms()?request.base_info().timeout_ms():300000;
    //向测量节点发送测量请求,并记录请求

    Error_manager code;
    Communication_message message;

    message::Base_info base_msg;
    base_msg.set_msg_type(message::eDispatch_request_msg);
    base_msg.set_sender(message::eMain);
    base_msg.set_receiver(message::eDispatch);
    base_msg.set_timeout_ms(timeout);
    message.reset(base_msg,request.SerializeAsString());

    code=encapsulate_msg(&message);
    if(code!=SUCCESS)
        return code;
    m_response_table[request.command_key()]=message::Dispatch_response_msg();
    //循环查询请求是否被处理
    auto start_time=std::chrono::system_clock::now();
    double time=0;
    do{
        //查询到记录
        message::Dispatch_response_msg response;
        ///查询是否存在,并且删除该记录,
        if(m_response_table.find(request.command_key(),response))
        {
            //判断是否接收到回应,若回应信息被赋值则证明有回应
            if (response.has_base_info() && response.has_command_key())
            {
                message::Base_info response_base = response.base_info();
                //检查类型是否匹配
                if (response_base.msg_type() != message::eDispatch_response_msg) {
                    return Error_manager(ERROR, CRITICAL_ERROR,
                                         "dispatch response basemsg type error");
                }
                //检查基本信息是否匹配
                if (response_base.sender() != message::eDispatch ||
                    response_base.receiver() != message::eMain ||
                    response.command_key() != request.command_key()) {
                    return Error_manager(ERROR, MAJOR_ERROR,
                                         "dispatch response basemsg info error");
                }
                result = response;
                m_response_table.erase(request.command_key());
                return SUCCESS;

            }
        }
        else
        {
            //未查询到记录,任务已经被提前取消,记录被删除
            return Error_manager(TASK_CANCEL,MINOR_ERROR,"dispatch request canceled");
        }

        auto end_time=std::chrono::system_clock::now();
        auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);
        time=1000.0*double(duration.count()) * std::chrono::microseconds::period::num / std::chrono::microseconds::period::den;
        std::this_thread::yield();
        if(time>double(timeout))
        {
            m_response_table.erase(request.command_key());
            return Error_manager(RESPONSE_TIMEOUT,MINOR_ERROR,"dispatch request timeout");
        }
    }while(cancel_condition.wait_for_ex(std::chrono::milliseconds(1))==false);

    m_response_table.erase(request.command_key());
    return Error_manager(TASK_CANCEL,MINOR_ERROR,"dispatch request task canceled");

}
/*
 * 提前取消请求
 */
Error_manager Dispatch_communicator::cancel_request(message::Dispatch_request_msg& request)
{
    message::Dispatch_response_msg t_response;
    if(m_response_table.find(request.command_key(),t_response))
    {
        m_response_table.erase(request.command_key());
    }
    return SUCCESS;
}


Error_manager Dispatch_communicator::check_entrance_statu(int terminal_id)
{
    if(m_storing_dispatch_statu_msg_map.find(terminal_id)==false || m_storing_dispatch_statu_time_map.find(terminal_id)==false)
        return Error_manager(FAILED,MINOR_ERROR,"停车口调度模块状态未找到");

    std::chrono::system_clock::time_point time_now=std::chrono::system_clock::now();
    auto durantion=time_now-m_storing_dispatch_statu_time_map[terminal_id];
    if(m_storing_dispatch_statu_msg_map[terminal_id].has_base_info()== false
       || durantion>std::chrono::seconds(5))
    {
        return Error_manager(ERROR,MINOR_ERROR,"调度节点通讯断开");
    }
    if(m_storing_dispatch_statu_msg_map[terminal_id].terminal_status()==message::E_TERMINAL_FAULT)
    {
        return Error_manager(ERROR,MINOR_ERROR,"调度节点故障");
    }
    if(m_storing_dispatch_statu_msg_map[terminal_id].terminal_status()==message::E_TERMINAL_UNKNOW)
    {
        return Error_manager(ERROR,MINOR_ERROR,"调度节点状态未知");
    }

    return SUCCESS;

}

Error_manager Dispatch_communicator::check_export_statu(int terminal_id)
{
    if(m_picking_dispatch_statu_msg_map.find(terminal_id)==false || m_picking_dispatch_statu_time_map.find(terminal_id)==false)
        return Error_manager(FAILED,MINOR_ERROR,"停车口调度模块状态未找到");

    std::chrono::system_clock::time_point time_now=std::chrono::system_clock::now();
    auto durantion=time_now-m_picking_dispatch_statu_time_map[terminal_id];
    if(m_picking_dispatch_statu_msg_map[terminal_id].has_base_info()== false
       || durantion>std::chrono::seconds(5))
    {
        return Error_manager(ERROR,MINOR_ERROR,"调度节点通讯断开");
    }
    if(m_picking_dispatch_statu_msg_map[terminal_id].terminal_status()==message::E_TERMINAL_FAULT)
    {
        return Error_manager(ERROR,MINOR_ERROR,"调度节点故障");
    }
    if(m_picking_dispatch_statu_msg_map[terminal_id].terminal_status()==message::E_TERMINAL_UNKNOW)
    {
        return Error_manager(ERROR,MINOR_ERROR,"调度节点状态未知");
    }

    return SUCCESS;
}

Dispatch_communicator::Dispatch_communicator()
{
}
Error_manager Dispatch_communicator::encapsulate_msg(Communication_message* message)
{
    Error_manager code;
    //记录请求
    switch (message->get_message_type())
    {
        case Communication_message::eDispatch_request_msg:
        {
            message::Dispatch_request_msg request;
            if(false==request.ParseFromString(message->get_message_buf()))
            {
                code=Error_manager(ERROR,CRITICAL_ERROR,"request message parse failed");
            }

            //发送请求
            code= Communication_socket_base::encapsulate_msg(message);

            break;
        }
        default:
            code= Error_manager(FAILED,CRITICAL_ERROR," measure发送任务类型不存在");
            break;
    }
    return code;
}
Error_manager Dispatch_communicator::execute_msg(Communication_message* p_msg)
{
    if(p_msg== nullptr)
        return Error_manager(POINTER_IS_NULL,CRITICAL_ERROR,"dispatch response msg pointer is null");


    //测量response消息
    switch (p_msg->get_message_type())
    {
        ///测量结果反馈消息
        case Communication_message::eDispatch_response_msg:
        {
            message::Dispatch_response_msg response;
            response.ParseFromString(p_msg->get_message_buf());
            ///查询请求表是否存在,并且更新
            if(m_response_table.find_update(response.command_key(),response)==false)
            {
                return Error_manager(ERROR,NEGLIGIBLE_ERROR,"dispatch response without request");
            }

            break;
        }
            ///测量系统状态
        case Communication_message::eDispatch_status_msg:
        {
            message::Dispatch_terminal_status_msg statu_msg;
            if(statu_msg.ParseFromString(p_msg->get_message_buf())==false)
            {
                return Error_manager(ERROR,CRITICAL_ERROR,"调度模块状态消息解析失败");
            }
            if(statu_msg.passageway_direction()==message::E_INLET)
            {
                m_storing_dispatch_statu_msg_map[statu_msg.terminal_id()]=statu_msg;
                m_storing_dispatch_statu_time_map[statu_msg.terminal_id()]=std::chrono::system_clock::now();
            }
            if(statu_msg.passageway_direction()==message::E_OUTLET)
            {
                m_picking_dispatch_statu_msg_map[statu_msg.terminal_id()]=statu_msg;
                m_picking_dispatch_statu_time_map[statu_msg.terminal_id()]=std::chrono::system_clock::now();
            }
            if(statu_msg.passageway_direction()==message::E_BILATERAL)
            {
                m_storing_dispatch_statu_msg_map[statu_msg.terminal_id()]=statu_msg;
                m_storing_dispatch_statu_time_map[statu_msg.terminal_id()]=std::chrono::system_clock::now();
                m_picking_dispatch_statu_msg_map[statu_msg.terminal_id()]=statu_msg;
                m_picking_dispatch_statu_time_map[statu_msg.terminal_id()]=std::chrono::system_clock::now();
            }
            break;
        }
    }
    return SUCCESS;
}
/*
 * 检测消息是否可被处理
 */
Error_manager Dispatch_communicator::check_msg(Communication_message* p_msg)
{
    //通过 p_msg->get_message_type() 和 p_msg->get_receiver() 判断这条消息是不是给我的.
    //子类重载时, 增加自己模块的判断逻辑, 以后再写.
    if ( (p_msg->get_message_type() == Communication_message::Message_type::eDispatch_response_msg
          ||p_msg->get_message_type() == Communication_message::Message_type::eDispatch_status_msg)
         && p_msg->get_receiver() == Communication_message::Communicator::eMain )
    {
        return Error_code::SUCCESS;
    }
    else
    {
        //认为接受人
        return Error_code::INVALID_MESSAGE;
    }
}
/*
 * 心跳发送函数,重载
 */
Error_manager Dispatch_communicator::encapsulate_send_data()
{
    return SUCCESS;
}
//检查消息是否可以被解析, 需要重载
Error_manager Dispatch_communicator::check_executer(Communication_message* p_msg)
{
    return SUCCESS;
}