|
@@ -0,0 +1,213 @@
|
|
|
+//
|
|
|
+// Created by zx on 2020/7/21.
|
|
|
+//
|
|
|
+
|
|
|
+#include "dispatch_communicator.h"
|
|
|
+
|
|
|
+
|
|
|
+Dispatch_communicator::~Dispatch_communicator(){}
|
|
|
+
|
|
|
+Error_manager Dispatch_communicator::dispatch_request(message::Dispatch_request_msg& request,
|
|
|
+ message::Dispatch_response_msg& result)
|
|
|
+{
|
|
|
+
|
|
|
+ /*
|
|
|
+ * 检查request合法性,以及模块状态
|
|
|
+ */
|
|
|
+
|
|
|
+ if(request.base_info().sender()!=message::eMain||request.base_info().receiver()!=message::eDispatch)
|
|
|
+ return Error_manager(ERROR,MINOR_ERROR,"dispatch request invalid");
|
|
|
+
|
|
|
+ if(m_response_table.find(request)==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());
|
|
|
+
|
|
|
+ code=encapsulate_msg(&message);
|
|
|
+ if(code!=SUCCESS)
|
|
|
+ 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,response))
|
|
|
+ {
|
|
|
+ //判断是否接收到回应,若回应信息被赋值则证明有回应
|
|
|
+ if (response.has_base_info() && response.has_command_id())
|
|
|
+ {
|
|
|
+ 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_id() != request.command_id()) {
|
|
|
+ return Error_manager(ERROR, MAJOR_ERROR,
|
|
|
+ "dispatch response basemsg info error");
|
|
|
+ }
|
|
|
+ result = response;
|
|
|
+ m_response_table.erase(request);
|
|
|
+
|
|
|
+ return SUCCESS;
|
|
|
+
|
|
|
+ }
|
|
|
+ }
|
|
|
+ else
|
|
|
+ {
|
|
|
+ //未查询到记录,任务已经被提前取消,记录被删除
|
|
|
+ return Error_manager(FAILED,MINOR_ERROR,"dispatch request canceled");
|
|
|
+ }
|
|
|
+
|
|
|
+ auto end_time=std::chrono::system_clock::now();
|
|
|
+ auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);
|
|
|
+ time=1000.0*double(duration.count()) * std::chrono::microseconds::period::num / std::chrono::microseconds::period::den;
|
|
|
+ std::this_thread::yield();
|
|
|
+ usleep(1000);
|
|
|
+ }while(time<double(timeout));
|
|
|
+ m_response_table.erase(request);
|
|
|
+ return Error_manager(RESPONSE_TIMEOUT,MINOR_ERROR,"dispatch request timeout");
|
|
|
+}
|
|
|
+/*
|
|
|
+ * 提前取消请求
|
|
|
+ */
|
|
|
+Error_manager Dispatch_communicator::cancel_request(message::Dispatch_request_msg& request)
|
|
|
+{
|
|
|
+ message::Dispatch_response_msg t_response;
|
|
|
+ if(m_response_table.find(request,t_response))
|
|
|
+ {
|
|
|
+ m_response_table.erase(request);
|
|
|
+ }
|
|
|
+ return SUCCESS;
|
|
|
+}
|
|
|
+Error_manager Dispatch_communicator::check_statu()
|
|
|
+{
|
|
|
+ return SUCCESS;
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+Dispatch_communicator::Dispatch_communicator()
|
|
|
+{
|
|
|
+}
|
|
|
+Error_manager Dispatch_communicator::encapsulate_msg(Communication_message* message)
|
|
|
+{
|
|
|
+ Error_manager code;
|
|
|
+ //记录请求
|
|
|
+ switch (message->get_message_type())
|
|
|
+ {
|
|
|
+ case Communication_message::eDispatch_request_msg:
|
|
|
+ {
|
|
|
+ message::Dispatch_request_msg request;
|
|
|
+ if(false==request.ParseFromString(message->get_message_buf()))
|
|
|
+ {
|
|
|
+ code=Error_manager(ERROR,CRITICAL_ERROR,"request message parse failed");
|
|
|
+ }
|
|
|
+ m_response_table[request]=message::Dispatch_response_msg();
|
|
|
+ //发送请求
|
|
|
+ code= Communication_socket_base::encapsulate_msg(message);
|
|
|
+ if(code!=SUCCESS)
|
|
|
+ {
|
|
|
+ m_response_table.erase(request);
|
|
|
+ }
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ default:
|
|
|
+ code= Error_manager(FAILED,CRITICAL_ERROR," measure发送任务类型不存在");
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ return code;
|
|
|
+}
|
|
|
+Error_manager Dispatch_communicator::execute_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());
|
|
|
+ message::Dispatch_request_msg request=create_request_by_response(response);
|
|
|
+ ///查询请求表是否存在,并且更新
|
|
|
+ if(m_response_table.find_update(request,response)==false)
|
|
|
+ {
|
|
|
+ return Error_manager(ERROR,NEGLIGIBLE_ERROR,"dispatch response without request");
|
|
|
+ }
|
|
|
+
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ ///测量系统状态
|
|
|
+ case Communication_message::eDispatch_status_msg:
|
|
|
+ {
|
|
|
+ //m_measure_statu_msg=*p_msg;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return SUCCESS;
|
|
|
+}
|
|
|
+/*
|
|
|
+ * 检测消息是否可被处理
|
|
|
+ */
|
|
|
+Error_manager Dispatch_communicator::check_msg(Communication_message* p_msg)
|
|
|
+{
|
|
|
+ //通过 p_msg->get_message_type() 和 p_msg->get_receiver() 判断这条消息是不是给我的.
|
|
|
+ //子类重载时, 增加自己模块的判断逻辑, 以后再写.
|
|
|
+ if ( (p_msg->get_message_type() == Communication_message::Message_type::eDispatch_response_msg
|
|
|
+ ||p_msg->get_message_type() == Communication_message::Message_type::eDispatch_status_msg)
|
|
|
+ && p_msg->get_receiver() == Communication_message::Communicator::eMain )
|
|
|
+ {
|
|
|
+ return Error_code::SUCCESS;
|
|
|
+ }
|
|
|
+ else
|
|
|
+ {
|
|
|
+ //认为接受人
|
|
|
+ return Error_code::INVALID_MESSAGE;
|
|
|
+ }
|
|
|
+}
|
|
|
+/*
|
|
|
+ * 心跳发送函数,重载
|
|
|
+ */
|
|
|
+Error_manager Dispatch_communicator::encapsulate_send_data()
|
|
|
+{
|
|
|
+ return SUCCESS;
|
|
|
+}
|
|
|
+//检查消息是否可以被解析, 需要重载
|
|
|
+Error_manager Dispatch_communicator::check_executer(Communication_message* p_msg)
|
|
|
+{
|
|
|
+ return SUCCESS;
|
|
|
+}
|
|
|
+
|
|
|
+/*
|
|
|
+ * 根据接收到的response,生成对应的 request
|
|
|
+ */
|
|
|
+message::Dispatch_request_msg Dispatch_communicator::create_request_by_response(message::Dispatch_response_msg& response)
|
|
|
+{
|
|
|
+ message::Dispatch_request_msg request;
|
|
|
+ message::Base_info baseInfo;
|
|
|
+ baseInfo.set_msg_type(message::eDispatch_request_msg);
|
|
|
+ baseInfo.set_sender(response.base_info().receiver());
|
|
|
+ baseInfo.set_receiver(response.base_info().sender());
|
|
|
+ request.mutable_base_info()->CopyFrom(baseInfo);
|
|
|
+ request.set_command_id(response.command_id());
|
|
|
+ return request;
|
|
|
+}
|