command_manager.cpp 8.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253
  1. //
  2. // Created by zx on 2020/7/14.
  3. //
  4. #include "command_manager.h"
  5. #include "StoreProcessTask.h"
  6. #include "PickupProcessTask.h"
  7. #include "system_communicator.h"
  8. Command_manager::Command_manager()
  9. :m_publish_statu_thread(nullptr)
  10. ,m_thread_queue_process(nullptr)
  11. {
  12. }
  13. Command_manager::~Command_manager()
  14. {
  15. //等待线程池完成
  16. if(m_thread_queue_process!=nullptr) {
  17. m_thread_queue_process->WaitForFinish();
  18. m_thread_queue_process->Stop();
  19. }
  20. //退出线程
  21. m_publish_exit_condition.set_pass_ever(true);
  22. if(m_publish_statu_thread!= nullptr)
  23. {
  24. if(m_publish_statu_thread->joinable())
  25. {
  26. m_publish_statu_thread->join();
  27. }
  28. delete m_publish_statu_thread;
  29. m_publish_statu_thread=nullptr;
  30. }
  31. }
  32. Error_manager Command_manager::init()
  33. {
  34. //创建线程池
  35. if(m_thread_queue_process== nullptr)
  36. {
  37. m_thread_queue_process=tq::TQFactory::CreateDefaultQueue();
  38. m_thread_queue_process->Start(12);
  39. }
  40. m_publish_exit_condition.reset(false, false, false);
  41. ///创建状态发布线程
  42. if(m_publish_statu_thread== nullptr)
  43. {
  44. m_publish_statu_thread=new std::thread(publish_thread_func,this);
  45. }
  46. return SUCCESS;
  47. }
  48. /*
  49. * 执行停车请求
  50. */
  51. Error_manager Command_manager::execute_store_command(message::Store_command_request_msg& request,message::Store_command_response_msg& response)
  52. {
  53. if(m_publish_statu_thread==nullptr)
  54. {
  55. return Error_manager(ERROR,CRITICAL_ERROR,"线程池未初始化,bug");
  56. }
  57. if(request.base_info().msg_type()==message::eStore_command_request_msg
  58. &&request.base_info().receiver()==message::eMain
  59. &&request.base_info().sender()==message::eTerminor)
  60. {
  61. if(request.has_locate_information())
  62. {
  63. message::Locate_information locate_info=request.locate_information();
  64. if(locate_info.has_locate_correct())
  65. {
  66. if(locate_info.locate_correct()==true)
  67. {
  68. if(locate_info.has_locate_width()&&locate_info.has_locate_height()
  69. &&locate_info.has_locate_x()&&locate_info.has_locate_y()
  70. &&locate_info.has_locate_angle()&&locate_info.has_locate_wheel_base())
  71. {
  72. message::Base_info base_info;
  73. base_info.set_msg_type(message::eStore_command_response_msg);
  74. base_info.set_sender(message::eMain);
  75. base_info.set_receiver(message::eTerminor);
  76. response.mutable_base_info()->CopyFrom(base_info);
  77. response.set_terminal_id(request.terminal_id());
  78. message::Error_manager error_msg;
  79. error_msg.set_error_code(0);
  80. StoreProcessTask* task=new StoreProcessTask();
  81. int command_id=rand()%900000+100000;
  82. //初始化流程
  83. task->init_task(command_id,request.terminal_id(),locate_info,request.car_info());
  84. //获取车位
  85. Error_manager code=task->alloc_space();
  86. if(code==SUCCESS)
  87. {
  88. //清理当前状态
  89. /////
  90. std::lock_guard<std::mutex> lk(m_entrance_msg_lock);
  91. auto it=m_entrance_status.mutable_store_msg_map()->find(command_id);
  92. if(it!=m_entrance_status.mutable_store_msg_map()->end()) {
  93. it->second=message::Process_store_statu();
  94. }
  95. m_thread_queue_process->AddTask(task);
  96. response.mutable_code()->CopyFrom(error_msg);
  97. return SUCCESS;
  98. }
  99. error_msg.set_error_code(code.get_error_code());
  100. error_msg.set_error_description(code.to_string());
  101. response.mutable_code()->CopyFrom(error_msg);
  102. LOG(ERROR)<<"创建停车流程失败(车位分配失败),终端:"<<request.terminal_id()<<
  103. "车牌:"<<request.car_info().license()<<" "<<code.to_string();
  104. return code;
  105. }
  106. }
  107. }
  108. }
  109. return Error_manager(FAILED,MINOR_ERROR,"创建停车流程失败......");
  110. }
  111. else
  112. {
  113. return Error_manager(INVALID_MESSAGE,MAJOR_ERROR,"停车请求基本信息错误");
  114. }
  115. }
  116. /*
  117. * 执行取车请求
  118. */
  119. Error_manager Command_manager::execute_pickup_command(message::Pickup_command_request_msg& request,message::Pickup_command_response_msg& response)
  120. {
  121. if(m_publish_statu_thread==nullptr)
  122. {
  123. return Error_manager(ERROR,CRITICAL_ERROR,"线程池未初始化,bug");
  124. }
  125. if(request.base_info().msg_type()==message::ePickup_command_request_msg
  126. &&request.base_info().receiver()==message::eMain
  127. &&request.base_info().sender()==message::eTerminor
  128. &&request.has_car_info()) {
  129. message::Base_info baseInfo;
  130. baseInfo.set_msg_type(message::ePickup_command_response_msg);
  131. baseInfo.set_sender(message::eMain);
  132. baseInfo.set_receiver(message::eTerminor);
  133. response.mutable_base_info()->CopyFrom(baseInfo);
  134. Error_manager code;
  135. PickupProcessTask *task = new PickupProcessTask();
  136. int command_id = rand() % 900000 + 100000;
  137. //初始化流程
  138. task->init_task(command_id, request.terminal_id(), request.car_info());
  139. /////查询车位
  140. code=task->search_space();
  141. message::Error_manager error_msg;
  142. error_msg.set_error_code(0);
  143. if(code==SUCCESS)
  144. {
  145. /////
  146. std::lock_guard<std::mutex> lk(m_entrance_msg_lock);
  147. auto it=m_entrance_status.mutable_pickup_msg_map()->find(command_id);
  148. if(it!=m_entrance_status.mutable_pickup_msg_map()->end()) {
  149. it->second=message::Process_pickup_statu();
  150. }
  151. m_thread_queue_process->AddTask(task);
  152. response.mutable_code()->CopyFrom(error_msg);
  153. return SUCCESS;
  154. }
  155. error_msg.set_error_code(code.get_error_code());
  156. error_msg.set_error_description(code.to_string());
  157. response.mutable_code()->CopyFrom(error_msg);
  158. LOG(ERROR)<<"创建取车流程失败(车位查询失败),终端:"<<request.terminal_id()<<
  159. "车牌:"<<request.car_info().license()<<" "<<code.to_string();
  160. return code;
  161. }
  162. else
  163. {
  164. return Error_manager(INVALID_MESSAGE,MAJOR_ERROR,"停车请求信息错误");
  165. }
  166. }
  167. /*
  168. * 更新状态
  169. */
  170. void Command_manager::updata_entrance_statu(int command_id,message::Process_store_statu& msg)
  171. {
  172. if(command_id<0 || command_id>100)
  173. return;
  174. ///
  175. std::lock_guard<std::mutex> lk(m_entrance_msg_lock);
  176. auto it=m_entrance_status.mutable_store_msg_map()->find(command_id);
  177. if(it==m_entrance_status.mutable_store_msg_map()->end())
  178. {
  179. m_entrance_status.mutable_store_msg_map()->insert(google::protobuf::MapPair<int,message::Process_store_statu>(command_id,msg));
  180. }
  181. else
  182. {
  183. it->second=msg;
  184. }
  185. }
  186. void Command_manager::updata_entrance_statu(int command_id,message::Process_pickup_statu& msg)
  187. {
  188. if(command_id<0 || command_id>100)
  189. return;
  190. ///
  191. std::lock_guard<std::mutex> lk(m_entrance_msg_lock);
  192. auto it=m_entrance_status.mutable_pickup_msg_map()->find(command_id);
  193. if(it==m_entrance_status.mutable_pickup_msg_map()->end())
  194. {
  195. m_entrance_status.mutable_pickup_msg_map()->insert(google::protobuf::MapPair<int,message::Process_pickup_statu>(command_id,msg));
  196. }
  197. else
  198. {
  199. it->second=msg;
  200. }
  201. }
  202. /*
  203. * 发布状态线程
  204. */
  205. void Command_manager::publish_thread_func(Command_manager* p_commander)
  206. {
  207. if(p_commander)
  208. {
  209. p_commander->publish_status();
  210. }
  211. }
  212. void Command_manager::publish_status()
  213. {
  214. //未收到退出信号
  215. while(false==m_publish_exit_condition.wait_for_ex(std::chrono::milliseconds(100)))
  216. {
  217. /*
  218. * 通过communicator 发布状态
  219. */
  220. if(System_communicator::get_instance_pointer())
  221. {
  222. if(m_entrance_status.has_base_info()==false)
  223. {
  224. message::Base_info baseInfo;
  225. baseInfo.set_msg_type(message::eEntrance_statu_msg);
  226. baseInfo.set_sender(message::eMain);
  227. baseInfo.set_receiver(message::eEmpty); //所有对象都可接收
  228. }
  229. System_communicator::get_instance_pointer()->send_entrance_statu(m_entrance_status);
  230. }
  231. }
  232. }