Terminal_process_CT/system/command_manager.cpp

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

#include <sstream>
#include <iomanip>
#include <Parkspace_communicator.h>
#include "command_manager.h"
#include "StoreProcessTask.h"
#include "PickupProcessTask.h"
#include "system_communicator.h"

Command_manager::Command_manager()
    :m_thread_queue_process(nullptr),
    m_publish_statu_thread(nullptr)
{

}

Command_manager::~Command_manager()
{
    //等待线程池完成
    if(m_thread_queue_process!=nullptr) {
        m_thread_queue_process->WaitForFinish();
        m_thread_queue_process->Stop();
    }

    //退出线程
    m_publish_exit_condition.set_pass_ever(true);
    if(m_publish_statu_thread!= nullptr)
    {
        if(m_publish_statu_thread->joinable())
        {
            m_publish_statu_thread->join();
        }
        delete m_publish_statu_thread;
        m_publish_statu_thread=nullptr;
    }

}

Error_manager Command_manager::init(setting::System_setting system_setting)
{
    /*
     * 检查参数
     */
    if(system_setting.has_bind_ip()==false || system_setting.has_entrance_num()== false
        ||system_setting.has_export_num()==false )
    {
        return Error_manager(ERROR,MAJOR_ERROR,"系统配置错误");
    }
    if(system_setting.entrance_num()<0 || system_setting.export_num()<0)
    {
        return Error_manager(ERROR,MAJOR_ERROR,"系统配置出入口数量错误");
    }

    //创建线程池
    if(m_thread_queue_process== nullptr)
    {
        m_thread_queue_process=tq::TQFactory::CreateDefaultQueue();
        m_thread_queue_process->Start(12);
    }


    /*
     * 此处添加等待各个通讯模块正常代码
     */
    std::chrono::system_clock::time_point t_start=std::chrono::system_clock::now();
    std::chrono::system_clock::time_point t_end=std::chrono::system_clock::now();

    Error_manager parkspace_code=ERROR;
    LOG(INFO)<<"初始化车位管理模块...";
    do
    {
        if (parkspace_code != SUCCESS) {
            parkspace_code = Parkspace_communicator::get_instance_pointer()->check_statu();
            LOG_IF(INFO, parkspace_code == SUCCESS) << "车位管理模块初始化完成!!!";
        }
        if(parkspace_code==SUCCESS)
            break;
        t_end=std::chrono::system_clock::now();
        usleep(1000*100);
    }while(t_end-t_start<std::chrono::seconds(300));

    LOG_IF(ERROR,parkspace_code!=SUCCESS)<<"车位管理节点连接超时";
    if(parkspace_code!=SUCCESS)
    {
        return Error_manager(ERROR,MAJOR_ERROR,"车位管理模块初始化超时");
    }

    //检查节点并创建停取车流程
    for(int i=0;i<system_setting.entrance_num();++i)
    {
        /*
         * 检查入口 i 的各个节点状态
         */
        LOG(INFO)<<"初始化停车入口:"<<i<<"  测量及调度模块...";
        Error_manager locate_code=ERROR,dispatch_code=ERROR;
        do {
            if (locate_code != SUCCESS) {
                locate_code = Locate_communicator::get_instance_pointer()->check_statu(i);
                LOG_IF(INFO, locate_code == SUCCESS) << "停车入口:"<<i<<" 测量模块初始化完成!!!";
            }
            if (dispatch_code != SUCCESS) {
                dispatch_code = Dispatch_communicator::get_instance_pointer()->check_entrance_statu(i);
                LOG_IF(INFO, dispatch_code == SUCCESS) << "停车入口:"<<i<<" 调度模块初始化完成!!!";
            }

            t_end=std::chrono::system_clock::now();
            if (locate_code == SUCCESS && dispatch_code == SUCCESS)
                break;
            usleep(1000*100);
        }while(t_end-t_start<std::chrono::seconds(300));

        LOG_IF(ERROR,locate_code!=SUCCESS||dispatch_code!=SUCCESS)<<"停车口:"<<i<<" 测量/调度节点连接超时";
        if(locate_code!=SUCCESS||dispatch_code!=SUCCESS)
        {
            return Error_manager(ERROR,MAJOR_ERROR,"停车入口通讯节点初始化超时");
        }

    }
    for(int i=0;i<system_setting.export_num();++i)
    {
        LOG(INFO)<<"初始化取车出口:"<<i<<"  调度模块...";
        Error_manager dispatch_code=ERROR;
        do {
            if (dispatch_code != SUCCESS) {
                dispatch_code = Dispatch_communicator::get_instance_pointer()->check_export_statu(i);
                LOG_IF(INFO, dispatch_code == SUCCESS) << "取车出口:" << i << " 调度模块初始化完成!!!";
            }

            t_end = std::chrono::system_clock::now();
            if (dispatch_code == SUCCESS)
                break;
            usleep(1000*100);
        }while(t_end-t_start<std::chrono::seconds(300));

        LOG_IF(ERROR,dispatch_code!=SUCCESS)<<"取车出口:"<<i<<" 调度节点连接超时";
        if(dispatch_code!=SUCCESS)
        {
            return Error_manager(ERROR,MAJOR_ERROR,"取车出口通讯节点初始化超时");
        }
    }

    ///创建状态发布线程
    message::Base_info base_info;
    base_info.set_sender(message::eMain);
    base_info.set_receiver(message::eEmpty);
    base_info.set_msg_type(message::eProcess_statu_map_msg);
    m_process_statu_msg.mutable_base_info()->CopyFrom(base_info);
    if(m_publish_statu_thread== nullptr)
    {
        m_publish_exit_condition.reset(false, false, false);
        m_publish_statu_thread=new std::thread(publish_thread_func,this);
    }
    return SUCCESS;
}

/*
 * 执行停车请求
 */
Error_manager Command_manager::execute_store_command(message::Store_command_request_msg& request,message::Store_command_response_msg& response)
{

    if(m_thread_queue_process==nullptr)
    {
        return Error_manager(ERROR,CRITICAL_ERROR,"线程池未初始化,bug");
    }
    if(request.base_info().msg_type()==message::eStore_command_request_msg
       &&request.base_info().receiver()==message::eMain
       &&request.base_info().sender()==message::eTerminor)
    {
        if(request.has_locate_information() &&request.has_base_info())
        {
            message::Locate_information locate_info=request.locate_information();
            if(locate_info.has_locate_correct())
            {
                if(locate_info.locate_correct()==true)
                {
                    if(locate_info.has_locate_width()&&locate_info.has_locate_height()
                       &&locate_info.has_locate_x()&&locate_info.has_locate_y()
                       &&locate_info.has_locate_angle()&&locate_info.has_locate_wheel_base())
                    {
                        /*
                         * 检查消息完毕,开始处理
                         */
                        message::Base_info base_info;
                        base_info.set_msg_type(message::eStore_command_response_msg);
                        base_info.set_sender(message::eMain);
                        base_info.set_receiver(message::eTerminor);
                        response.mutable_base_info()->CopyFrom(base_info);
                        response.set_terminal_id(request.terminal_id());

                        message::Error_manager error_msg;
                        error_msg.set_error_code(0);

                        /*
                        * 检查各个节点是否正常
                        */
                        Error_manager parkspace_code= Parkspace_communicator::get_instance_pointer()->check_statu();
                        if(parkspace_code!=SUCCESS)
                        {
                            error_msg.set_error_code(parkspace_code.get_error_code());
                            error_msg.set_error_description(parkspace_code.get_error_description());
                            response.mutable_code()->CopyFrom(error_msg);
                            return parkspace_code;
                        }
                        Error_manager locate_code= Locate_communicator::get_instance_pointer()->check_statu(request.terminal_id());
                        if(locate_code!=SUCCESS)
                        {
                            error_msg.set_error_code(locate_code.get_error_code());
                            error_msg.set_error_description(locate_code.get_error_description());
                            response.mutable_code()->CopyFrom(error_msg);
                            return locate_code;
                        }
                        Error_manager dispatch_code= Dispatch_communicator::get_instance_pointer()->check_entrance_statu(request.terminal_id());
                        if(dispatch_code!=SUCCESS)
                        {
                            error_msg.set_error_code(dispatch_code.get_error_code());
                            error_msg.set_error_description(dispatch_code.get_error_description());
                            response.mutable_code()->CopyFrom(error_msg);
                            return dispatch_code;
                        }

                        //一切正常,接受指令
                        message::Command_info command_info;
                        //获取当前时间
                        auto t = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
                        std::stringstream ss;
                        ss << std::put_time(std::localtime(&t), "%Y-%m-%d:%H-%M-%S");
                        std::string str_time = ss.str();
                        command_info.set_time(str_time);

                        char place[255]={0};
                        sprintf(place ,"027-01-%02d:%s",request.terminal_id(),request.car_info().license().c_str());
                        command_info.set_place(place);
                        command_info.set_event(message::eStoring);

                        LOG(INFO)<<"收到停车指令:"<<command_info.place()+"-"+command_info.time()<<", 车牌:"<<request.car_info().license()<<
                        ",终端:"<<request.terminal_id();
                        tq::BaseTask* ptask=new StoreProcessTask(request.terminal_id());
                        StoreProcessTask* pStore_task=(StoreProcessTask*)ptask;
                        //初始化流程
                        pStore_task->init_task(command_info,locate_info,request.car_info());

                        //获取车位
                        Error_manager code=pStore_task->alloc_space();
                        if(code==SUCCESS)
                        {
                            m_thread_queue_process->AddTask(pStore_task);
                            response.mutable_code()->CopyFrom(error_msg);
                            return SUCCESS;

                        }
                        usleep(1000*5000);
                        delete pStore_task;
                        error_msg.set_error_code(code.get_error_code());
                        error_msg.set_error_description(code.to_string());
                        response.mutable_code()->CopyFrom(error_msg);
                        LOG(ERROR)<<"创建停车流程失败(车位分配失败),终端:"<<request.terminal_id()<<
                                  "车牌:"<<request.car_info().license()<<"  "<<code.to_string();
                        return code;

                    }
                }
            }

        }

        return Error_manager(FAILED,MINOR_ERROR,"创建停车流程失败......");
    }
    else
    {
        return Error_manager(INVALID_MESSAGE,MAJOR_ERROR,"停车请求基本信息错误");
    }
}

/*
 * 执行取车请求
 */
Error_manager Command_manager::execute_pickup_command(message::Pickup_command_request_msg& request,message::Pickup_command_response_msg& response)
{

    if(m_thread_queue_process==nullptr)
    {
        return Error_manager(ERROR,CRITICAL_ERROR,"线程池未初始化,bug");
    }
    if(request.base_info().msg_type()==message::ePickup_command_request_msg
       &&request.base_info().receiver()==message::eMain
       &&request.base_info().sender()==message::eTerminor
       &&request.has_car_info()) {

        message::Base_info baseInfo;
        baseInfo.set_msg_type(message::ePickup_command_response_msg);
        baseInfo.set_sender(message::eMain);
        baseInfo.set_receiver(message::eTerminor);
        response.mutable_base_info()->CopyFrom(baseInfo);
        response.set_terminal_id(request.terminal_id());

        message::Error_manager error_msg;
        error_msg.set_error_code(0);

        /*
        * 检查各个节点是否正常
        */
        Error_manager parkspace_code= Parkspace_communicator::get_instance_pointer()->check_statu();
        if(parkspace_code!=SUCCESS)
        {
            error_msg.set_error_code(parkspace_code.get_error_code());
            error_msg.set_error_description(parkspace_code.get_error_description());
            response.mutable_code()->CopyFrom(error_msg);
            return parkspace_code;
        }

        Error_manager dispatch_code= Dispatch_communicator::get_instance_pointer()->check_export_statu(request.terminal_id());
        if(dispatch_code!=SUCCESS)
        {
            error_msg.set_error_code(dispatch_code.get_error_code());
            error_msg.set_error_description(dispatch_code.get_error_description());
            response.mutable_code()->CopyFrom(error_msg);
            return dispatch_code;
        }

        //一切正常,接受指令
        Error_manager code;

        message::Command_info command_info;
        //获取当前时间
        auto t = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
        std::stringstream ss;
        ss << std::put_time(std::localtime(&t), "%Y-%m-%d:%H-%M-%S-%ms");
        std::string str_time = ss.str();
        command_info.set_time(str_time);

        char place[255]={0};
        sprintf(place ,"027-01-%02d",request.terminal_id());
        command_info.set_place(place);
        command_info.set_event(message::ePicking);

        tq::BaseTask* ptask=new PickupProcessTask(request.terminal_id());
        PickupProcessTask* pPick_task=(PickupProcessTask*)ptask;
        //初始化流程
        pPick_task->init_task(command_info, request.car_info());
        /////查询车位
        code=pPick_task->search_space();

        if(code==SUCCESS)
        {
            m_thread_queue_process->AddTask(pPick_task);
            response.mutable_code()->CopyFrom(error_msg);
            return SUCCESS;
        }
        error_msg.set_error_code(code.get_error_code());
        error_msg.set_error_description(code.to_string());
        response.mutable_code()->CopyFrom(error_msg);
        LOG(ERROR)<<"创建取车流程失败(车位查询失败),终端:"<<request.terminal_id()<<
                  "车牌:"<<request.car_info().license()<<"  "<<code.to_string();
        return code;
    }
    else
    {
        return Error_manager(INVALID_MESSAGE,MAJOR_ERROR,"停车请求信息错误");
    }
}

void Command_manager::erase_statu(message::Storing_process_statu statu)
{
    std::lock_guard<std::mutex> lock(m_process_statu_lock);
    m_process_statu_msg.mutable_storing_process_msg_map()->erase(statu.license());
}
void Command_manager::erase_statu(message::Picking_process_statu statu)
{
    std::lock_guard<std::mutex> lock(m_process_statu_lock);
    m_process_statu_msg.mutable_picking_process_msg_map()->erase(statu.license());
}

void Command_manager::updata_statu(message::Storing_process_statu statu)
{
    std::lock_guard<std::mutex> lock(m_process_statu_lock);
    (*m_process_statu_msg.mutable_storing_process_msg_map())[statu.license()]=statu;
}
void Command_manager::updata_statu(message::Picking_process_statu statu)
{
    std::lock_guard<std::mutex> lock(m_process_statu_lock);
    (*m_process_statu_msg.mutable_picking_process_msg_map())[statu.license()]=statu;
}

void Command_manager::publish_step_status() {
    //未收到退出信号
    while(false==m_publish_exit_condition.wait_for_ex(std::chrono::milliseconds(100)))
    {
        /*
         * 通过communicator 发布状态
         */

        if(System_communicator::get_instance_pointer())
        {
            std::lock_guard<std::mutex> lock(m_process_statu_lock);
            if(m_process_statu_msg.has_base_info()==true)
            {
                System_communicator::get_instance_pointer()->post_entrance_statu(m_process_statu_msg);
            }

        }

    }
}

void Command_manager::publish_thread_func(Command_manager* command_manager)
{
    if(command_manager)
        command_manager->publish_step_status();
}