// // Created by zx on 2020/7/21. // #include "dispatch_excutor.h" #include "uniq_key.h" Dispatch_excutor::~Dispatch_excutor(){} Error_manager Dispatch_excutor::dispatch_request(message::Dispatch_request_msg& request, message::Dispatch_response_msg& result,Thread_condition& cancel_condition) { /* * 检查request合法性,以及模块状态 */ if(request.base_info().sender()!=message::eMain||request.base_info().receiver()!=message::eDispatch) return Error_manager(ERROR,MINOR_ERROR,"dispatch request invalid"); request.set_command_key(create_key()); if(m_response_table.find(request.command_key())==true) return Error_manager(ERROR,MAJOR_ERROR," dispatch request repeated"); //设置超时,若没有设置,默认300000 五分钟 int timeout=request.base_info().has_timeout_ms()?request.base_info().timeout_ms():300000; //向测量节点发送测量请求,并记录请求 Error_manager code; Communication_message message; message::Base_info base_msg; base_msg.set_msg_type(message::eDispatch_request_msg); base_msg.set_sender(message::eMain); base_msg.set_receiver(message::eDispatch); base_msg.set_timeout_ms(timeout); message.reset(base_msg,request.SerializeAsString()); //发送请求 m_response_table[request.command_key()]=message::Dispatch_response_msg(); code= Message_communicator::get_instance_pointer()->send_msg(&message); if(code!=SUCCESS) { m_response_table.erase(request.command_key()); return code; } //循环查询请求是否被处理 auto start_time=std::chrono::system_clock::now(); double time=0; do{ //查询到记录 message::Dispatch_response_msg response; ///查询是否存在,并且删除该记录, if(m_response_table.find(request.command_key(),response)) { //判断是否接收到回应,若回应信息被赋值则证明有回应 if (response.has_base_info() && response.has_command_key()) { message::Base_info response_base = response.base_info(); //检查类型是否匹配 if (response_base.msg_type() != message::eDispatch_response_msg) { return Error_manager(ERROR, CRITICAL_ERROR, "dispatch response basemsg type error"); } //检查基本信息是否匹配 if (response_base.sender() != message::eDispatch || response_base.receiver() != message::eMain || response.command_key() != request.command_key()) { return Error_manager(ERROR, MAJOR_ERROR, "dispatch response basemsg info error"); } result = response; m_response_table.erase(request.command_key()); return SUCCESS; } } else { //未查询到记录,任务已经被提前取消,记录被删除 return Error_manager(TASK_CANCEL,MINOR_ERROR,"dispatch request canceled"); } auto end_time=std::chrono::system_clock::now(); auto duration = std::chrono::duration_cast(end_time - start_time); time=1000.0*double(duration.count()) * std::chrono::microseconds::period::num / std::chrono::microseconds::period::den; std::this_thread::yield(); if(time>double(timeout)) { m_response_table.erase(request.command_key()); return Error_manager(RESPONSE_TIMEOUT,MINOR_ERROR,"dispatch request timeout"); } }while(cancel_condition.wait_for_ex(std::chrono::milliseconds(1))==false); m_response_table.erase(request.command_key()); return Error_manager(TASK_CANCEL,MINOR_ERROR,"dispatch request task canceled"); } /* * 提前取消请求 */ Error_manager Dispatch_excutor::cancel_request(message::Dispatch_request_msg& request) { message::Dispatch_response_msg t_response; if(m_response_table.find(request.command_key(),t_response)) { m_response_table.erase(request.command_key()); } return SUCCESS; } Error_manager Dispatch_excutor::check_entrance_statu(int terminal_id) { if(m_storing_dispatch_statu_msg_map.find(terminal_id)==false || m_storing_dispatch_statu_time_map.find(terminal_id)==false) return Error_manager(FAILED,MINOR_ERROR,"停车口调度模块状态未找到"); std::chrono::system_clock::time_point time_now=std::chrono::system_clock::now(); auto durantion=time_now-m_storing_dispatch_statu_time_map[terminal_id]; if(m_storing_dispatch_statu_msg_map[terminal_id].has_base_info()== false || durantion>std::chrono::seconds(5)) { return Error_manager(DISCONNECT,MINOR_ERROR,"调度节点通讯断开"); } if(m_storing_dispatch_statu_msg_map[terminal_id].terminal_status()==message::E_TERMINAL_FAULT) { return Error_manager(ERROR,MINOR_ERROR,"调度节点故障"); } if(m_storing_dispatch_statu_msg_map[terminal_id].terminal_status()==message::E_TERMINAL_UNKNOW) { return Error_manager(UNKNOW_STATU,MINOR_ERROR,"调度节点状态未知"); } return SUCCESS; } Error_manager Dispatch_excutor::check_export_statu(int terminal_id) { if(m_picking_dispatch_statu_msg_map.find(terminal_id)==false || m_picking_dispatch_statu_time_map.find(terminal_id)==false) return Error_manager(FAILED,MINOR_ERROR,"停车口调度模块状态未找到"); std::chrono::system_clock::time_point time_now=std::chrono::system_clock::now(); auto durantion=time_now-m_picking_dispatch_statu_time_map[terminal_id]; if(m_picking_dispatch_statu_msg_map[terminal_id].has_base_info()== false || durantion>std::chrono::seconds(5)) { return Error_manager(ERROR,MINOR_ERROR,"调度节点通讯断开"); } if(m_picking_dispatch_statu_msg_map[terminal_id].terminal_status()==message::E_TERMINAL_FAULT) { return Error_manager(ERROR,MINOR_ERROR,"调度节点故障"); } if(m_picking_dispatch_statu_msg_map[terminal_id].terminal_status()==message::E_TERMINAL_UNKNOW) { return Error_manager(ERROR,MINOR_ERROR,"调度节点状态未知"); } return SUCCESS; } Dispatch_excutor::Dispatch_excutor() { } Error_manager Dispatch_excutor::consume_msg(Communication_message* p_msg) { if(p_msg== nullptr) return Error_manager(POINTER_IS_NULL,CRITICAL_ERROR,"dispatch response msg pointer is null"); //测量response消息 switch (p_msg->get_message_type()) { ///测量结果反馈消息 case Communication_message::eDispatch_response_msg: { message::Dispatch_response_msg response; response.ParseFromString(p_msg->get_message_buf()); ///查询请求表是否存在,并且更新 if(m_response_table.find_update(response.command_key(),response)==false) { return Error_manager(ERROR,NEGLIGIBLE_ERROR,"dispatch response without request"); } break; } ///测量系统状态 case Communication_message::eDispatch_status_msg: { message::Dispatch_terminal_status_msg statu_msg; if(statu_msg.ParseFromString(p_msg->get_message_buf())==false) { return Error_manager(ERROR,CRITICAL_ERROR,"调度模块状态消息解析失败"); } if(statu_msg.passageway_direction()==message::E_INLET) { m_storing_dispatch_statu_msg_map[statu_msg.terminal_id()]=statu_msg; m_storing_dispatch_statu_time_map[statu_msg.terminal_id()]=std::chrono::system_clock::now(); } if(statu_msg.passageway_direction()==message::E_OUTLET) { m_picking_dispatch_statu_msg_map[statu_msg.terminal_id()]=statu_msg; m_picking_dispatch_statu_time_map[statu_msg.terminal_id()]=std::chrono::system_clock::now(); } if(statu_msg.passageway_direction()==message::E_BILATERAL) { m_storing_dispatch_statu_msg_map[statu_msg.terminal_id()]=statu_msg; m_storing_dispatch_statu_time_map[statu_msg.terminal_id()]=std::chrono::system_clock::now(); m_picking_dispatch_statu_msg_map[statu_msg.terminal_id()]=statu_msg; m_picking_dispatch_statu_time_map[statu_msg.terminal_id()]=std::chrono::system_clock::now(); } break; } } return SUCCESS; }