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_system_paused(false)
    ,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, "系统配置出入口数量错误");
    }
    //初始化出入口状态为 Enable
    m_input_entrance_statu.resize(system_setting.entrance_num());
    for(int i=0;i<system_setting.entrance_num();++i)
        m_input_entrance_statu[i]=Enable;
    m_output_entrance_statu.resize(system_setting.export_num());
    for(int i=0;i<system_setting.export_num();++i)
        m_output_entrance_statu[i]=Enable;

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

    /*
     * 此处添加等待各个通讯模块正常代码
     */
    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,"取车出口通讯节点初始化超时");
        }
    }
    m_system_setting=system_setting;

    /*
     * 创建发布线程
     */
    if(m_publish_statu_thread== nullptr)
    {
        m_publish_exit_condition.reset(false, false, false);
        m_publish_statu_thread=new std::thread(publish_statu_function,this);
    }

    return SUCCESS;

}
/*
 * 执行停车请求
 */

Error_manager Command_manager::execute_store_command(message::Store_command_request_msg& request,message::Store_command_response_msg& response)
{
    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);

    if (m_thread_queue_process == nullptr) {
        error_msg.set_error_code(ERROR);
        error_msg.set_error_description("bug,线程池未初始化");
        response.mutable_code()->CopyFrom(error_msg);
        return Error_manager(ERROR, MINOR_ERROR, "线程池未初始化,bug");
    }

    if(m_system_paused==true)
    {
        error_msg.set_error_code(PAUSE);
        error_msg.set_error_description("急停");
        response.mutable_code()->CopyFrom(error_msg);
        return Error_manager(PAUSE, MINOR_ERROR, "系统已急停");
    }

    //判断入口是否开放
    if(request.terminal_id()<0 || request.terminal_id()>=m_system_setting.entrance_num())
    {
        error_msg.set_error_code(ERROR);
        error_msg.set_error_description("terminal_id 参数设置错误");
        response.mutable_code()->CopyFrom(error_msg);
        return Error_manager(ERROR, MINOR_ERROR, "terminal id 设置错误");
    }
    if(m_input_entrance_statu[request.terminal_id()]!=Enable) {
        error_msg.set_error_code(ERROR);
        error_msg.set_error_description("入口已关闭 Disable");
        response.mutable_code()->CopyFrom(error_msg);
        return Error_manager(ERROR, MINOR_ERROR, "入口已禁止使用 Disable");
    }

    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())
                    {
                        /*
                         * 检查消息完毕,开始处理
                         */

                        response.set_terminal_id(request.terminal_id());

                        /*
                        * 检查各个节点是否正常
                        */
                        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;
                        }

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

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

                        }
                        usleep(1000 * 1000);
                        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)
{

    message::Base_info base_info;
    base_info.set_msg_type(message::ePickup_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);

    if (m_thread_queue_process == nullptr) {
        error_msg.set_error_code(ERROR);
        error_msg.set_error_description("bug,线程池未初始化");
        response.mutable_code()->CopyFrom(error_msg);
        return Error_manager(ERROR, MINOR_ERROR, "线程池未初始化,bug");
    }

    if(m_system_paused==true)
    {
        error_msg.set_error_code(PAUSE);
        error_msg.set_error_description("急停");
        response.mutable_code()->CopyFrom(error_msg);
        return Error_manager(PAUSE, MINOR_ERROR, "急停");
    }


    //判断出口是否开放
    if(request.terminal_id()<0 || request.terminal_id()>=m_system_setting.export_num())
    {
        error_msg.set_error_code(ERROR);
        error_msg.set_error_description("出口 id 设置错误");
        response.mutable_code()->CopyFrom(error_msg);
        return Error_manager(ERROR, MINOR_ERROR, "出口 id 设置错误");
    }
    if(m_output_entrance_statu[request.terminal_id()]!=Enable)
    {
        error_msg.set_error_code(ERROR);
        error_msg.set_error_description("中控出口已停止使用");
        response.mutable_code()->CopyFrom(error_msg);
        return Error_manager(ERROR, MINOR_ERROR, "中控出口已停止使用");
    }

    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()) {

        response.set_terminal_id(request.terminal_id());

        /*
        * 检查各个节点是否正常
        */
        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;
        LOG(WARNING)<<"收到取车-----------------------------"<<request.car_info().license();
        Process_task* ptask=new PickupProcessTask(request.terminal_id(),request.car_info());
        PickupProcessTask* pPick_task=(PickupProcessTask*)ptask;
        //初始化流程
        code=pPick_task->init_task();
        /////查询车位
        code=pPick_task->search_space();

        if(code==SUCCESS)
        {
            pPick_task->next_step();
            m_thread_queue_process->AddTask(pPick_task);
            response.mutable_code()->CopyFrom(error_msg);
            return SUCCESS;
        }
        usleep(1000*1000);
        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();
        delete pPick_task;
        return code;
    }
    else
    {
        return Error_manager(INVALID_MESSAGE,MAJOR_ERROR,"停车请求信息错误");
    }
}

/*
    * 控制入口 开放或者关闭
    */
Error_manager Command_manager::enable_entrance(int terminal_id,Entrance_statu statu)
{
    if(terminal_id<0 || terminal_id>m_system_setting.entrance_num())
        return Error_manager(ERROR,MINOR_ERROR,"xxxx");
    m_input_entrance_statu[terminal_id]=statu;
    return SUCCESS;
}

/*
 * 控制出口 开放或者关闭
 */
Error_manager Command_manager::enable_export(int terminal_id,Entrance_statu statu)
{
    if(terminal_id<0 || terminal_id>m_system_setting.export_num())
        return Error_manager(ERROR,MINOR_ERROR,"xxxx");
    m_output_entrance_statu[terminal_id]=statu;
}

/*
     * pause,系统急停
     */
void Command_manager::pause_system()
{
    m_system_paused=true;
}

void Command_manager::publish_statu_function(Command_manager* commander)
{
    if(commander== nullptr)
        return ;
    commander->publish_statu();
}
void Command_manager::publish_statu()
{
    while(m_publish_exit_condition.wait_for_millisecond(50)==false)
    {
        message::Central_controller_statu_msg msg;
        message::Base_info baseInfo;
        baseInfo.set_msg_type(message::eCentral_controller_statu_msg);
        baseInfo.set_sender(message::eMain);
        baseInfo.set_receiver(message::eEmpty);
        msg.mutable_base_info()->CopyFrom(baseInfo);
        for(int i=0;i<m_system_setting.entrance_num();++i)
        {
            message::Entrance_statu entrance_statu;
            entrance_statu=message::Entrance_statu (m_input_entrance_statu[i]);
            msg.add_entrance_statu_vector(entrance_statu);
        }

        for(int i=0;i<m_system_setting.export_num();++i)
        {
            message::Entrance_statu entrance_statu;
            entrance_statu=message::Entrance_statu (m_input_entrance_statu[i]);
            msg.add_export_statu_vector(entrance_statu);
        }

        System_communicator::get_instance_pointer()->post_central_statu(msg);
    }
}