Terminal_process_CT/test/Terminal_communication.cpp

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

#include "Terminal_communication.h"
#include "process_message.pb.h"
#include <condition_variable>

Terminal_communication::Terminal_communication()
    :m_update_msg_map_thread(nullptr)
{
    if(m_update_msg_map_thread== nullptr)
    {
        m_publish_exit_condition.reset(false, false, false);
        m_update_msg_map_thread=new std::thread(thread_update_map_function,this);
    }
}
Terminal_communication::~Terminal_communication(){}

//重载函数
Error_manager Terminal_communication::encapsulate_msg(Communication_message* message)
{
    Error_manager code;
    switch (message->get_message_type())
    {
        case Communication_message::eStore_command_request_msg:
        {
            message::Store_command_request_msg request;
            request.ParseFromString(message->get_message_buf());
            //清空记录
            m_store_response_msg=message::Store_command_response_msg();
            //发送请求
            code= Communication_socket_base::encapsulate_msg(message);
            break;
        }

        case Communication_message::ePickup_command_request_msg:
        {
            message::Pickup_command_request_msg request;
            request.ParseFromString(message->get_message_buf());
            //清空记录
            m_pickup_response_msg=message::Pickup_command_response_msg();
            //发送请求
            code= Communication_socket_base::encapsulate_msg(message);
            break;
        }

        default:
            code= Error_manager(ERROR,NEGLIGIBLE_ERROR,"terminal message table is not exist");
    }
    return code;
}
Error_manager Terminal_communication::execute_msg(Communication_message* p_msg)
{

    if(p_msg->get_message_type()==Communication_message::eStore_command_response_msg)
    {
        message::Store_command_response_msg response;
        if(false==response.ParseFromString(p_msg->get_message_buf()))
        {
            return Error_manager(ERROR,CRITICAL_ERROR,"停车指令反馈信息解析错误");
        }
        message::Base_info base_info=response.base_info();
        if(base_info.sender()==message::eMain && base_info.receiver()==message::eTerminor)
        {
            message::Error_manager error_code=response.code();
            if(error_code.error_code()==0)
            {
                m_store_response_msg=response;
                return SUCCESS;
            }
        }
    }

    if(p_msg->get_message_type()==Communication_message::ePickup_command_response_msg)
    {
        message::Pickup_command_response_msg response;
        if(false==response.ParseFromString(p_msg->get_message_buf()))
        {
            return Error_manager(ERROR,CRITICAL_ERROR,"停车指令反馈信息解析错误");
        }
        message::Base_info base_info=response.base_info();
        if(base_info.sender()==message::eMain && base_info.receiver()==message::eTerminor)
        {
            message::Error_manager error_code=response.code();
            if(error_code.error_code()==0)
            {
                m_pickup_response_msg=response;
                return SUCCESS;
            }
        }
    }

    if(p_msg->get_message_type()==Communication_message::eStoring_process_statu_msg)
    {
        message::Storing_process_statu_msg msg;
        if(msg.ParseFromString(p_msg->get_message_buf())==false)
        {
            std::cout<<" ERROR : Storing_process_statu_msg 失败"<<std::endl;
        }
        message::Base_info base_info=msg.base_info();
        if(base_info.sender()==message::eMain)
        {
            if(m_storing_statu_map.find(msg.license())==false)
            {
                m_storing_license_queue.push(msg.license());
            }
            m_storing_statu_map[msg.license()]=msg;
            std::chrono::system_clock::time_point time_point=std::chrono::system_clock::now();
            m_storing_statu_time_point_map[msg.license()]=time_point;


        }
    }

    if(p_msg->get_message_type()==Communication_message::ePicking_process_statu_msg)
    {
        message::Picking_process_statu_msg msg;
        if(msg.ParseFromString(p_msg->get_message_buf())==false)
        {
            std::cout<<" ERROR : Picking_process_statu_msg 失败"<<std::endl;
        }
        message::Base_info base_info=msg.base_info();
        if(base_info.sender()==message::eMain)
        {
            if(m_picking_statu_map.find(msg.license())==false)
            {
                m_picking_license_queue.push(msg.license());
            }
            m_picking_statu_map[msg.license()]=msg;
            std::chrono::system_clock::time_point time_point=std::chrono::system_clock::now();
            m_picking_statu_time_point_map[msg.license()]=time_point;
        }
    }

    return SUCCESS;

    return Error_manager(FAILED,MINOR_ERROR,"terminal communication 未知消息");
}
/*
 * 检测消息是否可被处理
 */
Error_manager Terminal_communication::check_msg(Communication_message* p_msg)
{
    return SUCCESS;
}
/*
 * 心跳发送函数,重载
 */
Error_manager Terminal_communication::encapsulate_send_data()
{
    return SUCCESS;
}
//检查消息是否可以被解析, 需要重载
Error_manager Terminal_communication::check_executer(Communication_message* p_msg)
{
    return SUCCESS;
}


/*
    * 发送停车指令请求
    */
Error_manager Terminal_communication::store_request(message::Store_command_request_msg& request,message::Store_command_response_msg& response)
{

    if(request.has_locate_information()==false ||
        request.has_car_info()==false)
    {
        return Error_manager(ERROR,CRITICAL_ERROR,"停车指令请求消息缺少必要信息");
    }

    Error_manager code;
    Communication_message message;
    message.reset(request.base_info(),request.SerializeAsString());
    int timeout=1000*3;

    code=encapsulate_msg(&message);
    if(code!=SUCCESS)
        return code;

    //循环查询请求是否被处理
    auto start_time=std::chrono::system_clock::now();
    double time=0;
    do {
        //查询到记录
        ///查询是否存在,并且删除该记录,
        //判断是否接收到回应,若回应信息被赋值则证明有回应
        if (m_store_response_msg.has_base_info() && m_store_response_msg.has_code()) {
            message::Base_info response_base = m_store_response_msg.base_info();
            //检查类型是否匹配
            if (response_base.msg_type() != message::eStore_command_response_msg) {
                return Error_manager(ERROR, CRITICAL_ERROR,
                                     "停车指令反馈消息 response msg type error");
            }
            //检查基本信息是否匹配
            if (response_base.sender() != message::eMain ||
                response_base.receiver() != message::eTerminor ) {
                return Error_manager(PARKSPACE_RELEASE_RESPONSE_INFO_ERROR, MAJOR_ERROR,
                                     "parkspace store response msg info error");
            }
            response.CopyFrom(m_store_response_msg);
            m_store_response_msg.clear_base_info();
            return SUCCESS;

        }

        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();
        usleep(1000);
    }while(time<double(timeout));
    //超时,删除记录,返回错误
    return Error_manager(RESPONSE_TIMEOUT,MINOR_ERROR,"parkspace store request timeout");

}

/*
 * 发送取车指令请求
 */
Error_manager Terminal_communication::pickup_request(message::Pickup_command_request_msg& request,message::Pickup_command_response_msg& response)
{
    if(request.has_car_info()==false)
    {
        return Error_manager(ERROR,CRITICAL_ERROR,"取车指令请求消息缺少车辆信息");
    }


    Error_manager code;
    Communication_message message;
    message.reset(request.base_info(),request.SerializeAsString());
    int timeout=1000*3;

    code=encapsulate_msg(&message);
    if(code!=SUCCESS)
        return code;

   // m_statu=eTerminal_picking;
    //循环查询请求是否被处理
    auto start_time=std::chrono::system_clock::now();
    double time=0;
    do {
        //查询到记录
        ///查询是否存在,并且删除该记录,
        //判断是否接收到回应,若回应信息被赋值则证明有回应
        if (m_pickup_response_msg.has_base_info() && m_pickup_response_msg.has_code())
        {
            message::Base_info response_base = m_pickup_response_msg.base_info();
            //检查类型是否匹配
            if (response_base.msg_type() != message::ePickup_command_response_msg) {
                return Error_manager(ERROR, CRITICAL_ERROR,
                                     "停车指令反馈消息 response msg type error");
            }
            //检查基本信息是否匹配
            if (response_base.sender() != message::eMain ||
                response_base.receiver() != message::eTerminor ) {
                return Error_manager(PARKSPACE_RELEASE_RESPONSE_INFO_ERROR, MAJOR_ERROR,
                                     "parkspace release response msg info error");
            }
            response.CopyFrom(m_pickup_response_msg);
            m_pickup_response_msg.clear_base_info();
            //m_pickup_response_msg=message::Pickup_command_response_msg();
            return SUCCESS;

        }

        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();
        usleep(1000);
    }while(time<double(timeout));
    //超时,删除记录,返回错误
    return Error_manager(RESPONSE_TIMEOUT,MINOR_ERROR,"parkspace release request timeout");

}


/*
     * 获取停车流程状态
     */
Error_manager Terminal_communication::get_storing_statu(std::string car_license,message::Storing_process_statu_msg& msg)
{
    if(m_storing_statu_map.find(car_license,msg)) {
        return SUCCESS;
    }
    return FAILED;
}

/*
 * 获取取车流程状态
 */
Error_manager Terminal_communication::get_picking_statu(std::string car_license,message::Picking_process_statu_msg& msg)
{
    if(m_picking_statu_map.find(car_license,msg))
        return SUCCESS;
    return FAILED;
}

void Terminal_communication::update_map()
{
    while(false==m_publish_exit_condition.wait_for_ex(std::chrono::milliseconds(200)))
    {
        std::string license;
        if (m_storing_license_queue.try_pop(license))
        {
            std::chrono::system_clock::time_point t1 = std::chrono::system_clock::now();
            std::chrono::system_clock::time_point start = m_storing_statu_time_point_map[license];
            auto duration = std::chrono::duration_cast<std::chrono::microseconds>(t1 - start);
            double time = 1000.0 * double(duration.count()) * std::chrono::microseconds::period::num /
                          std::chrono::microseconds::period::den;

            if (time > 3000) {
                m_storing_statu_time_point_map.erase(license);
                m_storing_statu_map.erase(license);

            }
            else
            {
                m_storing_license_queue.push(license);
            }
        }

        if (m_picking_license_queue.try_pop(license))
        {
            std::chrono::system_clock::time_point t1 = std::chrono::system_clock::now();
            std::chrono::system_clock::time_point start = m_picking_statu_time_point_map[license];
            auto duration = std::chrono::duration_cast<std::chrono::microseconds>(t1 - start);
            double time = 1000.0 * double(duration.count()) * std::chrono::microseconds::period::num /
                          std::chrono::microseconds::period::den;

            if (time > 3000) {
                m_picking_statu_time_point_map.erase(license);
                m_picking_statu_map.erase(license);

            }
            else
            {
                m_picking_license_queue.push(license);
            }
        }

    }
}

void Terminal_communication::thread_update_map_function(Terminal_communication *terminal) {
    terminal->update_map();
}