main.cpp 8.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319
  1. #include <QCoreApplication>
  2. #include <yaml-cpp/yaml.h>
  3. #include <QDateTime>
  4. #include <iostream>
  5. #include <vector>
  6. #include <memory>
  7. #include <QMutex>
  8. #include <thread>
  9. #include "modulecomm.h"
  10. #include "rpc_client.hpp"
  11. #include "cloud.pb.h"
  12. std::string gstrserverip = "123.57.212.138";
  13. std::string gstrserverport = "9000";
  14. std::string gstruploadinterval = "1000";
  15. void * gpa;
  16. QMutex gMutexMsg;
  17. std::thread * guploadthread;
  18. namespace iv {
  19. struct msgunit
  20. {
  21. char mstrmsgname[256];
  22. int mnBufferSize = 10000;
  23. int mnBufferCount = 1;
  24. void * mpa;
  25. std::shared_ptr<char> mpstrmsgdata;
  26. int mndatasize = 0;
  27. bool mbRefresh = false;
  28. };
  29. }
  30. std::vector<iv::msgunit> mvectormsgunit;
  31. std::vector<iv::msgunit> mvectorctrlmsgunit;
  32. std::string gstrVIN = "AAAAAAAAAAAAAAAAA";
  33. std::string gstrqueryMD5 = "5d41402abc4b2a76b9719d911017c592";
  34. std::string gstrctrlMD5 = "5d41402abc4b2a76b9719d911017c592";
  35. struct uploadresponsemsg {
  36. int nres; //0 no message 1 have message
  37. int id;
  38. qint64 ntime;
  39. std::string strVIN;
  40. std::string strctrlMD5;
  41. std::vector<char> xdata;
  42. MSGPACK_DEFINE(nres,id, ntime,strVIN, strctrlMD5,xdata);
  43. };
  44. int gindex = 0;
  45. void ListenData(const char * strdata,const unsigned int nSize,const unsigned int index,const QDateTime * dt,const char * strmemname)
  46. {
  47. int nsize = mvectormsgunit.size();
  48. int i;
  49. for(i=0;i<nsize;i++)
  50. {
  51. if(strncmp(strmemname,mvectormsgunit[i].mstrmsgname,255) == 0)
  52. {
  53. gMutexMsg.lock();
  54. char * strtem = new char[nSize];
  55. memcpy(strtem,strdata,nSize);
  56. mvectormsgunit[i].mpstrmsgdata.reset(strtem);
  57. mvectormsgunit[i].mndatasize = nSize;
  58. mvectormsgunit[i].mbRefresh = true;
  59. gMutexMsg.unlock();
  60. break;
  61. }
  62. }
  63. }
  64. void sharectrlmsg(iv::cloud::cloudmsg * pxmsg)
  65. {
  66. int i;
  67. int nsize = pxmsg->xclouddata_size();
  68. for(i=0;i<nsize;i++)
  69. {
  70. int j;
  71. int nquerysize = mvectorctrlmsgunit.size();
  72. for(j=0;j<nquerysize;j++)
  73. {
  74. if(strncmp(pxmsg->xclouddata(i).msgname().data(), mvectorctrlmsgunit[j].mstrmsgname,255) == 0)
  75. {
  76. // qDebug("size is %d ",pxmsg->xclouddata(i).data().size());
  77. iv::modulecomm::ModuleSendMsg(mvectorctrlmsgunit[j].mpa,pxmsg->xclouddata(i).data().data(),pxmsg->xclouddata(i).data().size());
  78. break;
  79. }
  80. }
  81. }
  82. }
  83. void threadupload()
  84. {
  85. int nsize = mvectormsgunit.size();
  86. int i;
  87. int ninterval = atoi(gstruploadinterval.data());
  88. if(ninterval<=0)ninterval = 100;
  89. QTime xTime;
  90. xTime.start();
  91. int nlastsend = xTime.elapsed();
  92. bool r;
  93. rest_rpc::rpc_client client;
  94. client.enable_auto_reconnect(); //automatic reconnect
  95. client.enable_auto_heartbeat(); //automatic heartbeat
  96. // bool r = client.connect("140.143.237.38", 9000);
  97. r = client.connect(gstrserverip, atoi(gstrserverport.data()));
  98. int count = 0;
  99. CT:
  100. while (true) {
  101. if (client.has_connected()) {
  102. std::cout << "connected ok\n";
  103. break;
  104. }
  105. else {
  106. std::cout << "connected failed: "<< count++<<"\n";
  107. }
  108. std::this_thread::sleep_for(std::chrono::seconds(1));
  109. }
  110. while(true)
  111. {
  112. std::this_thread::sleep_for(std::chrono::milliseconds(1));
  113. if((xTime.elapsed()-nlastsend)<ninterval)
  114. {
  115. continue;
  116. }
  117. try
  118. {
  119. iv::cloud::cloudmsg xmsg;
  120. xmsg.set_xtime(QDateTime::currentMSecsSinceEpoch());
  121. gMutexMsg.lock();
  122. for(i=0;i<nsize;i++)
  123. {
  124. if(mvectormsgunit[i].mbRefresh)
  125. {
  126. mvectormsgunit[i].mbRefresh = false;
  127. iv::cloud::cloudunit xcloudunit;
  128. xcloudunit.set_msgname(mvectormsgunit[i].mstrmsgname);
  129. xcloudunit.set_data(mvectormsgunit[i].mpstrmsgdata.get(),mvectormsgunit[i].mndatasize);
  130. iv::cloud::cloudunit * pcu = xmsg.add_xclouddata();
  131. pcu->CopyFrom(xcloudunit);
  132. }
  133. }
  134. gMutexMsg.unlock();
  135. int nbytesize = xmsg.ByteSize();
  136. std::vector<char> pvectordata;
  137. pvectordata.resize(nbytesize);
  138. if(xmsg.SerializeToArray(pvectordata.data(),nbytesize))
  139. {
  140. uploadresponsemsg result = client.call<60000,uploadresponsemsg>("get_clientresp",gindex,QDateTime::currentMSecsSinceEpoch(),
  141. gstrVIN,gstrqueryMD5,gstrctrlMD5,pvectordata);
  142. // qDebug("res is %d ",result.nres);
  143. if(result.nres == 1)
  144. {
  145. iv::cloud::cloudmsg xmsg;
  146. if(xmsg.ParseFromArray(result.xdata.data(),result.xdata.size()))
  147. {
  148. sharectrlmsg(&xmsg);
  149. }
  150. }
  151. gindex++;
  152. }
  153. nlastsend = xTime.elapsed();
  154. pvectordata.clear();
  155. }
  156. catch (const std::exception & e) {
  157. std::cout << e.what() << std::endl;
  158. // client.close();
  159. goto CT;
  160. }
  161. }
  162. }
  163. void dec_yaml(const char * stryamlpath)
  164. {
  165. YAML::Node config;
  166. try
  167. {
  168. config = YAML::LoadFile(stryamlpath);
  169. }
  170. catch(YAML::BadFile e)
  171. {
  172. qDebug("load error.");
  173. return;
  174. }
  175. std::vector<std::string> vecmodulename;
  176. if(config["server"])
  177. {
  178. gstrserverip = config["server"].as<std::string>();
  179. }
  180. if(config["port"])
  181. {
  182. gstrserverport = config["port"].as<std::string>();
  183. }
  184. if(config["uploadinterval"])
  185. {
  186. gstruploadinterval = config["uploadinterval"].as<std::string>();
  187. }
  188. std::string strmsgname;
  189. if(config["uploadmessage"])
  190. {
  191. for(YAML::const_iterator it= config["uploadmessage"].begin(); it != config["uploadmessage"].end();++it)
  192. {
  193. std::string strtitle = it->first.as<std::string>();
  194. std::cout<<strtitle<<std::endl;
  195. if(config["uploadmessage"][strtitle]["msgname"]&&config["uploadmessage"][strtitle]["buffersize"]&&config["uploadmessage"][strtitle]["buffercount"])
  196. {
  197. iv::msgunit xmu;
  198. strmsgname = config["uploadmessage"][strtitle]["msgname"].as<std::string>();
  199. strncpy(xmu.mstrmsgname,strmsgname.data(),255);
  200. xmu.mnBufferSize = config["uploadmessage"][strtitle]["buffersize"].as<int>();
  201. xmu.mnBufferCount = config["uploadmessage"][strtitle]["buffercount"].as<int>();
  202. mvectormsgunit.push_back(xmu);
  203. }
  204. }
  205. }
  206. else
  207. {
  208. }
  209. if(config["ctrlmessage"])
  210. {
  211. std::string strnodename = "ctrlmessage";
  212. for(YAML::const_iterator it= config[strnodename].begin(); it != config[strnodename].end();++it)
  213. {
  214. std::string strtitle = it->first.as<std::string>();
  215. std::cout<<strtitle<<std::endl;
  216. if(config[strnodename][strtitle]["msgname"]&&config[strnodename][strtitle]["buffersize"]&&config[strnodename][strtitle]["buffercount"])
  217. {
  218. iv::msgunit xmu;
  219. strmsgname = config[strnodename][strtitle]["msgname"].as<std::string>();
  220. strncpy(xmu.mstrmsgname,strmsgname.data(),255);
  221. xmu.mnBufferSize = config[strnodename][strtitle]["buffersize"].as<int>();
  222. xmu.mnBufferCount = config[strnodename][strtitle]["buffercount"].as<int>();
  223. mvectorctrlmsgunit.push_back(xmu);
  224. }
  225. }
  226. }
  227. else
  228. {
  229. }
  230. return;
  231. }
  232. int main(int argc, char *argv[])
  233. {
  234. QCoreApplication a(argc, argv);
  235. char stryamlpath[256];
  236. if(argc<2)
  237. {
  238. snprintf(stryamlpath,255,"driver_cloud_client.yaml");
  239. // strncpy(stryamlpath,abs_ymlpath,255);
  240. }
  241. else
  242. {
  243. strncpy(stryamlpath,argv[1],255);
  244. }
  245. dec_yaml(stryamlpath);
  246. int i;
  247. for(i=0;i<mvectormsgunit.size();i++)
  248. {
  249. mvectormsgunit[i].mpa = iv::modulecomm::RegisterRecv(mvectormsgunit[i].mstrmsgname,ListenData);
  250. }
  251. for(i=0;i<mvectorctrlmsgunit.size();i++)
  252. {
  253. mvectorctrlmsgunit[i].mpa = iv::modulecomm::RegisterSend(mvectorctrlmsgunit[i].mstrmsgname,mvectorctrlmsgunit[i].mnBufferSize,
  254. mvectorctrlmsgunit[i].mnBufferCount);
  255. }
  256. guploadthread = new std::thread(threadupload);
  257. return a.exec();
  258. }