// // Created by zx on 2020/7/21. // #include "notify_excutor.h" #include "uniq_key.h" Notify_excutor::~Notify_excutor(){} Error_manager Notify_excutor::notify_request(message::Notify_request_msg& request, message::Notify_response_msg& result,Thread_condition& cancel_condition) { /* * 检查request合法性,以及模块状态 */ if(request.base_info().sender()!=message::eMain||request.base_info().receiver()!=message::eNotify) return Error_manager(ERROR,MINOR_ERROR,"notify request invalid"); request.set_command_key(create_key()); if(m_response_table.find(request.command_key())==true) return Error_manager(ERROR,MAJOR_ERROR," notify request repeated"); //设置超时,若没有设置,默认60000000 1000分钟 int timeout=request.base_info().has_timeout_ms()?request.base_info().timeout_ms():60000000; //向测量节点发送测量请求,并记录请求 Error_manager code; Communication_message message; message::Base_info base_msg; base_msg.set_msg_type(message::eNotify_request_msg); base_msg.set_sender(message::eMain); base_msg.set_receiver(message::eNotify); base_msg.set_timeout_ms(timeout); message.reset(base_msg,request.SerializeAsString()); //发送请求 m_response_table[request.command_key()]=message::Notify_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::Notify_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::eNotify_response_msg) { return Error_manager(ERROR, CRITICAL_ERROR, "notify response basemsg type error"); } //检查基本信息是否匹配 if (response_base.sender() != message::eNotify || response_base.receiver() != message::eMain || response.command_key() != request.command_key()) { return Error_manager(ERROR, MAJOR_ERROR, "notify response basemsg info error"); } result = response; m_response_table.erase(request.command_key()); return SUCCESS; } } else { //未查询到记录,任务已经被提前取消,记录被删除 return Error_manager(TASK_CANCEL,MINOR_ERROR,"notify 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,"notify 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,"notify request task canceled"); } /* * 提前取消请求 */ Error_manager Notify_excutor::cancel_request(message::Notify_request_msg& request) { message::Notify_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 Notify_excutor::check_node_statu() { std::chrono::system_clock::time_point time_now=std::chrono::system_clock::now(); auto durantion=time_now-m_notify_statu_time; if(m_notify_statu_msg.has_base_info()== false || durantion>std::chrono::seconds(5) || m_notify_statu_msg.has_error_manager()==false) { return Error_manager(DISCONNECT,MINOR_ERROR,"取车等候区通知节点通讯断开"); } if(m_notify_statu_msg.error_manager().error_code()!=SUCCESS) { return Error_manager(ERROR,MINOR_ERROR,"取车等候区通知节点故障"); } return SUCCESS; } Notify_excutor::Notify_excutor() { } Error_manager Notify_excutor::consume_msg(Communication_message* p_msg) { if(p_msg== nullptr) return Error_manager(POINTER_IS_NULL,CRITICAL_ERROR,"notify response msg pointer is null"); //response消息 switch (p_msg->get_message_type()) { ///测量结果反馈消息 case Communication_message::eNotify_response_msg: { message::Notify_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,"notify response without request"); } break; } ///测量系统状态 case Communication_message::eNotify_status_msg: { message::Notify_status_msg statu_msg; if(statu_msg.ParseFromString(p_msg->get_message_buf())==false) { return Error_manager(ERROR,CRITICAL_ERROR,"取车等候区通知节点状态消息解析失败"); } /*std::cout<<"plc msg:"<