Terminal_process_CT/system/command_manager.cpp

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

#include "command_manager.h"
#include "StoreProcessTask.h"
#include "PickupProcessTask.h"
#include "system_communicator.h"

Command_manager::Command_manager()
    :m_publish_statu_thread(nullptr)
    ,m_thread_queue_process(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()
{
    //创建线程池
    if(m_thread_queue_process== nullptr)
    {
        m_thread_queue_process=tq::TQFactory::CreateDefaultQueue();
        m_thread_queue_process->Start(12);
    }

    m_publish_exit_condition.reset(false, false, false);
    ///创建状态发布线程
    if(m_publish_statu_thread== nullptr)
    {
        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_publish_statu_thread==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())
        {
            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);

                        StoreProcessTask* task=new StoreProcessTask();
                        int command_id=request.terminal_id()+10000;
                        //初始化流程
                        task->init_task(command_id,request.terminal_id(),locate_info,request.car_info());
                        //获取车位
                        Error_manager code=task->alloc_space();
                        if(code==SUCCESS)
                        {
                            //更新状态
                            Command_manager::get_instance_pointer()->updata_store_entrance_statu(request.terminal_id(),message::eWorking);
                            m_thread_queue_process->AddTask(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;

                    }
                }
            }

        }

        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_publish_statu_thread==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());

        Error_manager code;
        PickupProcessTask *task = new PickupProcessTask();
        int command_id = request.terminal_id()+20000;
        //初始化流程
        task->init_task(command_id, request.terminal_id(), request.car_info());
        /////查询车位
        code=task->search_space();
        message::Error_manager error_msg;
        error_msg.set_error_code(0);
        if(code==SUCCESS)
        {
            //更新状态
            Command_manager::get_instance_pointer()->updata_pickup_entrance_statu(request.terminal_id(),message::eWorking);
            m_thread_queue_process->AddTask(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::updata_store_entrance_statu(int command_id,message::Entrance_statu  msg)
{
    if(command_id<0 || command_id>100)
        return;
    ///
    std::lock_guard<std::mutex> lk(m_entrance_msg_lock);
    auto it=m_entrance_status.mutable_store_msg_map()->find(command_id);
    if(it==m_entrance_status.mutable_store_msg_map()->end())
    {
        m_entrance_status.mutable_store_msg_map()->insert(google::protobuf::MapPair<int,message::Entrance_statu >(command_id,msg));
    }
    else
    {
        it->second=msg;
    }
    //立刻发送
    System_communicator::get_instance_pointer()->send_entrance_statu(m_entrance_status);
}
void Command_manager::updata_pickup_entrance_statu(int command_id,message::Entrance_statu statu)
{
    if(command_id<0 || command_id>100)
        return;
    ///
    std::lock_guard<std::mutex> lk(m_entrance_msg_lock);
    auto it=m_entrance_status.mutable_pickup_msg_map()->find(command_id);
    if(it==m_entrance_status.mutable_pickup_msg_map()->end())
    {
        m_entrance_status.mutable_pickup_msg_map()->insert(google::protobuf::MapPair<int,message::Entrance_statu>(command_id,statu));
    }
    else
    {
        it->second=statu;
    }
    System_communicator::get_instance_pointer()->send_entrance_statu(m_entrance_status);
}

/*
 * 发布状态线程
 */
void Command_manager::publish_thread_func(Command_manager* p_commander)
{
    if(p_commander)
    {
        p_commander->publish_status();
    }
}
void Command_manager::publish_status()
{
    //未收到退出信号
    while(false==m_publish_exit_condition.wait_for_ex(std::chrono::milliseconds(100)))
    {
        /*
         * 通过communicator 发布状态
         */
        if(System_communicator::get_instance_pointer())
        {
            if(m_entrance_status.has_base_info()==false)
            {
                message::Base_info baseInfo;
                baseInfo.set_msg_type(message::eEntrance_statu_msg);
                baseInfo.set_sender(message::eMain);
                baseInfo.set_receiver(message::eEmpty); //所有对象都可接收
                m_entrance_status.mutable_base_info()->CopyFrom(baseInfo);
            }
            System_communicator::get_instance_pointer()->send_entrance_statu(m_entrance_status);
        }

    }
}