Terminal_process_CT/system/command_manager.cpp

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

#include <sstream>
#include <iomanip>
#include <parkspace_excutor.h>
#include <notify_excutor.h>
#include "command_manager.h"
#include "StoreProcessTask.h"
#include "PickupProcessTask.h"
#include "command_accepter.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_paused.resize(system_setting.entrance_num());
    for(int i=0;i<system_setting.entrance_num();++i)
        m_input_entrance_paused[i]=false;
    m_output_entrance_paused.resize(system_setting.export_num());
    for(int i=0;i<system_setting.export_num();++i)
        m_output_entrance_paused[i]=false;

    //创建线程池
    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_excutor::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 = Measure_excutor::get_instance_pointer()->check_statu(i);
                LOG_IF(INFO, locate_code == SUCCESS) << "停车入口:"<<i<<" 测量模块初始化完成!!!";
            }
            if (dispatch_code != SUCCESS) {
                dispatch_code = Dispatch_excutor::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_excutor::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(const message::Store_command_request_msg& request)
{
    message::Error_manager error_msg;
    message::Store_command_response_msg response;
    if (m_system_paused == true)
    {
        error_msg.set_error_code(PAUSE);
        error_msg.set_error_description("急停");
        response.mutable_code()->CopyFrom(error_msg);

        Communication_message msg;
        msg.reset(response.base_info(), response.SerializeAsString());
        Message_communicator::get_instance_pointer()->send_msg(&msg);
        return Error_manager(PAUSE, MINOR_ERROR, "急停");
    }

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

        Communication_message msg;
        msg.reset(response.base_info(), response.SerializeAsString());
        Message_communicator::get_instance_pointer()->send_msg(&msg);
        return Error_manager(ERROR, MINOR_ERROR, "中控entrance已停止使用");
    }



    message::Base_info base_info_response;
    base_info_response.set_msg_type(message::eStore_command_response_msg);
    base_info_response.set_sender(message::eMain);
    base_info_response.set_receiver(message::eTerminor);
    response.mutable_base_info()->CopyFrom(base_info_response);
    response.mutable_id_struct()->CopyFrom(request.id_struct());

    Error_manager code;
    if (request.base_info().msg_type() == message::eStore_command_request_msg
        && request.base_info().receiver() == message::eMain
        && request.base_info().sender() == message::eTerminor) {
//        LOG(WARNING) << request.DebugString();
        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())
                    {

                        LOG(INFO) << "------  停  ------- 收到停车,车牌:" << request.car_info().license() <<
                                  ",终端:" << request.id_struct().terminal_id() << "............................";
                        //check hardwared
                        Error_manager parkspace_code = Parkspace_excutor::get_instance_pointer()->check_statu();
                        Error_manager dispatch_code = Dispatch_excutor::get_instance_pointer()->check_entrance_statu(request.id_struct().unit_id());
                        if(parkspace_code!=SUCCESS)
                        {
                            error_msg.set_error_code(parkspace_code.get_error_code());
                            error_msg.set_error_description(parkspace_code.to_string());
                            response.mutable_code()->CopyFrom(error_msg);
                            LOG(ERROR) << "xxxx 创建停车流程失败:" << request.id_struct().terminal_id() <<
                                       "车牌:" << request.car_info().license() << "  " << parkspace_code.to_string();
                            Communication_message msg;
                            msg.reset(response.base_info(), response.SerializeAsString());
                            Message_communicator::get_instance_pointer()->send_msg(&msg);
                            return parkspace_code;
                        }
                        if(dispatch_code!=SUCCESS)
                        {
                            error_msg.set_error_code(dispatch_code.get_error_code());
                            error_msg.set_error_description(dispatch_code.to_string());
                            response.mutable_code()->CopyFrom(error_msg);
                            LOG(ERROR) << "xxxx 创建停车流程失败:" << request.id_struct().terminal_id() <<
                                       "车牌:" << request.car_info().license() << "  " << dispatch_code.to_string();
                            Communication_message msg;
                            msg.reset(response.base_info(), response.SerializeAsString());
                            Message_communicator::get_instance_pointer()->send_msg(&msg);
                            return dispatch_code;
                        }


                        /*
                         * 检查消息完毕,开始处理
                         */

                        Process_task *ptask = new StoreProcessTask(request.id_struct(), request.car_info());
                        StoreProcessTask *pStore_task = (StoreProcessTask *) ptask;
                        //初始化流程
                        code = pStore_task->init_task(locate_info);
                        if (code != SUCCESS)
                        {
                            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) << "xxxx 创建停车流程失败(车位分配失败),终端:" << request.id_struct().terminal_id() <<
                                       "车牌:" << request.car_info().license() << "  " << code.to_string();

                            Communication_message msg;
                            msg.reset(response.base_info(), response.SerializeAsString());
                            Message_communicator::get_instance_pointer()->send_msg(&msg);
                            return code;
                        }
                        m_thread_queue_process->AddTask(pStore_task);
                        return SUCCESS;

                    }
                }
            }

        }
        error_msg.set_error_code(FAILED);
        error_msg.set_error_description("创建停车流程失败...指令缺少必要信息...");
        response.mutable_code()->CopyFrom(error_msg);

        Communication_message msg;
        msg.reset(response.base_info(), response.SerializeAsString());
        Message_communicator::get_instance_pointer()->send_msg(&msg);

        return Error_manager(FAILED, MINOR_ERROR, "创建停车流程失败...指令缺少必要信息...");
    } else {
        error_msg.set_error_code(INVALID_MESSAGE);
        error_msg.set_error_description("停车请求基本信息错误");
        response.mutable_code()->CopyFrom(error_msg);

        Communication_message msg;
        msg.reset(response.base_info(),response.SerializeAsString());
        Message_communicator::get_instance_pointer()->send_msg(&msg);
        return Error_manager(INVALID_MESSAGE, MAJOR_ERROR, "停车请求基本信息错误");
    }
}


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

    message::Base_info base_info_response;
    base_info_response.set_msg_type(message::ePickup_command_response_msg);
    base_info_response.set_sender(message::eMain);
    base_info_response.set_receiver(message::eTerminor);
    response.mutable_base_info()->CopyFrom(base_info_response);

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

        Communication_message msg;
        msg.reset(response.base_info(), response.SerializeAsString());
        Message_communicator::get_instance_pointer()->send_msg(&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);

        Communication_message msg;
        msg.reset(response.base_info(), response.SerializeAsString());
        Message_communicator::get_instance_pointer()->send_msg(&msg);
        return Error_manager(PAUSE, MINOR_ERROR, "急停");
    }

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

        Communication_message msg;
        msg.reset(response.base_info(), response.SerializeAsString());
        Message_communicator::get_instance_pointer()->send_msg(&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.mutable_id_struct()->CopyFrom(request.id_struct());
        /*
        * 检查各个节点是否正常
        */
        Error_manager parkspace_code = Parkspace_excutor::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);

            Communication_message msg;
            msg.reset(response.base_info(), response.SerializeAsString());
            Message_communicator::get_instance_pointer()->send_msg(&msg);
            return parkspace_code;
        }

        Error_manager dispatch_code = Dispatch_excutor::get_instance_pointer()->check_export_statu(request.id_struct().unit_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);

            Communication_message msg;
            msg.reset(response.base_info(), response.SerializeAsString());
            Message_communicator::get_instance_pointer()->send_msg(&msg);
            return dispatch_code;
        }
        //检查取车口等候区通知节点
        Error_manager notify_code=Notify_excutor::get_instance_pointer()->check_node_statu();
        if(notify_code!=SUCCESS)
        {
            error_msg.set_error_code(notify_code.get_error_code());
            error_msg.set_error_description(notify_code.get_error_description());
            response.mutable_code()->CopyFrom(error_msg);

            Communication_message msg;
            msg.reset(response.base_info(), response.SerializeAsString());
            Message_communicator::get_instance_pointer()->send_msg(&msg);
            return notify_code;
        }

        //一切正常,接受指令
        Error_manager code;
        LOG(WARNING) << "---------  取  -------收到取车-----------------------------" << request.car_info().license();
        Process_task *ptask = new PickupProcessTask(request.id_struct(), request.car_info());
        PickupProcessTask *pPick_task = (PickupProcessTask *) ptask;
        //初始化流程
        code = pPick_task->init_task(request);

        if (code == SUCCESS)
        {
            m_thread_queue_process->AddTask(pPick_task);
            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.id_struct().unit_id() <<
                   "车牌:" << request.car_info().license() << "  " << code.to_string();
        delete pPick_task;
        Communication_message msg;
        msg.reset(response.base_info(), response.SerializeAsString());
        Message_communicator::get_instance_pointer()->send_msg(&msg);
        return code;
    }
    else
    {
        error_msg.set_error_code(INVALID_MESSAGE);
        error_msg.set_error_description("停车请求信息错误");
        response.mutable_code()->CopyFrom(error_msg);
        Communication_message msg;
        msg.reset(response.base_info(), response.SerializeAsString());
        Message_communicator::get_instance_pointer()->send_msg(&msg);
        return Error_manager(INVALID_MESSAGE, MAJOR_ERROR, "停车请求信息错误");
    }
}

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

message::Entrance_statu Command_manager::get_entrance_statu(int terminal_id)
{
    message::Entrance_statu entrance_statu;

    if(terminal_id<0 || terminal_id>=m_system_setting.entrance_num())
        return entrance_statu;

    message::Module_statu  statu=message::eConnected;
    Error_manager code=Parkspace_excutor::get_instance_pointer()->check_statu();
    statu=(code==SUCCESS)?message::eConnected:message::eFault;
    entrance_statu.set_parkspace_statu(statu);


    code=Dispatch_excutor::get_instance_pointer()->check_entrance_statu(terminal_id);
    if (code==ERROR) statu=message::eFault;
    if(code==DISCONNECT) statu=message::eDisconnected;
    entrance_statu.set_dispatch_statu(statu);

    entrance_statu.set_paused(m_input_entrance_paused[terminal_id]);

    return entrance_statu;
}
message::Entrance_statu Command_manager::get_export_statu(int terminal_id)
{
    message::Entrance_statu entrance_statu;
    if(terminal_id<0 || terminal_id>=m_system_setting.export_num())
        return entrance_statu;

    message::Module_statu  statu=message::eConnected;
    Error_manager code=Parkspace_excutor::get_instance_pointer()->check_statu();
    statu=(code==SUCCESS)?message::eConnected:message::eFault;
    entrance_statu.set_parkspace_statu(statu);

    code=Dispatch_excutor::get_instance_pointer()->check_entrance_statu(terminal_id);
    if (code==ERROR) statu=message::eFault;
    if(code==DISCONNECT) statu=message::eDisconnected;
    entrance_statu.set_dispatch_statu(statu);

    code=Notify_excutor::get_instance_pointer()->check_node_statu();
    if (code==ERROR) statu=message::eFault;
    if(code==DISCONNECT) statu=message::eDisconnected;
    entrance_statu.set_notify_statu(statu);

    entrance_statu.set_paused(m_output_entrance_paused[terminal_id]);

    return entrance_statu;
}
/*
 * 控制出口 开放或者关闭
 */
Error_manager Command_manager::pause_export(int terminal_id,bool paused)
{
    if(terminal_id<0 || terminal_id>m_system_setting.export_num())
        return Error_manager(ERROR,MINOR_ERROR,"xxxx");
    m_output_entrance_paused[terminal_id]=paused;
    return SUCCESS;
}

/*
     * 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(100)==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 statu=get_entrance_statu(i);
            message::Entrance_statu* entrance_statu=msg.mutable_entrance_statu_vector()->Add();;
            entrance_statu->CopyFrom(statu);
        }

        for(int i=0;i<m_system_setting.export_num();++i)
        {
            message::Entrance_statu statu=get_export_statu(i);
            message::Entrance_statu* entrance_statu=msg.mutable_export_statu_vector()->Add();
            entrance_statu->CopyFrom(statu);

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

        //std::cout<<" 123 \n "<<msg.DebugString()<<std::endl;
    }
}