Terminal_process_CT/parkspace/Parkspace_communicator.cpp

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

#include "Parkspace_communicator.h"


Parkspace_communicator::~Parkspace_communicator(){}
/*
 * 请求分配车位
 */
Error_manager Parkspace_communicator::alloc_request(message::Parkspace_allocation_request_msg& request,message::Parkspace_allocation_response_msg& result)
{
    /*
     * 检查request合法性,以及模块状态
     */
    if(request.base_info().sender()!=message::eMain||request.base_info().receiver()!=message::eParkspace)
        return Error_manager(PARKSPACE_ALLOC_REQUEST_INVALID,MINOR_ERROR,"车位分配请求信息错误!!!");

    if(m_alloc_table.find(request)==true)
        return Error_manager(PARKSPACE_ALLOC_REQUEST_REPEATED,MINOR_ERROR,"车位分配请求已经存在,重复!!!");
    //设置超时,若没有设置,默认1000
    int timeout=request.base_info().has_timeout_ms()?request.base_info().timeout_ms():1000;
    //向测量节点发送测量请求,并记录请求

    Error_manager code;
    Communication_message message;

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

    code=encapsulate_msg(&message);
    if(code!=SUCCESS)
        return code;
    //循环查询请求是否被处理
    auto start_time=std::chrono::system_clock::now();
    double time=0;
    do{
        //查询到记录
        message::Parkspace_allocation_response_msg response;
        ///查询是否存在,并且删除该记录,
        if(m_alloc_table.find(request,response))
        {
            //判断是否接收到回应,若回应信息被赋值则证明有回应
            if (response.has_base_info() && response.has_command_info())
            {
                message::Base_info response_base = response.base_info();
                //检查类型是否匹配
                if (response_base.msg_type() != message::eParkspace_allocation_response_msg) {
                    return Error_manager(PARKSPACE_ALLOC_RESPONSE_TYPE_ERROR, MINOR_ERROR,
                                         "parkspace alloc response msg type error");
                }
                //检查基本信息是否匹配
                if (response_base.sender() != message::eParkspace ||
                    response_base.receiver() != message::eMain ||
                    response.command_info() != request.command_info()) {
                    return Error_manager(PARKSPACE_ALLOC_RESPONSE_INFO_ERROR, MINOR_ERROR,
                                         "parkspace alloc response msg info error");
                }
                result = response;
                m_alloc_table.erase(request);

                return SUCCESS;

            }
        }
        else
        {
            //未查询到记录,任务已经被提前取消,记录被删除
            return Error_manager(PARKSPACE_ALLOC_REQUEST_CANCELED,MINOR_ERROR,"分配车位请求提前取消!!!");
        }

        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));
    //超时,删除记录,返回错误
    m_alloc_table.erase(request);
    return Error_manager(RESPONSE_TIMEOUT,MINOR_ERROR,"parkspace alloc request timeout");

}

/*
 * 查询车辆所在位置请求
 */
Error_manager Parkspace_communicator::search_request(message::Parkspace_search_request_msg& request,
        message::Parkspace_search_response_msg& result)
{
    /*
     * 检查request合法性,以及模块状态
     */
    if(request.base_info().sender()!=message::eMain||request.base_info().receiver()!=message::eParkspace)
        return Error_manager(PARKSPACE_SEARCH_REQUEST_INVALID,MINOR_ERROR,"parkspace search request invalid");

    if(m_search_table.find(request)==true)
        return Error_manager(PARKSPACE_SEARCH_REQUEST_REPEATED,MAJOR_ERROR," parkspace search request repeated");
    //设置超时,若没有设置,默认1000
    int timeout=request.base_info().has_timeout_ms()?request.base_info().timeout_ms():1000;
    //向测量节点发送测量请求,并记录请求

    Error_manager code;
    Communication_message message;

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

    code=encapsulate_msg(&message);
    if(code!=SUCCESS)
        return code;
    //循环查询请求是否被处理
    auto start_time=std::chrono::system_clock::now();
    double time=0;
    do{
        //查询到记录
        message::Parkspace_search_response_msg response;
        ///查询是否存在,并且删除该记录,
        if(m_search_table.find(request,response))
        {
            //判断是否接收到回应,若回应信息被赋值则证明有回应
            if (response.has_base_info() && response.has_command_info())
            {
                message::Base_info response_base = response.base_info();
                //检查类型是否匹配
                if (response_base.msg_type() != message::eParkspace_search_response_msg) {
                    return Error_manager(PARKSPACE_SEARCH_RESPONSE_TYPE_ERROR, MAJOR_ERROR,
                                         "parkspace search response msg type error");
                }
                //检查基本信息是否匹配
                if (response_base.sender() != message::eParkspace ||
                    response_base.receiver() != message::eMain ||
                    response.command_info() != request.command_info()) {
                    return Error_manager(PARKSPACE_SEARCH_RESPONSE_INFO_ERROR, MAJOR_ERROR,
                                         "parkspace search response msg info error");
                }
                result = response;
                m_search_table.erase(request);
                if(response.has_error_manager())
                {
                    if(response.error_manager().error_code()==0)
                        return SUCCESS;
                }
                return Error_manager(FAILED,MINOR_ERROR,"车位查询返回错误码");

            }
        }
        else
        {
            //未查询到记录,任务已经被提前取消,记录被删除
            return Error_manager(PARKSPACE_SEARCH_REQUEST_CANCELED,MINOR_ERROR,"parkspace search 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();
        usleep(1000);
    }while(time<double(timeout));
    //超时,删除记录,返回错误
    m_search_table.erase(request);
    return Error_manager(RESPONSE_TIMEOUT,MINOR_ERROR,"parkspace search request timeout");
}

/*
 * 释放车位请求(停车失败或者取车完成时调用)
 */
Error_manager Parkspace_communicator::release_request(message::Parkspace_release_request_msg& request,
        message::Parkspace_release_response_msg& result)
{
    /*
    * 检查request合法性,以及模块状态
    */
    if(request.base_info().sender()!=message::eMain||request.base_info().receiver()!=message::eParkspace)
        return Error_manager(PARKSPACE_RELEASE_REQUEST_INVALID,MINOR_ERROR,"parkspace release request invalid");

    if(m_release_table.find(request)==true)
        return Error_manager(PARKSPACE_RELEASE_REQUEST_REPEATED,MAJOR_ERROR," parkspace release request repeated");
    //设置超时,若没有设置,默认1000
    int timeout=request.base_info().has_timeout_ms()?request.base_info().timeout_ms():1000;
    //向测量节点发送测量请求,并记录请求

    Error_manager code;
    Communication_message message;

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

    code=encapsulate_msg(&message);
    if(code!=SUCCESS)
        return code;
    //循环查询请求是否被处理
    auto start_time=std::chrono::system_clock::now();
    double time=0;
    do{
        //查询到记录
        message::Parkspace_release_response_msg response;
        ///查询是否存在,并且删除该记录,
        if(m_release_table.find(request,response))
        {
            //判断是否接收到回应,若回应信息被赋值则证明有回应
            if (response.has_base_info() && response.has_command_info())
            {
                message::Base_info response_base = response.base_info();
                //检查类型是否匹配
                if (response_base.msg_type() != message::eParkspace_release_response_msg) {
                    return Error_manager(PARKSPACE_RELEASE_RESPONSE_TYPE_ERROR, MAJOR_ERROR,
                                         "parkspace release response msg type error");
                }
                //检查基本信息是否匹配
                if (response_base.sender() != message::eParkspace ||
                    response_base.receiver() != message::eMain ||
                    response.command_info() != request.command_info()) {
                    return Error_manager(PARKSPACE_RELEASE_RESPONSE_INFO_ERROR, MAJOR_ERROR,
                                         "parkspace release response msg info error");
                }
                result = response;
                m_release_table.erase(request);

                return SUCCESS;

            }
        }
        else
        {
            //未查询到记录,任务已经被提前取消,记录被删除
            return Error_manager(PARKSPACE_RELEASE_REQUEST_CANCELED,MINOR_ERROR,"parkspace release 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();
        usleep(1000);
    }while(time<double(timeout));
    //超时,删除记录,返回错误
    m_release_table.erase(request);
    return Error_manager(RESPONSE_TIMEOUT,MINOR_ERROR,"parkspace release request timeout");
}

/*
     * 确认占用车位消息
     */
Error_manager Parkspace_communicator::confirm_request(message::Parkspace_confirm_alloc_request_msg& request,
        message::Parkspace_confirm_alloc_response_msg& result)
{
    /*
    * 检查request合法性,以及模块状态
    */
    if(request.base_info().sender()!=message::eMain||request.base_info().receiver()!=message::eParkspace)
        return Error_manager(FAILED,MINOR_ERROR,"parkspace confirm request invalid");


    if(m_confirm_table.find(request)==true)
    {
        return Error_manager(FAILED,MAJOR_ERROR," parkspace confirm request repeated");
    }
    //设置超时,若没有设置,默认1000
    int timeout=request.base_info().has_timeout_ms()?request.base_info().timeout_ms():1000;
    //向测量节点发送测量请求,并记录请求

    Error_manager code;
    Communication_message message;
    message.reset(request.base_info(),request.SerializeAsString());

    code=encapsulate_msg(&message);
    if(code!=SUCCESS)
        return code;
    //循环查询请求是否被处理
    auto start_time=std::chrono::system_clock::now();
    double time=0;
    do{
        //查询到记录
        message::Parkspace_confirm_alloc_response_msg response;
        ///查询是否存在,并且删除该记录,
        if(m_confirm_table.find(request,response))
        {
            //判断是否接收到回应,若回应信息被赋值则证明有回应
            if (response.has_base_info() && response.has_command_info())
            {
                message::Base_info response_base = response.base_info();
                //检查类型是否匹配
                if (response_base.msg_type() != message::eParkspace_confirm_alloc_response_msg) {
                    return Error_manager(PARKSPACE_RELEASE_RESPONSE_TYPE_ERROR, MAJOR_ERROR,
                                         "parkspace confirm response msg type error");
                }
                //检查基本信息是否匹配
                if (response_base.sender() != message::eParkspace ||
                    response_base.receiver() != message::eMain ||
                    response.command_info() != request.command_info()) {
                    return Error_manager(PARKSPACE_RELEASE_RESPONSE_INFO_ERROR, MAJOR_ERROR,
                                         "parkspace confirm response msg info error");
                }
                result = response;
                m_confirm_table.erase(request);

                return SUCCESS;

            }
        }
        else
        {
            //未查询到记录,任务已经被提前取消,记录被删除
            return Error_manager(FAILED,MINOR_ERROR,"parkspace confirm 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();
        usleep(1000);
    }while(time<double(timeout));
    //超时,删除记录,返回错误
    m_confirm_table.erase(request);
    return Error_manager(RESPONSE_TIMEOUT,MINOR_ERROR,"parkspace confirm request timeout");
}

Error_manager Parkspace_communicator::check_statu()
{
    //return SUCCESS;
    std::chrono::system_clock::time_point time_now=std::chrono::system_clock::now();
    auto durantion=time_now-m_parkspace_statu_time;
    if(m_parkspace_status_msg.has_base_info()== false
       || durantion>std::chrono::seconds(2))
    {
        return Error_manager(ERROR,MINOR_ERROR,"车位管理节点通讯断开");
    }

    return SUCCESS;
}



Parkspace_communicator::Parkspace_communicator(){}

Error_manager Parkspace_communicator::encapsulate_msg(Communication_message* message)
{
    Error_manager code;
    switch (message->get_message_type())
    {
        case Communication_message::eParkspace_allocation_request_msg:
        {
            message::Parkspace_allocation_request_msg request;
            request.ParseFromString(message->get_message_buf());
            //记录数据
            m_alloc_table[request]=message::Parkspace_allocation_response_msg();
            //发送请求
            code= Communication_socket_base::encapsulate_msg(message);
            if(code!=SUCCESS)
            {
                m_alloc_table.erase(request);
            }
            break;
        }
        case Communication_message::eParkspace_search_request_msg:
        {
            message::Parkspace_search_request_msg request;
            request.ParseFromString(message->get_message_buf());
            //记录数据
            m_search_table[request]=message::Parkspace_search_response_msg();
            //发送请求
            code= Communication_socket_base::encapsulate_msg(message);
            if(code!=SUCCESS)
            {
                m_search_table.erase(request);
            }
            break;
        }
        case Communication_message::eParkspace_release_request_msg:
        {
            message::Parkspace_release_request_msg request;
            request.ParseFromString(message->get_message_buf());
            //记录数据
            m_release_table[request]=message::Parkspace_release_response_msg();
            //发送请求
            code= Communication_socket_base::encapsulate_msg(message);
            if(code!=SUCCESS)
            {
                m_release_table.erase(request);
            }
            break;
        }
        case Communication_message::eParkspace_confirm_alloc_request_msg:
        {
            message::Parkspace_confirm_alloc_request_msg request;
            request.ParseFromString(message->get_message_buf());
            //记录数据
            LOG(INFO)<<" Add  request :  "<<request.command_info().DebugString();
            m_confirm_table[request]=message::Parkspace_confirm_alloc_response_msg();
            //发送请求
            code= Communication_socket_base::encapsulate_msg(message);
            if(code!=SUCCESS)
            {
                m_confirm_table.erase(request);
            }
            break;
        }
        default:
            code= Error_manager(PARKSPACE_REQUEST_MSG_TYPE_ERROR,NEGLIGIBLE_ERROR,"不存在该类型消息的发送处理!!!");
    }
    return code;

}
Error_manager Parkspace_communicator::execute_msg(Communication_message* p_msg)
{
    if(p_msg== nullptr)
        return Error_manager(POINTER_IS_NULL,CRITICAL_ERROR,"parkspace response msg pointer is null");


    //车位response消息
    switch (p_msg->get_message_type())
    {
        ///测量结果反馈消息
        case Communication_message::eParkspace_allocation_response_msg:
        {
            message::Parkspace_allocation_response_msg response;
            response.ParseFromString(p_msg->get_message_buf());
            message::Parkspace_allocation_request_msg request=create_request_by_response(response);
            ///查询请求表是否存在,并且更新
            if(m_alloc_table.find_update(request,response)==false)
            {
                return Error_manager(PARKSPACE_ALLOCMSG_RESPONSE_HAS_NO_REQUEST,NEGLIGIBLE_ERROR,"parkspace alloc response without request");
            }

            break;
        }
        case Communication_message::eParkspace_search_response_msg:
        {
            message::Parkspace_search_response_msg response;
            if(false==response.ParseFromString(p_msg->get_message_buf()))
            {
                return Error_manager(ERROR,CRITICAL_ERROR,"parkspace search response 解析失败");
            }
            message::Parkspace_search_request_msg request=create_request_by_response(response);
            ///查询请求表是否存在,并且更新
            if(m_search_table.find_update(request,response)==false)
            {
                return Error_manager(PARKSPACE_SEARCHMSG_RESPONSE_HAS_NO_REQUEST,NEGLIGIBLE_ERROR,"parkspace search response without request");
            }
            break;
        }
        case Communication_message::eParkspace_release_response_msg:
        {
            message::Parkspace_release_response_msg response;
            response.ParseFromString(p_msg->get_message_buf());
            message::Parkspace_release_request_msg request=create_request_by_response(response);
            ///查询请求表是否存在,并且更新
            if(m_release_table.find_update(request,response)==false)
            {
                return Error_manager(PARKSPACE_RELEASEMSG_RESPONSE_HAS_NO_REQUEST,NEGLIGIBLE_ERROR,"parkspace release response without request");
            }
            break;
        }

        case Communication_message::eParkspace_confirm_alloc_response_msg:
        {
            message::Parkspace_confirm_alloc_response_msg response;
            response.ParseFromString(p_msg->get_message_buf());
            message::Parkspace_confirm_alloc_request_msg request=create_request_by_response(response);
            ///查询请求表是否存在,并且更新
            if(m_confirm_table.find_update(request,response)==true)
            {
                return SUCCESS;
            }
            break;
        }
            ///测量系统状态
        case Communication_message::eParkspace_allocation_status_msg:
        {
            if(m_parkspace_status_msg.ParseFromString(p_msg->get_message_buf())==false)
            {
                return Error_manager(ERROR,CRITICAL_ERROR,"车位管理模块状态消息解析失败");
            }
            m_parkspace_statu_time=std::chrono::system_clock::now();
            break;
        }
    }
    return SUCCESS;

}
/*
 * 检测消息类型是否可被处理
 */
Error_manager Parkspace_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::eParkspace_allocation_response_msg
        ||p_msg->get_message_type() == Communication_message::Message_type::eParkspace_search_response_msg
        ||p_msg->get_message_type() == Communication_message::Message_type::eParkspace_release_response_msg
        ||p_msg->get_message_type() == Communication_message::Message_type::eParkspace_allocation_status_msg
        ||p_msg->get_message_type()==Communication_message::eParkspace_confirm_alloc_response_msg)
         && p_msg->get_receiver() == Communication_message::Communicator::eMain
         && p_msg->get_sender() == Communication_message::Communicator::eTable)
    {
        return Error_code::SUCCESS;
    }
    else
    {
        //认为接受人
        return Error_code::INVALID_MESSAGE;
    }
    return SUCCESS;
}
/*
 * 心跳发送函数,重载
 */
Error_manager Parkspace_communicator::encapsulate_send_data()
{
    return SUCCESS;
}
//检查消息是否可以被解析, 需要重载
Error_manager Parkspace_communicator::check_executer(Communication_message* p_msg)
{
    return SUCCESS;
}

/*
 * 根据接收到的response,生成对应的 request
 */
message::Parkspace_allocation_request_msg
Parkspace_communicator::create_request_by_response(message::Parkspace_allocation_response_msg& response)
{
    message::Parkspace_allocation_request_msg request;
    message::Base_info baseInfo;
    baseInfo.set_msg_type(message::eParkspace_allocation_request_msg);
    baseInfo.set_sender(response.base_info().receiver());
    baseInfo.set_receiver(response.base_info().sender());
    request.mutable_base_info()->CopyFrom(baseInfo);
    request.mutable_command_info()->CopyFrom(response.command_info());
    return request;
}

message::Parkspace_search_request_msg
Parkspace_communicator::create_request_by_response(message::Parkspace_search_response_msg& response)
{
    message::Parkspace_search_request_msg request;
    message::Base_info baseInfo;
    baseInfo.set_msg_type(message::eParkspace_search_request_msg);
    baseInfo.set_sender(response.base_info().receiver());
    baseInfo.set_receiver(response.base_info().sender());
    request.mutable_base_info()->CopyFrom(baseInfo);
    request.mutable_command_info()->CopyFrom(response.command_info());


    return request;
}

message::Parkspace_release_request_msg
Parkspace_communicator::create_request_by_response(message::Parkspace_release_response_msg& response)
{
    message::Parkspace_release_request_msg request;
    message::Base_info baseInfo;
    baseInfo.set_msg_type(message::eParkspace_release_request_msg);
    baseInfo.set_sender(response.base_info().sender());
    baseInfo.set_receiver(response.base_info().receiver());
    request.mutable_base_info()->CopyFrom(baseInfo);
    request.mutable_command_info()->CopyFrom(response.command_info());
    return request;
}

message::Parkspace_confirm_alloc_request_msg
Parkspace_communicator::create_request_by_response(message::Parkspace_confirm_alloc_response_msg& response)
{
    message::Parkspace_confirm_alloc_request_msg request;
    message::Base_info baseInfo;
    baseInfo.set_msg_type(message::eParkspace_confirm_alloc_request_msg);
    baseInfo.set_sender(response.base_info().sender());
    baseInfo.set_receiver(response.base_info().receiver());
    request.mutable_base_info()->CopyFrom(baseInfo);
    request.mutable_command_info()->CopyFrom(response.command_info());
    return request;
}