LogicClient.cpp 53 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059
  1. //#include "SysIF.h"
  2. #include "LocalConfig.h"
  3. #include "LogicClient.h"
  4. #include "common_api.h"
  5. #include "PacketAnalizer.h"
  6. //#include "Logger.h"
  7. #define ATTRIBUTE ("Attribute")
  8. #include <iostream>
  9. #include<chrono>
  10. #ifndef _CRT_SECURE_NO_WARNINGS
  11. #define _CRT_SECURE_NO_WARNINGS
  12. #endif
  13. //Log4CPP::Logger* ////mLog::gLogger = nullptr;
  14. string LogHost = "";
  15. // Linux下获取进程ID
  16. inline DWORD GetCurrentProcessId() {
  17. return getpid();
  18. }
  19. inline string CurrentDateTime2()
  20. {
  21. std::time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
  22. char buf[100] = { 0 };
  23. std::strftime(buf, sizeof(buf), " %Y-%m-%d %H:%M:%S ", std::localtime(&now));
  24. return buf;
  25. }
  26. inline std::string CurrentDateTime()
  27. {
  28. // 获取当前时间点
  29. auto now = std::chrono::system_clock::now();
  30. // 将时间长度转换为微秒数
  31. auto now_us = std::chrono::duration_cast<std::chrono::microseconds>(now.time_since_epoch());
  32. // 再转成tm格式
  33. auto now_time_t = std::chrono::system_clock::to_time_t(now);
  34. auto now_tm = std::localtime(&now_time_t);
  35. // 可以直接输出到标准输出
  36. // std::cout << std::put_time(now_tm, "%Y-%m-%d %H:%M:%S.") << std::setfill('0') << std::setw(6) << now_us.count() % 1000000 << std::endl;
  37. // 格式化字符,年月日时分秒
  38. std::string now_time_str;
  39. char buf[64];
  40. std::strftime(buf, sizeof(buf), "%Y-%m-%d %H:%M:%S", now_tm);
  41. now_time_str += buf;
  42. // 格式化微秒
  43. //snprintf(buf, sizeof(buf), ".%06lld ", now_us.count() % 1000000);
  44. snprintf(buf, sizeof(buf), ".%06ld ", now_us.count() % 1000000);
  45. now_time_str += buf;
  46. //printf("%s\n", now_time_str.c_str());
  47. return now_time_str;
  48. }
  49. LogicClient::LogicClient(string szClientName, string szTransaction, string szType, bool bNeedNotify)
  50. {
  51. m_ReqIdx = 0;
  52. pLogicDeivce = NULL;
  53. m_SyncMode = ACTION_SYNC;
  54. m_FileFlags = 0;
  55. m_ClosedFlag = true;
  56. m_OwnerThreadId = 0;
  57. m_pActionName = new string();
  58. m_pFilePath = new string();
  59. m_pLastMsgInfo = new string();
  60. m_pPacketWaitQue = new MsgQueue<ResDataObject>();
  61. m_pPacketReceivedQue = new MsgQueue<ResDataObject>();
  62. m_pPacketNotifyQue = new MsgQueue<ResDataObject>();
  63. m_NotifyEvent = LinuxEvent::CreateEvent(LinuxEvent::MANUAL_RESET, false);
  64. m_ResponseEvent = LinuxEvent::CreateEvent(LinuxEvent::AUTO_RESET, false);
  65. m_bNeedNotify = bNeedNotify;
  66. //m_pSysIF = new ClientSysIF();
  67. m_pMqttConn = nullptr;
  68. m_strClientName = szClientName;
  69. if (m_strClientName.length() <= 0) {
  70. string stream;
  71. stream = "LogicClient_" + to_string( GetCurrentProcessId()) + "_" + string( (const char*)getLocalEbusId()) + "_" +to_string( GetCurrentThreadId());
  72. m_strClientName = stream;
  73. stream.clear();
  74. string stream2;
  75. stream2 = CCOS_CLIENT_ROOT_TOPIC + string( "/Resp_") + to_string( GetCurrentProcessId()) + "_" + string( (const char*)getLocalEbusId()) + "_" + to_string( GetCurrentThreadId());
  76. m_strDefaultTopic = stream2;
  77. }
  78. else
  79. {
  80. stringstream stream;
  81. stream << "LogicClient_" << szClientName << "_" << (const char*)getLocalEbusId() << "_" << GetCurrentProcessId();
  82. m_strClientName = stream.str();
  83. stream.clear();
  84. stringstream stream2;
  85. stream2 << CCOS_CLIENT_ROOT_TOPIC << (szType.length()<=0?"":"/") << (szType.length()<=0?"":szType) << "/" << szClientName << "/Resp_" << (const char*)getLocalEbusId() << "_" << GetCurrentProcessId();
  86. m_strDefaultTopic = stream2.str();
  87. }
  88. m_strCurTransaction = szTransaction;
  89. m_pFileHandle = new CcosDevFileHandle();
  90. m_ClientLock = (HANDLE)new CcosLock();
  91. m_Router = CCOS_PACKET_ROUTE_ANY;
  92. m_hDeviceOpenOK = LinuxEvent::CreateEvent(LinuxEvent::MANUAL_RESET, false);// CreateEvent(NULL, FALSE, FALSE, NULL);
  93. m_hDeviceCloseOK = LinuxEvent::CreateEvent(LinuxEvent::AUTO_RESET, false);
  94. m_hReOpen = LinuxEvent::CreateEvent(LinuxEvent::AUTO_RESET, false);
  95. m_pPacketArrivedCallbackFunc = nullptr;
  96. m_openCallbackFunc = nullptr;
  97. SetName(m_strClientName.c_str());
  98. }
  99. LogicClient::~LogicClient(void)
  100. {
  101. ////mLog::FWARN("Free LogicClient Object [{$}]", m_strClientName);
  102. cout << "Free LogicClient Object == " << m_strClientName << endl;
  103. if (IsClosed() == false) {
  104. Close();
  105. }
  106. if (m_Base_Thread != 0) {
  107. StopThread(3000); // 再次尝试停止线程
  108. }
  109. //delete m_pSysIF;
  110. CloseConnection(m_pMqttConn);
  111. //m_pSysIF = NULL;
  112. m_pMqttConn = nullptr;
  113. delete m_pActionName;
  114. m_pActionName = NULL;
  115. delete m_pFilePath;
  116. m_pFilePath = NULL;
  117. delete m_pLastMsgInfo;
  118. m_pLastMsgInfo = NULL;
  119. delete m_pPacketWaitQue;
  120. m_pPacketWaitQue = NULL;
  121. delete m_pPacketReceivedQue;
  122. m_pPacketReceivedQue = NULL;
  123. delete m_pPacketNotifyQue;
  124. m_pPacketNotifyQue = NULL;
  125. delete m_pFileHandle;
  126. m_pFileHandle = NULL;
  127. delete (CcosLock*)m_ClientLock;
  128. m_ClientLock = NULL;
  129. }
  130. bool LogicClient::GetFileHandle(CcosDevFileHandle &Handle)
  131. {
  132. if (IsClosed() == false)
  133. {
  134. Lock();
  135. Handle = (*m_pFileHandle);
  136. UnLock();
  137. return true;
  138. }
  139. //make client handle
  140. Handle[CCOS_PACK_HANDLE_LANG] = "en-US";//NOT FINISHED YET
  141. Handle[CCOS_PACK_HANDLE_ROUTE] = m_Router;//route
  142. Handle[CCOS_PACK_HANDLE_FLAGS] = 0;//
  143. Handle[CCOS_PACKET_HANDLE_KEY] = 0;//
  144. //target
  145. Handle.GetTarget(false)[CCOS_PACK_TARGET_ADDR] = (UINT64)0;
  146. Handle.GetTarget(false)[CCOS_PACK_TARGET_PROCID] = (UINT64)0;
  147. Handle.GetTarget(false)[CCOS_PACK_TARGET_BUSID] = "";
  148. Handle.GetTarget(false)[CCOS_PACK_TARGET_MACHINEID] = "";
  149. //local
  150. Handle.GetTarget()[CCOS_PACK_TARGET_ADDR] = (UINT64)(PVOID*)m_pMqttConn;
  151. Handle.GetTarget()[CCOS_PACK_TARGET_PROCID] = (UINT64)GetCurrentProcessId();
  152. Handle.GetTarget()[CCOS_PACK_TARGET_BUSID] = (const char*)getLocalEbusId();
  153. Handle.GetTarget()[CCOS_PACK_TARGET_MACHINEID] = (const char*)getLocalMachineId();
  154. return false;
  155. }
  156. bool LogicClient::SetRouter(CCOS_PACKET_ROUTE Route)
  157. {
  158. if (IsClosed())
  159. {
  160. m_Router = Route;
  161. return true;
  162. }
  163. return false;
  164. }
  165. bool LogicClient::Lock(DWORD timeout)
  166. {
  167. if (m_ClientLock == NULL)
  168. {
  169. ////mLog::FERROR("{$} @ Trans {$} Lock handle is null", m_strClientName, m_strCurTransaction);
  170. return false;
  171. }
  172. bool LockRet = (((CcosLock*)m_ClientLock)->Thread_Lock(timeout) == WAIT_OBJECT_0);
  173. std::cout << m_strClientName << " [LLKK] "<< (UINT64)this << " Try Lock Result " << LockRet << " With timeout " << timeout << endl;
  174. if (!LockRet)
  175. std::cout << m_strClientName << " [LLKK] " << (UINT64)this << " Lock Failed" << endl;
  176. return LockRet;
  177. }
  178. void LogicClient::UnLock()
  179. {
  180. if (m_ClientLock == NULL)
  181. {
  182. ////mLog::FERROR("{$} @ Trans {$} UnLock handle is null", m_strClientName, m_strCurTransaction);
  183. return ;
  184. }
  185. std::cout << m_strClientName << " [LLKK] " << (UINT64)this << "Try to Unlock " << endl;
  186. ((CcosLock*)m_ClientLock)->Thread_UnLock();
  187. std::cout << m_strClientName << " [LLKK] " << (UINT64)this << " Unlock " << endl;
  188. }
  189. bool LogicClient::InitClient(const char *pPath, int flags)
  190. {
  191. std::cout <<"Enter InitClient:pPath:"<< pPath << endl;
  192. ResDataObject rMsg;
  193. m_strDevicePath = pPath;
  194. m_OwnerThreadId = 0;
  195. (*m_pLastMsgInfo) = "";
  196. std::cout << "Start ClearQueue_Internal()" << endl;
  197. ClearQueue_Internal();
  198. ////mLog::FWARN("m_DeviceResource.clear() [{$}]", m_strDevicePath);
  199. m_DeviceResource.clear();
  200. //m_pFileHandle->m_Lang.SetVal("en-US");//NOT FINISHED YET
  201. //m_pFileHandle->m_Route = m_Router;//route
  202. //m_pFileHandle->m_Flags = flags;
  203. //m_pFileHandle->m_HandleId = 0;
  204. (*m_pFileHandle)[CCOS_PACK_HANDLE_LANG] = "en-US";//NOT FINISHED YET
  205. (*m_pFileHandle)[CCOS_PACK_HANDLE_ROUTE] = m_Router;//route
  206. (*m_pFileHandle)[CCOS_PACK_HANDLE_FLAGS] = flags;
  207. (*m_pFileHandle)[CCOS_PACKET_HANDLE_KEY] = (UINT64)0;//
  208. //target
  209. m_pFileHandle->GetTarget(false)[CCOS_PACK_TARGET_MACHINEID] = "";
  210. m_pFileHandle->GetTarget(false)[CCOS_PACK_TARGET_BUSID] = "";
  211. m_pFileHandle->GetTarget(false)[CCOS_PACK_TARGET_PROCID] = (UINT64)0;
  212. m_pFileHandle->GetTarget(false)[CCOS_PACK_TARGET_ADDR] = (UINT64)0;
  213. //m_pFileHandle->m_Dev.m_Addr = (UINT64)0;
  214. //m_pFileHandle->m_Dev.m_ProcID = (UINT64)0;
  215. //m_pFileHandle->m_Dev.m_busID = "";
  216. //m_pFileHandle->m_Dev.m_MachineID = "";
  217. //local
  218. m_pFileHandle->GetTarget()[CCOS_PACK_TARGET_MACHINEID] = (const char*)getLocalMachineId();
  219. m_pFileHandle->GetTarget()[CCOS_PACK_TARGET_BUSID] = (const char*)getLocalEbusId();
  220. m_pFileHandle->GetTarget()[CCOS_PACK_TARGET_PROCID] = (UINT64)GetCurrentProcessId();
  221. m_pFileHandle->GetTarget()[CCOS_PACK_TARGET_ADDR] = (UINT64)(PVOID*)m_pMqttConn;
  222. //m_pFileHandle->m_Owner.m_Addr = (UINT64)(PVOID*)m_pSysIF;
  223. //m_pFileHandle->m_Owner.m_ProcID = (UINT64)GetCurrentProcessId();
  224. //m_pFileHandle->m_Owner.m_busID = (const char*)getLocalEbusId();
  225. //m_pFileHandle->m_Owner.m_MachineID = (const char*)getLocalMachineId();
  226. m_FileFlags = flags;
  227. if (m_strDevicePath.length() <= 0)
  228. {
  229. //允许透传消息
  230. m_SyncMode = ACTION_SYNC;
  231. }
  232. if (pPath)
  233. {
  234. string busId;
  235. std::cout << "Start getBusIdFromFilePath()" << endl;
  236. if (getBusIdFromFilePath(pPath, busId) == false)
  237. {
  238. (*m_pFilePath) = "";
  239. //return false;
  240. }
  241. //m_pFileHandle->m_Dev.m_busID = busId;
  242. m_pFileHandle->GetTarget(false)[CCOS_PACK_TARGET_BUSID] = busId.c_str();
  243. (*m_pFilePath) = pPath;
  244. }
  245. //ClientSysIF *p = (ClientSysIF*)m_pSysIF;
  246. //p->Clear();
  247. //if (////mLog::gLogger == nullptr)
  248. //{
  249. // string strLogPath = GetProcessDirectory() + R"(\Conf\Log4CPP.Config.xml)";
  250. // LogHost = ((string)getLogRootpath()).c_str();
  251. // if (LogHost.length() <= 1)
  252. // {
  253. // char szName[256];
  254. // sprintf(szName, "/LogicClient_%08d", GetCurrentProcessId());
  255. // LogHost = szName;
  256. // }
  257. // Log4CPP::ThreadContext::Map::Set(ECOM::Utility::Hash("LogFileName"), "LogicClient");
  258. // //Log4CPP::GlobalContext::Map::Set("LogHost", LogHost.c_str());
  259. // Log4CPP::ThreadContext::Map::Set(ECOM::Utility::Hash("LogHost"), LogHost.c_str() + 1);
  260. // auto rc = Log4CPP::LogManager::LoadConfigFile(strLogPath.c_str());
  261. // ////mLog::gLogger = Log4CPP::LogManager::GetLogger("LogicClient");
  262. // ////mLog::FINFO("Code Build datetime [{$} {$}]", __DATE__, __TIME__);
  263. //}
  264. //else
  265. //{
  266. // string strRoot = ((string)getLogRootpath()).c_str();
  267. // if (strRoot.length() > 1 && strRoot != LogHost)
  268. // {
  269. // string strLogPath = GetProcessDirectory() + R"(\Conf\Log4CPP.Config.xml)";
  270. // LogHost = strRoot;
  271. // Log4CPP::ThreadContext::Map::Set(ECOM::Utility::Hash("LogFileName"), "LogicClient");
  272. // //Log4CPP::GlobalContext::Map::Set("LogHost", LogHost.c_str());
  273. // Log4CPP::ThreadContext::Map::Set(ECOM::Utility::Hash("LogHost"), LogHost.c_str() + 1);
  274. // auto rc = Log4CPP::LogManager::LoadConfigFile(strLogPath.c_str());
  275. // ////mLog::gLogger = Log4CPP::LogManager::GetLogger("LogicClient");
  276. // ////mLog::FINFO("Code Build datetime [{$} {$}]", __DATE__, __TIME__);
  277. // }
  278. //}
  279. //string version;
  280. //if (GetVersion(version, hMyModule))
  281. // ////mLog::FINFO("\n===============log begin : version:{$} ===================\n", version.c_str());
  282. //else
  283. ////mLog::FWARN("\n{$}=============== LogicClient log begin : version:1.0.0.0 ===================\n", m_strClientName);
  284. return true;
  285. }
  286. bool LogicClient::GetFilePath(ResDataObject &path)
  287. {
  288. if (IsClosed() == false)
  289. {
  290. path = (*m_pFilePath).c_str();
  291. return true;
  292. }
  293. return false;
  294. }
  295. RET_STATUS LogicClient::ProcessActionResponse(ResDataObject &packet, ResDataObject &OutputParam)
  296. {
  297. RET_STATUS ret = RET_FAILED;
  298. if (PacketAnalizer::GetPacketRetCode(&packet, ret))
  299. {
  300. //get error or warning Msg here
  301. string errInfo;
  302. PacketAnalizer::GetPacketMsgInfo(&packet, errInfo);
  303. if (errInfo.size() > 0)
  304. {
  305. string actionname = PacketAnalizer::GetPacketKey(&packet);
  306. (*m_pLastMsgInfo) = actionname + ":" + errInfo;
  307. }
  308. if (ret >= RET_SUCCEED)
  309. {
  310. if (PacketAnalizer::GetPacketContext(&packet, OutputParam))
  311. {
  312. //succeed
  313. //printf("Thread:%d,Action OK\n", GetCurrentThreadId());
  314. return ret;
  315. }
  316. }
  317. if (ret == RET_INVALID)
  318. {
  319. //DisconnectCDI();
  320. }
  321. }
  322. return ret;
  323. }
  324. RET_STATUS LogicClient::ProcessResource(ResDataObject &packet)
  325. {
  326. ResDataObject resParam;
  327. RET_STATUS ret = RET_FAILED;
  328. //got match response
  329. if (PacketAnalizer::GetPacketRetCode(&packet, ret))
  330. {
  331. //get error or warning Msg here
  332. PacketAnalizer::GetPacketMsgInfo(&packet, (*m_pLastMsgInfo));
  333. if (ret >= RET_SUCCEED)
  334. {
  335. //whole device resource
  336. if (PacketAnalizer::GetPacketContext(&packet, m_DeviceResource))
  337. {
  338. ////mLog::FDEBUG("Update Device {$} Resource [{$}]", m_strDevicePath, m_DeviceResource.encode());
  339. return ret;
  340. }
  341. }
  342. }
  343. ////mLog::FERROR("Device Open Result Error", packet.encode());
  344. //put some log here
  345. ////mLog::FINFO("m_DeviceResource.clear() [{$}] after Device Open Result Error", m_strDevicePath);
  346. m_DeviceResource.clear();
  347. return RET_FAILED;
  348. }
  349. RET_STATUS LogicClient::ProcessClose(ResDataObject &packet)
  350. {
  351. //we have disconnect here
  352. PacketAnalizer::GetPacketMsgInfo(&packet, (*m_pLastMsgInfo));
  353. //make the object status as CLOSED
  354. //DisconnectCDI();
  355. //clear the info
  356. //m_pPacketReceivedQue->Lock();
  357. m_pPacketWaitQue->Clear();
  358. m_pPacketReceivedQue->Clear();
  359. //m_DeviceResource.clear();
  360. //m_pPacketReceivedQue->UnLock();
  361. //m_pPacketNotifyQue->Lock();
  362. m_pPacketNotifyQue->Clear();
  363. //m_pPacketNotifyQue->UnLock();
  364. return RET_INVALID;
  365. }
  366. RET_STATUS LogicClient::ProcessOpenResponse(ResDataObject &packet)
  367. {
  368. ResDataObject resParam;
  369. RET_STATUS ret = RET_FAILED;
  370. //got match response
  371. if (PacketAnalizer::GetPacketRetCode(&packet, ret))
  372. {
  373. //get error or warning Msg here
  374. PacketAnalizer::GetPacketMsgInfo(&packet, (*m_pLastMsgInfo));
  375. if (ret >= RET_SUCCEED)
  376. {
  377. if (PacketAnalizer::GetPacketHandle(&packet, resParam))
  378. {
  379. if (m_pFileHandle == NULL)
  380. return RET_FAILED;
  381. //got file handle
  382. if (m_pFileHandle->SetResDataObject(resParam))
  383. {
  384. //succeed here
  385. //Get Device Resource
  386. if (ProcessResource(packet) >= RET_SUCCEED)
  387. {
  388. string res = packet.encode();
  389. ////mLog::FDEBUG("OpenResponse ok {$}", res.substr(0, 16384));
  390. //////mLog::FDEBUG("OpenResponse ok\n");
  391. if (m_bNeedNotify && m_ClosedFlag) {
  392. const char* realPath = NULL;
  393. if (m_strDevicePath[0] == '/')
  394. realPath = ((char*)m_strDevicePath.c_str()) + 1; //去掉首字符"/"
  395. else
  396. realPath = m_strDevicePath.c_str();
  397. string notifyTopic = realPath;
  398. notifyTopic += "/Notify/#";
  399. SubscribeTopic(m_pMqttConn, notifyTopic.c_str());
  400. }
  401. if (m_openCallbackFunc != nullptr)
  402. m_openCallbackFunc(this, m_strDevicePath.c_str(), 1);
  403. m_OwnerThreadId = (DWORD)GetCurrentThreadId();
  404. std::cout << "TID [" << m_OwnerThreadId << "] LogicClient::ProcessOpenResponse OpenResponse ok." << m_strClientName << endl;
  405. ////mLog::FDEBUG("LogicClient::ProcessOpenResponse {$} OpenResponse ok. {$}", m_OwnerThreadId, m_strClientName);
  406. m_hDeviceOpenOK->SetEvent();
  407. SetThreadOnTheRun(true); //线程可以正常运转
  408. m_ClosedFlag = false;
  409. return ret;
  410. }
  411. }
  412. }
  413. }
  414. }
  415. ////mLog::FERROR("Thread:{$} ,OpenResponse ng ",GetCurrentThreadId());
  416. //printf("Thread:%d,OpenResponse ng\n", GetCurrentThreadId());
  417. //put some log here
  418. return RET_FAILED;
  419. }
  420. RET_STATUS LogicClient::CloseDevice()
  421. {
  422. RET_STATUS ret = RET_FAILED;
  423. const char* realPath = NULL;
  424. if (m_strDevicePath[0] == '/')
  425. realPath = ((char*)m_strDevicePath.c_str()) + 1; //去掉首字符"/"
  426. else
  427. realPath = m_strDevicePath.c_str();
  428. ////mLog::FDEBUG("{$} CloseDevice {$} and use default topic {$}", m_strClientName, realPath, m_strDefaultTopic);
  429. if (Lock(1000) == false)
  430. {
  431. ////mLog::FERROR("CloseDev Lock Timeout for Dev: {$} ", m_strClientName);
  432. //GPRINTA_ERROR("OpenDev Lock Timeout for Dev:%s", pPath);
  433. return RET_TIMEOUT;
  434. }
  435. if (m_ClosedFlag)
  436. {
  437. UnSubscribe(m_pMqttConn, m_strDefaultTopic.c_str());
  438. if (m_bNeedNotify) {
  439. string notifyTopic = realPath;
  440. notifyTopic += "/Notify/#";
  441. UnSubscribe(m_pMqttConn, notifyTopic.c_str());
  442. }
  443. }
  444. UnLock();
  445. m_hDeviceOpenOK->ResetEvent();
  446. m_hDeviceCloseOK->SetEvent();
  447. //标志
  448. //m_ClosedFlag = true;
  449. ////make close req
  450. ResDataObject req;
  451. PacketAnalizer::MakeCloseRequest(req, (*m_pFileHandle));
  452. if ((m_FileFlags & CCOS_FILE_FLAGS::ABSCRACT_ONLINE) == CCOS_FILE_FLAGS::ABSCRACT_ONLINE)
  453. {
  454. if (m_strWS.length() > 0)
  455. {
  456. //需要显式 上设备上线
  457. ResDataObject resContext;
  458. resContext.add("Offline", m_strWS.c_str());
  459. PacketAnalizer::UpdatePacketContext(req, resContext);
  460. PublishAction(&req, m_pFilePath->c_str(), m_pMqttConn);
  461. }
  462. }
  463. ////send it
  464. ////ret = pLogicDeivce->CmdFromLogicDev(&req);
  465. ////UnLock();
  466. //DWORD dwret = WaitForSingleObject(m_hCloseOK, 1000);
  467. //if (dwret != WAIT_OBJECT_0)
  468. //{
  469. // //printf("Thread:%d,Send Close OK\n", GetCurrentThreadId());
  470. //}
  471. //else
  472. //{
  473. // printf("Thread:%d,Send Close Failed\n", GetCurrentThreadId());
  474. //}
  475. return ret;
  476. }
  477. RET_STATUS LogicClient::OpenDevice(bool bAsync)
  478. {
  479. std::cout << "[LogicClient::OpenDevice-Info] IN" << std::endl;
  480. if (m_strDevicePath.empty())
  481. {
  482. m_OwnerThreadId = (DWORD)GetCurrentThreadId();
  483. std::cout << "TID [" << m_OwnerThreadId << "] LogicClient::ProcessOpenResponse OpenResponse ok." << m_strClientName << std::endl;
  484. m_hDeviceOpenOK->SetEvent();
  485. SetThreadOnTheRun(true);
  486. m_ClosedFlag = false;
  487. return RET_SUCCEED;
  488. }
  489. const char* realPath = (m_strDevicePath[0] == '/') ? (m_strDevicePath.c_str() + 1) : m_strDevicePath.c_str();
  490. if (!Lock(1000))
  491. {
  492. return RET_TIMEOUT;
  493. }
  494. if (m_ClosedFlag)
  495. {
  496. std::cout << "[LogicClient::OpenDevice-Info] " << m_strClientName << " - Subscribing to default MQTT topic: " << m_strDefaultTopic << std::endl;
  497. SubscribeTopic(m_pMqttConn, m_strDefaultTopic.c_str());
  498. // 等待订阅完成
  499. const int delayMs = 100;
  500. std::cout << "[LogicClient::OpenDevice-Info] " << m_strClientName << " - Waiting " << delayMs << "ms to ensure MQTT subscription completes" << std::endl;
  501. usleep(delayMs * 1000);
  502. }
  503. // 输出调试信息
  504. std::cout << "[LogicClient::OpenDevice-Debug] " << m_strClientName << " - OpenDevice (Path: " << realPath
  505. << ", Default Topic: " << m_strDefaultTopic
  506. << ", Async Mode: " << (bAsync ? "Enabled" : "Disabled")
  507. << (m_ClosedFlag ? "" : ", MQTT Topic: No Re-Subscription Required") << std::endl;
  508. // 构建并发送请求
  509. ResDataObject req;
  510. DWORD reqidx = PacketAnalizer::MakeOpenRequest(req, (*m_pFileHandle), realPath);
  511. PacketAnalizer::UpdateOpenRequest(req, getLocalMachineId(), getLocalEbusId(), GetCurrentProcessId(), (UINT64)m_pMqttConn);
  512. PacketAnalizer::UpdateContextTopic(&req, m_strDefaultTopic.c_str());
  513. PacketAnalizer::UpdatePacketTopic(&req, realPath, m_strClientName.c_str());
  514. // 处理在线标志
  515. if ((m_FileFlags & CCOS_FILE_FLAGS::ABSCRACT_ONLINE) == CCOS_FILE_FLAGS::ABSCRACT_ONLINE && !m_strWS.empty())
  516. {
  517. ResDataObject resContext;
  518. PacketAnalizer::GetPacketContext(&req, resContext);
  519. resContext.add("Online", m_strWS.c_str());
  520. PacketAnalizer::UpdatePacketContext(req, resContext);
  521. }
  522. // 发布请求
  523. int publishResult = PublishAction(&req, realPath, m_pMqttConn);
  524. UnLock();
  525. if (publishResult < 0)
  526. {
  527. return RET_FAILED;
  528. }
  529. if (!bAsync)
  530. {
  531. auto waitAndCheck = [&]() -> bool {
  532. std::cout << "try Wait for Open ok." << m_strClientName << std::endl;
  533. if (m_hDeviceOpenOK->Wait(4000))
  534. {
  535. std::cout << "Open ok. done" << m_strClientName << std::endl;
  536. if (m_openCallbackFunc) m_openCallbackFunc(this, realPath, 1);
  537. return true;
  538. }
  539. return false;
  540. };
  541. if (waitAndCheck())
  542. {
  543. return RET_SUCCEED;
  544. }
  545. }
  546. else
  547. {
  548. return RET_FAILED;
  549. }
  550. return RET_SUCCEED;
  551. }
  552. bool LogicClient::OnStartThread()
  553. {
  554. RET_STATUS ret = RET_FAILED;
  555. const char* realPath = NULL;
  556. std::cout << "OnStartThreadL:m_strDevicePath: " << m_strDevicePath << std::endl;
  557. if (m_strDevicePath[0] == '/')
  558. realPath = ((char*)m_strDevicePath.c_str()) + 1; //去掉首字符"/"
  559. else
  560. realPath = m_strDevicePath.c_str();
  561. //if (Lock(10000) == false)
  562. //{
  563. // //GPRINTA_ERROR("OpenDev Lock Timeout for Dev[flag:%d]:%s", flags, pPath);
  564. // return false;
  565. //}
  566. //GPRINTA_DEBUG("Thread:%d,Open Entry", GetCurrentThreadId());
  567. //make connection to CDI
  568. //ConnectCDI();
  569. ResDataObject req, res, resParam;
  570. //make open req
  571. //HANDLE hEvent = CreateEvent(NULL, FALSE, FALSE, NULL);
  572. // 新增调试输出
  573. if (m_pMqttConn) {
  574. std::cout << "m_pMqttConn Memory address: " << m_pMqttConn << std::endl;
  575. // 解包元组内容并输出
  576. void* mqttClient = std::get<0>(*m_pMqttConn);
  577. mqtt_topic_list* topicList = std::get<1>(*m_pMqttConn);
  578. void* msgHook = std::get<2>(*m_pMqttConn);
  579. std::string clientId = std::get<3>(*m_pMqttConn);
  580. ccos_mqtt_callback callback = std::get<4>(*m_pMqttConn);
  581. pthread_t threadId = std::get<5>(*m_pMqttConn);
  582. CcosLock* connLock = std::get<6>(*m_pMqttConn);
  583. mqtt_msg_list* msgList = std::get<7>(*m_pMqttConn);
  584. sem_t* semaphore = std::get<8>(*m_pMqttConn);
  585. std::cout << "MQTT Client Handle: " << mqttClient << std::endl;
  586. std::cout << "MQTT Topic List: " << topicList << std::endl;
  587. std::cout << "Message Hook ID: " << msgHook << std::endl;
  588. std::cout << "Client ID: " << (!clientId.empty() ? clientId : "null") << std::endl;
  589. std::cout << "Thread ID: " << threadId << std::endl;
  590. std::cout << "Connection Lock: " << connLock << std::endl;
  591. std::cout << "Message List: " << msgList << std::endl;
  592. std::cout << "Semaphore Handle: " << semaphore << std::endl;
  593. }
  594. else {
  595. std::cout << "m_pMqttConn is nullptr" << std::endl;
  596. }
  597. //HANDLE hOpened = CreateEvent(NULL, FALSE, FALSE, NULL);
  598. if (m_pMqttConn == nullptr)
  599. {
  600. ////mLog::FINFO("LogicClient::Open {$} try open path:[{$}]", m_strClientName, realPath );
  601. std::cout << CurrentDateTime() << " LogicClient::Open [" << m_strClientName << "] try open path: " << realPath << endl << endl;
  602. m_pMqttConn = NewConnection(m_strClientName.c_str(), [&ret, realPath, this](ResDataObject* rsp, const char* topic, void* conn_void) {
  603. //m_pMqttConn = CreateConnection
  604. std::cout << "OnStartThread::NewConnnect callback in "<< std::endl;
  605. if (m_pMqttConn == NULL)
  606. return;
  607. //回调函数不再加锁,要确保每一步都是ok的
  608. /*
  609. if (Lock(1000) == false)
  610. {
  611. ////mLog::FERROR("callback Lock Timeout for Dev: {$} ", m_strClientName);
  612. //GPRINTA_ERROR("OpenDev Lock Timeout for Dev:%s", pPath);
  613. return ;
  614. }*/
  615. ccos_mqtt_connection* conn = (ccos_mqtt_connection*)conn_void;
  616. if (PacketAnalizer::GetPacketType(rsp) == PACKET_TYPE_RES)
  617. {
  618. //check cmd
  619. string path = realPath;
  620. if ((PacketAnalizer::GetPacketCmd(rsp) == PACKET_CMD_OPEN) &&
  621. (PacketAnalizer::GetPacketKey(rsp).substr(0, path.length()) == path || PacketAnalizer::GetPacketKey(rsp) == realPath))
  622. {
  623. ////mLog::FINFO("{$} LogicClient::Open {$} OK ", m_strClientName, realPath);
  624. std::cout << "\n\n-----------" << CurrentDateTime() << "LogicClient::Open OK [" << m_strClientName << "] try open path:" << realPath << endl << endl;
  625. if (Lock(2000))
  626. {
  627. ret = ProcessOpenResponse(*rsp);
  628. UnLock();
  629. }
  630. else
  631. {
  632. ////mLog::FERROR("{$} Process Open Result with Lock Client failed ", m_strClientName);
  633. }
  634. }
  635. else
  636. {
  637. ////mLog::FINFO("Unkown Packet: cmd {$} packet key: {$}", (int)PacketAnalizer::GetPacketCmd(rsp), PacketAnalizer::GetPacketKey(rsp) );
  638. std::cout << m_strClientName << CurrentDateTime() << "Try Conn callback get unkown packet pointer [" << (UINT64)this << endl;
  639. std::cout << "Unkown Packet: cmd [" << PacketAnalizer::GetPacketCmd(rsp) << "] packet key: " << PacketAnalizer::GetPacketKey(rsp) << "" << endl;
  640. if (m_pPacketArrivedCallbackFunc != nullptr)
  641. {
  642. m_pPacketArrivedCallbackFunc(rsp, topic, conn);
  643. }
  644. else
  645. {
  646. //m_pPacketReceivedQue->Lock();
  647. m_pPacketReceivedQue->InQueue(*rsp);
  648. //SetEvent(m_NotifyEvent);//notify to user
  649. m_ResponseEvent->SetEvent();
  650. //m_pPacketReceivedQue->UnLock();
  651. }
  652. }
  653. }
  654. else
  655. {
  656. //got other one
  657. //printf("Thread:%d,in open.got other packet\n", GetCurrentThreadId());
  658. //GPRINTA_DEBUG("in open.got other packet\n\n");
  659. //check priority one => CLOSE
  660. if (PacketAnalizer::GetPacketCmd((ResDataObject*)rsp) == PACKET_CMD_CLOSE)
  661. {
  662. std::cout << CurrentDateTime() << m_strClientName << "Got Close Paket ..." << endl;
  663. //we have disconnect here
  664. //GPRINTA_DEBUG("in open.got Close packet\n\n");
  665. ret = ProcessClose(*rsp);
  666. }
  667. else {
  668. std::cout << "*****-----***** " << CurrentDateTime() << m_strClientName << " Received msg " << rsp->encode() << endl;
  669. if (m_pPacketArrivedCallbackFunc != nullptr)
  670. m_pPacketArrivedCallbackFunc(rsp, topic, conn);
  671. else
  672. {
  673. if (PacketAnalizer::GetPacketType(rsp) == PACKET_TYPE_NOTIFY )
  674. {
  675. PACKET_CMD cmd = PACKET_CMD_NONE;
  676. PACKET_TYPE rType;
  677. string key;
  678. rType = PacketAnalizer::GetPacketType(rsp);
  679. cmd = PacketAnalizer::GetPacketCmd(rsp);
  680. if (rType == PACKET_TYPE_NOTIFY /*&& m_pActionName->size() <= 0*/ && cmd == PACKET_CMD_UPDATE)
  681. {
  682. //只有在非Action期间,更新,如果有遗漏,再说 2024.9.2
  683. cmd = PacketAnalizer::GetPacketCmd(rsp);
  684. key = PacketAnalizer::GetPacketKey(rsp);
  685. //2.notify
  686. UpdateDeviceResource(cmd, *rsp);
  687. }
  688. else
  689. {
  690. ////mLog::FWARN("{$} Got unexpected.. Packet {$}", m_strClientName, rsp->encode());
  691. }
  692. //m_pPacketNotifyQue->Lock();
  693. m_pPacketNotifyQue->InQueue(*rsp);
  694. //SetEvent(m_NotifyEvent);//notify to user
  695. //m_pPacketNotifyQue->UnLock();
  696. m_NotifyEvent->SetEvent();
  697. }
  698. else
  699. PacketArrived(rsp);
  700. }
  701. }
  702. }
  703. //UnLock();
  704. });
  705. }
  706. if (m_pMqttConn == nullptr) {
  707. ////mLog::FERROR("LogicClient {$} Connect Mqtt client failed with open device : {$}", m_strClientName, realPath);
  708. std::cout << "XXXXXX XXXXXX [" << m_strClientName << "] LogicClient Connect Mqtt client failed with open device [" << realPath << "]" << endl;
  709. //UnLock();
  710. if (m_openCallbackFunc != nullptr)
  711. m_openCallbackFunc(this,realPath, 0);
  712. return false;
  713. }
  714. //ret = RET_SUCCEED;
  715. //SubscribeTopic(m_pMqttConn, m_strDefaultTopic.c_str());
  716. //PacketAnalizer::UpdateOpenRequest(req, getLocalMachineId(), getLocalEbusId(), GetCurrentProcessId(), (UINT64)m_pMqttConn);
  717. //PacketAnalizer::UpdateContextTopic(&req, m_strDefaultTopic.c_str());
  718. //PacketAnalizer::UpdatePacketTopic(&req, realPath, m_strClientName.c_str());
  719. //PublishAction(&req, realPath, m_pMqttConn);
  720. ////UnLock();
  721. //DWORD dwret = WaitForSingleObject(m_hOpenOK, 4000);
  722. //if (dwret == WAIT_OBJECT_0)
  723. //{
  724. // if (m_openCallbackFunc != nullptr)
  725. // m_openCallbackFunc(this, realPath, 1);
  726. // return true;
  727. //}
  728. if (OpenDevice() == RET_SUCCEED)
  729. return true;
  730. ////mLog::FINFO("LogicClient {$} Open Device {$} failed ", m_strClientName, realPath);
  731. if (IsClosed()) {
  732. std::cout << CurrentDateTime()<< " XXXXXX XXXXXX [Thread Exiting] LogicClient Open Device failed" << endl;
  733. }
  734. else {
  735. std::cout << "XXXXXX XXXXXX [" << m_strClientName << "]LogicClient Open Device [" << realPath << "] failed " << endl;
  736. }
  737. if (m_openCallbackFunc != nullptr)
  738. m_openCallbackFunc(this, realPath, 0);
  739. CloseConnection(m_pMqttConn);
  740. m_pMqttConn = nullptr;
  741. return false;
  742. }
  743. bool LogicClient::OnEndThread()
  744. {
  745. std::cout << "Thread End..." << endl;
  746. ////mLog::FINFO("End Client Thread");
  747. //CloseDevice();
  748. //ResDataObject req;
  749. ////make close req
  750. //PacketAnalizer::MakeCloseRequest(req, (*m_pFileHandle));
  751. ////send it
  752. ////ret = pLogicDeivce->CmdFromLogicDev(&req);
  753. //PublishAction(&req, m_pFilePath->c_str(), m_pMqttConn);
  754. ////UnLock();
  755. //DWORD dwret = WaitForSingleObject(m_hCloseOK, 1000);
  756. //if (dwret != WAIT_OBJECT_0)
  757. //{
  758. // //printf("Thread:%d,Send Close OK\n", GetCurrentThreadId());
  759. //
  760. //}
  761. //else
  762. //{
  763. // printf("Thread:%d,Send Close Failed\n", GetCurrentThreadId());
  764. //}
  765. //std::cout << CurrentDateTime() << "[" << m_strClientName << "] try Close Mqtt " << endl;
  766. //CloseConnection(m_pMqttConn);
  767. m_pMqttConn = nullptr;
  768. return true;
  769. }
  770. bool LogicClient::Exec(void)
  771. {
  772. //ResDataObject req;
  773. //if (ReadCmd(req))
  774. //{
  775. // return true;
  776. //}
  777. std::vector<std::shared_ptr<LinuxEvent>> hWait = {
  778. m_ExitFlag,
  779. m_hReOpen,
  780. };
  781. DWORD dwRet = LinuxEvent::WaitForMultipleEvents(hWait, 10);//WaitForSingleObject(m_ExitFlag, 100);
  782. if (dwRet == WAIT_OBJECT_0)
  783. {
  784. std::cout << m_strClientName << " MQTT Prepare Exit " << endl;
  785. return false;
  786. }
  787. else if (dwRet == WAIT_OBJECT_0 + 1)
  788. {
  789. ////mLog::FDEBUG(" {$} try Real ReOpen", m_strClientName);
  790. //CloseDevice();
  791. m_hDeviceOpenOK->ResetEvent();
  792. OpenDevice(false);
  793. ////mLog::FDEBUG(" {$} try ReOpen over", m_strClientName);
  794. }
  795. usleep(1000);
  796. return true;
  797. }
  798. int LogicClient::ReOpenDevice()
  799. {
  800. ////mLog::FINFO(" {$} set to ReOpen",m_strClientName );
  801. m_hDeviceOpenOK->ResetEvent();
  802. m_hReOpen->SetEvent();
  803. return RET_SUCCEED;
  804. }
  805. int LogicClient::Open(const char *pPath, int flags, const char* pszWS, DWORD timeout, logic_client_open_callback calback)
  806. {
  807. std::cout << CurrentDateTime() << " LogicClient::Open[" << m_strClientName << "] try Open " << pPath << endl;
  808. if (Lock(timeout) == false)
  809. {
  810. std::cout << "FERROR::OpenDev Lock Timeout for Dev[flag: " << flags << " ]: " << pPath << endl;
  811. ////mLog::FERROR("OpenDev Lock Timeout for Dev[flag: {$} ]: {$} ", flags, pPath);
  812. //GPRINTA_ERROR("OpenDev Lock Timeout for Dev[flag:%d]:%s", flags, pPath);
  813. return RET_TIMEOUT;
  814. }
  815. m_strWS = pszWS;
  816. if (InitClient(pPath, flags) == false)
  817. {
  818. UnLock();
  819. std::cout << "FERROR::InitClient Failed for Dev [flag: " << flags << " ]: " << pPath << endl;
  820. ////mLog::FERROR("InitClient Failed for Dev [flag: {$} ]: {$} ", flags, pPath);
  821. //GPRINTA_ERROR("InitClient Failed for Dev[flag:%d]:%s", flags, pPath);
  822. return RET_FAILED;
  823. }
  824. std::cout <<"IsClosed" << endl;
  825. if (IsClosed() == false)
  826. {
  827. std::cout << "LogicClient::Open " << m_strClientName << " try is Already Opend: " << m_strDevicePath << endl;
  828. ////mLog::FINFO("LogicClient::Open {$} try is Already Opend: ", m_strClientName, m_strDevicePath);
  829. std::cout << CurrentDateTime() << " LogicClient::Open [" << m_strClientName << "] try is Already Opend: " << m_strDevicePath << endl;
  830. if (Close() == RET_PENDING)
  831. {
  832. UnLock();
  833. std::cout << "OpenDev Close Method is onGoing.. for Dev [flag: " << flags << " ]: " << pPath << endl;
  834. ////mLog::FERROR("OpenDev Close Method is onGoing.. for Dev [flag: {$} ]: {$} ", flags, pPath);
  835. //GPRINTA_ERROR("OpenDev Close Method is onGoing.. for Dev[flag:%d]:%s", flags, pPath);
  836. return RET_PENDING;
  837. }
  838. }
  839. m_openCallbackFunc = calback;
  840. std::cout << "UnLock" << endl;
  841. UnLock();
  842. std::cout << "StartThread" << endl;
  843. StartThread(false, true);
  844. if (calback == nullptr)
  845. {
  846. if (WaitTheThreadOnTheRun(timeout))
  847. return RET_SUCCEED;
  848. std::cout << "WaitTheThreadOnTheRun Timeout " << timeout << " for Open " << pPath << endl;
  849. ////mLog::FERROR("WaitTheThreadOnTheRun Timeout {$} for Open {$}", timeout, pPath);
  850. //GPRINTA_ERROR("WaitTheThreadOnTheRun Timeout %i for Open %s", timeout, pPath);
  851. return RET_FAILED;
  852. }
  853. return RET_SUCCEED;
  854. }
  855. int LogicClient::Close()
  856. {
  857. RET_STATUS ret = RET_SUCCEED;
  858. ////mLog::FINFO(" LogicClient::Close {$} try Close path:",m_strClientName , m_strDevicePath);
  859. std::cout << CurrentDateTime() << " LogicClient::Close [" << m_strClientName << "] try Close path: " << m_strDevicePath << endl;
  860. if (Lock(1000) == false)
  861. {
  862. ////mLog::FINFO("close wait timeout");
  863. std::cout << "close wait timeout " << endl;
  864. //return RET_PENDING;
  865. }
  866. if (IsClosed() == false)
  867. {
  868. StopThread(3000);
  869. CloseConnection(m_pMqttConn);
  870. }
  871. //printf("Thread:%d,Close Entry\n", GetCurrentThreadId());
  872. UnLock();
  873. ////mLog::FINFO(" LogicClient::Close {$} CloseConnection : {$}", m_strClientName, m_strDevicePath);
  874. std::cout << CurrentDateTime() << " LogicClient::Close [" << m_strClientName << "] CloseConnection : " << m_strDevicePath << endl;
  875. //Disconnect
  876. //DisconnectCDI();
  877. //底层MQTT连接并不关闭,除非显式关闭,或者对象被释放
  878. m_hDeviceOpenOK->ResetEvent();
  879. m_pMqttConn = nullptr;
  880. m_ClosedFlag = true;
  881. ////mLog::FINFO(" LogicClient::Close {$} Close Over.:", m_strClientName, m_strDevicePath);
  882. std::cout << CurrentDateTime() << " LogicClient::Close [" << m_strClientName << "] Close Over.: " << m_strDevicePath << endl;
  883. return ret;
  884. }
  885. const char *LogicClient::GetLastErrorInfo()
  886. {
  887. return m_pLastMsgInfo->c_str();
  888. }
  889. bool LogicClient::IsClosed()
  890. {
  891. return m_ClosedFlag;
  892. }
  893. //bool LogicClient::IsOpenOK()
  894. //{
  895. // if (m_ClosedFlag)
  896. // return false;
  897. // DWORD dwRet = WaitForSingleObject(m_hDeviceOpenOK, 0);
  898. // if (dwRet == WAIT_OBJECT_0)
  899. // {
  900. // return true;
  901. // }
  902. // else
  903. // {
  904. // return false;
  905. // }
  906. //}
  907. /*
  908. bool LogicClient::IsAllReqFinished()
  909. {
  910. ClientSysIF *p = (ClientSysIF*)m_pSysIF;
  911. bool ret = false;
  912. ret = p->IsAllRequestFinished();
  913. return (ret);
  914. }
  915. bool LogicClient::SetRequestSyncMode(ACTION_SYNC_MODE Sync)
  916. {
  917. if (IsAllReqFinished())
  918. {
  919. m_SyncMode = Sync;
  920. return true;
  921. }
  922. return false;
  923. }
  924. ACTION_SYNC_MODE LogicClient::GetRequestSyncMode()
  925. {
  926. return m_SyncMode;
  927. }*/
  928. RET_STATUS LogicClient::UpdateDeviceResource(DWORD timeout)
  929. {
  930. RET_STATUS ret = RET_FAILED;
  931. Lock();
  932. ResDataObject req, res;
  933. try {
  934. ret = (RET_STATUS)Action("UpdateDeviceResource", req, res, timeout);
  935. if (ret >= RET_SUCCEED)
  936. {
  937. ////mLog::FINFO("m_DeviceResource.clear() [{$}]", m_strDevicePath);
  938. //update
  939. m_DeviceResource.clear();
  940. m_DeviceResource = res;
  941. ////mLog::FINFO("m_DeviceResource clear [{$}] .set [{$}]", m_strDevicePath, res.encode());
  942. }
  943. }
  944. catch (...)
  945. {
  946. ret = RET_FAILED;
  947. }
  948. UnLock();
  949. return ret;
  950. }
  951. bool LogicClient::HasAction(const char* pszAction)
  952. {
  953. if (string(pszAction) == "UpdateDeviceResource")
  954. return true;
  955. int actid = m_DeviceResource.GetFirstOf("Action");
  956. if (actid < 0)
  957. {
  958. ////mLog::FINFO("Device [{$}] has no Action Items for Action [{$}] in [{$}]", m_strDevicePath, pszAction, m_DeviceResource.encode());
  959. return false;
  960. }
  961. ResDataObject actions = m_DeviceResource[actid];
  962. if (actions.GetFirstOf(pszAction) >= 0)
  963. return true;
  964. ////mLog::FINFO("Device [{$}] Action no Items for Action [{$}] in [{$}]", m_strDevicePath, pszAction, actions.encode());
  965. return false;
  966. }
  967. bool LogicClient::HasProperty(const char* pszProperty)
  968. {
  969. int atrid = m_DeviceResource.GetFirstOf("Attribute");
  970. if (atrid < 0)
  971. return false;
  972. ResDataObject actions = m_DeviceResource[atrid];
  973. if (actions.GetFirstOf(pszProperty) >= 0)
  974. return true;
  975. return false;
  976. }
  977. bool LogicClient::GetProperty(const char* pszProperty, ResDataObject& resRet)
  978. {
  979. int atrid = m_DeviceResource.GetFirstOf("Attribute");
  980. if (atrid < 0)
  981. return false;
  982. ResDataObject propertys = m_DeviceResource[atrid];
  983. if (propertys.GetFirstOf(pszProperty) >= 0)
  984. {
  985. resRet = propertys[pszProperty];
  986. return true;
  987. }
  988. return false;
  989. }
  990. RET_STATUS LogicClient::GetDeviceType(GUID &DevType)
  991. {
  992. RET_STATUS ret = RET_FAILED;
  993. Lock();
  994. try {
  995. if (IsClosed() == false)
  996. {
  997. if (string_2_guid((const char *)m_DeviceResource["DeviceType"], DevType))
  998. {
  999. ret = RET_SUCCEED;
  1000. }
  1001. }
  1002. }
  1003. catch (...)
  1004. {
  1005. ret = RET_FAILED;
  1006. }
  1007. UnLock();
  1008. return ret;
  1009. }
  1010. RET_STATUS LogicClient::GetDeviceResource(ResDataObject PARAM_OUT *pDeviceResource)
  1011. {
  1012. RET_STATUS ret = RET_FAILED;
  1013. bool blocked = Lock();
  1014. try {
  1015. if (IsClosed() == false)
  1016. {
  1017. (*pDeviceResource) = m_DeviceResource;
  1018. string res = m_DeviceResource.encode();
  1019. ////mLog::FINFO("GetDeviceResource ok. {$}", res.substr(0, 16384));
  1020. ret = RET_SUCCEED;
  1021. UnLock();
  1022. return ret;
  1023. }
  1024. }
  1025. catch (...)
  1026. {
  1027. ret = RET_FAILED;
  1028. }
  1029. UnLock();
  1030. ////mLog::FINFO("{$} GetDeviceResource failed with closedflag [{$}]",m_strClientName, m_ClosedFlag?1:0);
  1031. return ret;
  1032. }
  1033. void LogicClient::PacketArrived(ResDataObject *pResponse)
  1034. {
  1035. //two kind packet.
  1036. //1.response
  1037. //1.1 open response
  1038. //1.2 close response
  1039. //no Locks!!!!!!
  1040. //assert(0);
  1041. //check priority one => CLOSE
  1042. //if (PacketAnalizer::GetPacketCmd(pResponse) == PACKET_CMD_CLOSE)
  1043. //{
  1044. // //we have disconnect here
  1045. // printf("Thread:%d,Got Close Notify.Passive Close\n", GetCurrentThreadId());
  1046. // ProcessClose(*pResponse);
  1047. // SetEvent(m_NotifyEvent);//notify to user
  1048. // SetEvent(m_ResponseEvent);//notify to user
  1049. // return;
  1050. //}
  1051. //1.3 update response
  1052. //1.4 action response
  1053. //1.5 data ??
  1054. //2.notify
  1055. //2.1 data notify
  1056. //m_pPacketReceivedQue->Lock();
  1057. m_pPacketReceivedQue->InQueue(*pResponse);
  1058. //for test
  1059. //GPRINTA_DEBUG("Client:%d,Thread %d:Got Packet--JG:%s",this, GetCurrentThreadId(), pResponse->encode());
  1060. if (m_ReqIdx != 0)
  1061. {
  1062. if (m_ReqIdx == PacketAnalizer::GetPacketIdx(pResponse))
  1063. {
  1064. if (PacketAnalizer::GetPacketType(pResponse) == PACKET_TYPE_RES)
  1065. {
  1066. //check cmd
  1067. if ((PacketAnalizer::GetPacketCmd(pResponse) == PACKET_CMD_EXE) &&
  1068. (PacketAnalizer::GetPacketKey(pResponse) == (*m_pActionName)))
  1069. {
  1070. //got match response
  1071. //for test
  1072. //GPRINTA_DEBUG("Client:%d,Got Response--JG",this);
  1073. m_ResponseEvent->SetEvent();
  1074. //m_pPacketReceivedQue->UnLock();
  1075. return;
  1076. }
  1077. }
  1078. ////mLog::FERROR(" {$} Action Resp Error {$}", m_strClientName, pResponse->encode());
  1079. std::cout << "[" << m_strClientName << "] Action Resp Error " << pResponse->encode() << endl;
  1080. }
  1081. else
  1082. {
  1083. ////mLog::FDEBUG(" {$} When Actioned.. Get Notify {$} ", m_strClientName, pResponse->encode());
  1084. std::cout << "[" << m_strClientName << "] When Actioned.. Get Notify " << pResponse->encode() << endl;
  1085. }
  1086. }
  1087. else {
  1088. ////mLog::FDEBUG(" {$} Get Notify {$}", m_strClientName, pResponse->encode());
  1089. std::cout << "[" << m_strClientName << "] Get Notify " << pResponse->encode() << endl;
  1090. }
  1091. m_ResponseEvent->SetEvent();//notify to user
  1092. //m_pPacketReceivedQue->UnLock();
  1093. }
  1094. int LogicClient::Action_Trans(ResDataObject* pOrgReq, DWORD timeout)
  1095. {
  1096. RET_STATUS ret = RET_FAILED;
  1097. //ResDataObject req;
  1098. if (Lock(timeout) == false)
  1099. {
  1100. return RET_TIMEOUT;
  1101. }
  1102. //printf("Thread:%d,Action Entry\n", GetCurrentThreadId());
  1103. if (m_ClosedFlag)
  1104. {
  1105. UnLock();
  1106. return ret;
  1107. }
  1108. if (IsThreadSafe() == false)
  1109. {
  1110. UnLock();
  1111. return RET_THREAD_INVALID;
  1112. }
  1113. //must be in SYNC mode
  1114. if (m_SyncMode != ACTION_SYNC)
  1115. {
  1116. UnLock();
  1117. return ret;
  1118. }
  1119. //make packet request
  1120. //m_ReqIdx = PacketAnalizer::MakeActionRequest(req, (*m_pFileHandle), pActionName, ReqParams, m_SyncMode);
  1121. //(*m_pActionName) = pActionName;
  1122. ////mLog::FDEBUG(" {$} at {$} Action_Trans Req: {$} ", m_strClientName, m_pFilePath->c_str(), pOrgReq->encode());
  1123. PacketAnalizer::UpdateContextTopic(pOrgReq, m_strDefaultTopic.c_str());
  1124. string actionName = PacketAnalizer::GetPacketKey(pOrgReq);
  1125. PacketAnalizer::UpdatePacketTransaction(*pOrgReq, m_strCurTransaction);
  1126. //send packet
  1127. //ret = pLogicDeivce->CmdFromLogicDev(&req);
  1128. int nret = PublishAction(pOrgReq, PacketAnalizer::GetActionTopic(m_pFilePath->c_str(), actionName.c_str()).c_str(), m_pMqttConn);
  1129. if (nret > 0)
  1130. {
  1131. ret = RET_SUCCEED;
  1132. }
  1133. else
  1134. {
  1135. m_ReqIdx = 0;
  1136. (*m_pActionName) = "";
  1137. }
  1138. UnLock();
  1139. return ret;
  1140. }
  1141. int LogicClient::Action_Req(const char *pActionName, ResDataObject &ReqParams, DWORD timeout, const char* pszDevicePath)
  1142. {
  1143. RET_STATUS ret = RET_FAILED;
  1144. ResDataObject req;
  1145. if (Lock(timeout) == false)
  1146. {
  1147. ////mLog::FERROR("try Lock failed for :Action[{$}]", pActionName);
  1148. return RET_TIMEOUT;
  1149. }
  1150. //printf("Thread:%d,Action Entry\n", GetCurrentThreadId());
  1151. if (m_ClosedFlag)
  1152. {
  1153. ////mLog::FERROR("Clinet {$} aready closed. :Action[{$}]", m_strClientName, pActionName);
  1154. UnLock();
  1155. return ret;
  1156. }
  1157. if (IsThreadSafe() == false)
  1158. {
  1159. ////mLog::FERROR("IsThreadSafe() == false ThreadName {$}. :Action[{$}]", m_strName, pActionName);
  1160. UnLock();
  1161. return RET_THREAD_INVALID;
  1162. }
  1163. if (pszDevicePath == nullptr && !HasAction(pActionName))
  1164. {
  1165. ////mLog::FERROR("pszDevicePath == nullptr && !HasAction(pActionName) ThreadName {$}. :Action[{$}] with devPath[{$}]", m_strName, pActionName, pszDevicePath);
  1166. return RET_SUCCEED;
  1167. }
  1168. //must be in SYNC mode
  1169. if (m_SyncMode != ACTION_SYNC)
  1170. {
  1171. ////mLog::FERROR("m_SyncMode != ACTION_SYNC {$} for :Action[{$}]", m_strClientName, pActionName);
  1172. UnLock();
  1173. return ret;
  1174. }
  1175. if (m_ReqIdx != 0)
  1176. {
  1177. ////mLog::FERROR(" {$} try call more than once Action_Req.Ignore first ActionReq Idx[{$}]:Action[{$}]", m_strClientName, m_ReqIdx, m_pActionName->c_str());
  1178. //GPRINTA_ERROR("try call more than once Action_Req.Ignore first ActionReq Idx[%u]:Action[%s] ",m_ReqIdx,m_pActionName->c_str());
  1179. m_ReqIdx = 0;
  1180. (*m_pActionName) = "";
  1181. }
  1182. ////mLog::FDEBUG(" {$} at {$} Action_Req {$} Req: {$} ", m_strClientName, m_pFilePath->c_str() , pActionName, ReqParams.encode());
  1183. //make packet request
  1184. m_ReqIdx = PacketAnalizer::MakeActionRequest(req, (*m_pFileHandle), pActionName, ReqParams, m_SyncMode);
  1185. (*m_pActionName) = pActionName;
  1186. PacketAnalizer::UpdateContextTopic(&req, m_strDefaultTopic.c_str());
  1187. PacketAnalizer::UpdatePacketTransaction(req, m_strCurTransaction);
  1188. //send packet
  1189. //ret = pLogicDeivce->CmdFromLogicDev(&req);
  1190. int nret = PublishAction(&req, PacketAnalizer::GetActionTopic(pszDevicePath == nullptr ? m_pFilePath->c_str() : pszDevicePath, pActionName).c_str(), m_pMqttConn);
  1191. if (nret > 0)
  1192. {
  1193. ret = RET_SUCCEED;
  1194. }
  1195. else
  1196. {
  1197. m_ReqIdx = 0;
  1198. (*m_pActionName) = "";
  1199. }
  1200. UnLock();
  1201. return ret;
  1202. }
  1203. int LogicClient::Action_Res(const char *pActionName, ResDataObject &ResParams, DWORD timeout)
  1204. {
  1205. RET_STATUS ret = RET_FAILED;
  1206. ResDataObject res;
  1207. if (Lock(timeout) == false)
  1208. {
  1209. return RET_TIMEOUT;
  1210. }
  1211. if (m_ClosedFlag)
  1212. {
  1213. //for test
  1214. //GPRINTA_DEBUG("Client:%d,Action_Res Closed ReqIdx:%d,Name:%s",this, m_ReqIdx, (*m_pActionName).c_str());
  1215. m_ReqIdx = 0;
  1216. (*m_pActionName) = "";
  1217. UnLock();
  1218. return RET_INVALID;
  1219. }
  1220. if (IsThreadSafe() == false)
  1221. {
  1222. //for test
  1223. //GPRINTA_DEBUG("Client:%d,Action_Res ThreadUnSafe ReqIdx:%d,Name:%s",this, m_ReqIdx, (*m_pActionName).c_str());
  1224. m_ReqIdx = 0;
  1225. (*m_pActionName) = "";
  1226. UnLock();
  1227. return RET_THREAD_INVALID;
  1228. }
  1229. //must be in SYNC mode
  1230. if (m_SyncMode != ACTION_SYNC)
  1231. {
  1232. //for test
  1233. //GPRINTA_DEBUG("Client:%d,Action_Res No Sync ReqIdx:%d,Name:%s",this, m_ReqIdx, (*m_pActionName).c_str());
  1234. m_ReqIdx = 0;
  1235. (*m_pActionName) = "";
  1236. UnLock();
  1237. return ret;
  1238. }
  1239. if (m_ReqIdx == 0)
  1240. {
  1241. //for test
  1242. //GPRINTA_DEBUG("Client:%d,Action_Res ReqIdx == 0 ReqIdx:%d,Name:%s",this, m_ReqIdx, (*m_pActionName).c_str());
  1243. (*m_pActionName) = "";
  1244. UnLock();
  1245. return ret;
  1246. }
  1247. //ReqIdx must not zero
  1248. while (m_pPacketReceivedQue->WaitForInQue(timeout) == WAIT_OBJECT_0)
  1249. {
  1250. //got response
  1251. if (m_pPacketReceivedQue->DeQueue(res))
  1252. {
  1253. //update notify evt
  1254. IsDataArrived();
  1255. if ((m_ReqIdx == PacketAnalizer::GetPacketIdx(&res)) &&
  1256. (PacketAnalizer::GetPacketType(&res) == PACKET_TYPE_RES))
  1257. {
  1258. //check cmd
  1259. if ((PacketAnalizer::GetPacketCmd(&res) != PACKET_CMD_EXE) ||
  1260. (PacketAnalizer::GetPacketKey(&res) != pActionName))
  1261. {
  1262. //wrong packet
  1263. continue;
  1264. }
  1265. //got match response
  1266. ret = ProcessActionResponse(res, ResParams);
  1267. //for test
  1268. //GPRINTA_DEBUG("Client:%d,Action_Res Finished ReqIdx:%d,Name:%s",this, m_ReqIdx, (*m_pActionName).c_str());
  1269. m_ReqIdx = 0;
  1270. (*m_pActionName) = "";
  1271. UnLock();
  1272. return ret;
  1273. }
  1274. else
  1275. {
  1276. //got other one
  1277. //check priority one => CLOSE
  1278. if (PacketAnalizer::GetPacketCmd(&res) == PACKET_CMD_CLOSE)
  1279. {
  1280. //we have disconnect here
  1281. ret = ProcessClose(res);
  1282. //for test
  1283. //GPRINTA_DEBUG("Client:%d,Action_Res Got Closed ReqIdx:%d,Name:%s",this, m_ReqIdx, (*m_pActionName).c_str());
  1284. m_ReqIdx = 0;
  1285. (*m_pActionName) = "";
  1286. UnLock();
  1287. return ret;
  1288. }
  1289. ////mLog::FERROR("{$} got device[{$}] what packet {$}",m_strClientName, m_strDevicePath, res.encode());
  1290. m_pPacketWaitQue->InQueue(res);
  1291. IsDataArrived();
  1292. }
  1293. }
  1294. }
  1295. ret = RET_TIMEOUT;
  1296. //for test
  1297. //GPRINTA_DEBUG("Client:%d,Action_Res Got Timeout ReqIdx:%d,Name:%s",this, m_ReqIdx, (*m_pActionName).c_str());
  1298. //m_ReqIdx = 0;
  1299. //(*m_pActionName) = "";
  1300. UnLock();
  1301. return ret;
  1302. }
  1303. std::shared_ptr<LinuxEvent> LogicClient::GetResponseHandle()
  1304. {
  1305. return m_ResponseEvent;
  1306. }
  1307. void LogicClient::SubScribeTopic(const char* pszTopic, bool bSubscribe)
  1308. {
  1309. if (bSubscribe)
  1310. SubscribeTopic(m_pMqttConn, pszTopic);
  1311. else
  1312. UnSubscribe(m_pMqttConn, pszTopic);
  1313. }
  1314. int LogicClient::Action(const char *pActionName, ResDataObject &ReqParams, ResDataObject &ResParams, DWORD timeout, const char* pszDevicePath)
  1315. {
  1316. RET_STATUS ret = RET_FAILED;
  1317. ResDataObject req, res;
  1318. if (Lock(timeout) == false)
  1319. {
  1320. std::cout << "[ERROR] " << m_strClientName << " try Action " << pActionName << " LockTimeout" << std::endl;
  1321. ////mLog::FERROR("{$} try Action {$} LockTimeout", m_strClientName, pActionName);
  1322. return RET_TIMEOUT;
  1323. }
  1324. //printf("Thread:%d,Action Entry\n", GetCurrentThreadId());
  1325. std::cout << "[DEBUG] Thread:" << GetCurrentThreadId() << ", Action Entry - " << pActionName << std::endl;
  1326. if (m_ClosedFlag)
  1327. {
  1328. ////mLog::FERROR("{$} try Action {$} Aready Closed", m_strClientName, pActionName);
  1329. std::cout << "[ERROR] " << m_strClientName << " try Action " << pActionName << " Already Closed" << std::endl;
  1330. UnLock();
  1331. return ret;
  1332. }
  1333. if (IsThreadSafe() == false)
  1334. {
  1335. ////mLog::FERROR("{$} try Action {$} IsThreadSafe false", m_strClientName, pActionName);
  1336. std::cout << "[ERROR] " << m_strClientName << " try Action " << pActionName << " IsThreadSafe false" << std::endl;
  1337. UnLock();
  1338. return RET_THREAD_INVALID;
  1339. }
  1340. if (pszDevicePath == nullptr && !HasAction(pActionName))
  1341. {
  1342. std::cout << "[ERROR] " << m_strClientName << " try Action " << pActionName
  1343. << " with pszDevicePath == nullptr && !HasAction(pActionName)" << std::endl;
  1344. ////mLog::FERROR("{$} try Action {$} with pszDevicePath == nullptr && !HasAction(pActionName)", m_strClientName, pActionName);
  1345. UnLock();
  1346. return RET_SUCCEED;
  1347. }
  1348. std::cout << "[DEBUG] " << m_strClientName << " at "
  1349. << (pszDevicePath == nullptr ? m_pFilePath->c_str() : pszDevicePath)
  1350. << " Action " << pActionName << " Req: " << ReqParams.encode() << std::endl;
  1351. ////mLog::FDEBUG(" {$} at {$} Action {$} Req: {$} ", m_strClientName, pszDevicePath == nullptr ? m_pFilePath->c_str() : pszDevicePath, pActionName, ReqParams.encode());
  1352. //make packet request
  1353. DWORD reqidx = PacketAnalizer::MakeActionRequest(req, (*m_pFileHandle), pActionName, ReqParams, m_SyncMode);
  1354. PacketAnalizer::UpdateContextTopic(&req, m_strDefaultTopic.c_str());
  1355. PacketAnalizer::UpdatePacketTransaction(req, m_strCurTransaction);
  1356. UnLock();
  1357. //send packet int ActionAndResp(ccos_mqtt_connection* hConnection,const char* pAction, ResDataObject* pContext, const char* pszTopic, ResDataObject& resObj, DWORD dwWaitTime )
  1358. //ret = pLogicDeivce->CmdFromLog icDev(&req);
  1359. int nret = ActionAndRespWithConnDefalt(m_pMqttConn, pActionName,req, &ReqParams,
  1360. PacketAnalizer::GetActionTopic(pszDevicePath==nullptr?m_pFilePath->c_str():pszDevicePath,pActionName).c_str(),
  1361. res, timeout);
  1362. if (nret > 0)
  1363. {
  1364. ret = RET_SUCCEED;
  1365. DWORD retId = PacketAnalizer::GetPacketIdx(&res);
  1366. PACKET_TYPE tpRes = PacketAnalizer::GetPacketType(&res);
  1367. if ((reqidx == retId) &&
  1368. (tpRes == PACKET_TYPE_RES))
  1369. {
  1370. //check cmd
  1371. if ((PacketAnalizer::GetPacketCmd(&res) == PACKET_CMD_EXE))
  1372. {
  1373. ret = ProcessActionResponse(res, ResParams);
  1374. //UnLock();
  1375. return ret;
  1376. }
  1377. }
  1378. else
  1379. {
  1380. ////mLog::FINFO("{$} What Resp ? [{$}]", m_strClientName, PacketAnalizer::GetPacketKey(&res));
  1381. std::cout << "[INFO] " << m_strClientName << " What Resp ? [" << PacketAnalizer::GetPacketKey(&res) << "]" << std::endl;
  1382. }
  1383. }
  1384. ////mLog::FERROR("Failed={$} : {$} at {$} Action {$} Req: {$} ", (int)ret, m_strClientName, pszDevicePath == nullptr ? m_pFilePath->c_str() : pszDevicePath, pActionName, ReqParams.encode());
  1385. //else
  1386. //{
  1387. // ReOpenDevice();
  1388. //}
  1389. return ret;
  1390. }
  1391. bool LogicClient::IsThreadSafe()
  1392. {
  1393. return true;
  1394. //return (m_OwnerThreadId == GetCurrentThreadId());
  1395. }
  1396. std::shared_ptr<LinuxEvent> LogicClient::GetNotifyHandle()
  1397. {
  1398. return m_NotifyEvent;
  1399. }
  1400. void LogicClient::ClearQueue_Internal()
  1401. {
  1402. //m_pPacketReceivedQue->Lock();
  1403. m_pPacketReceivedQue->Clear();
  1404. m_pPacketWaitQue->Clear();
  1405. m_NotifyEvent->ResetEvent();
  1406. //m_pPacketReceivedQue->UnLock();
  1407. }
  1408. bool LogicClient::DeQueuePacket_Internal(ResDataObject &Packet)
  1409. {
  1410. bool ret = false;
  1411. //m_pPacketReceivedQue->Lock();
  1412. //wait que first
  1413. if (m_pPacketWaitQue->DeQueue(Packet))
  1414. {
  1415. ret = true;
  1416. }
  1417. else if (m_pPacketReceivedQue->DeQueue(Packet))
  1418. {
  1419. ret = true;
  1420. }
  1421. else
  1422. {
  1423. ret = false;
  1424. }
  1425. //set event
  1426. if (m_pPacketWaitQue->size() > 0 || m_pPacketReceivedQue->size() > 0)
  1427. {
  1428. m_NotifyEvent->SetEvent();
  1429. }
  1430. else
  1431. {
  1432. m_NotifyEvent->ResetEvent();
  1433. }
  1434. //m_pPacketReceivedQue->UnLock();
  1435. return ret;
  1436. }
  1437. void LogicClient::CleanupQueue()
  1438. {
  1439. if (Lock(1000) == false)
  1440. {
  1441. //printf("Can't Lock Que for cleanup\n");
  1442. return;
  1443. }
  1444. ClearQueue_Internal();
  1445. UnLock();
  1446. }
  1447. bool LogicClient::IsDataArrived()
  1448. {
  1449. bool ret = false;
  1450. //the packet queue must be encapsulated
  1451. //m_pPacketReceivedQue->Lock();
  1452. if (m_pPacketWaitQue->size() > 0 || m_pPacketReceivedQue->size() > 0)
  1453. {
  1454. ////mLog::FDEBUG("{$} have packet to process ..m_pPacketWaitQue {$} .m_pPacketReceivedQue {$}.", m_strClientName, m_pPacketWaitQue->size(), m_pPacketReceivedQue->size());
  1455. ret = true;
  1456. m_ResponseEvent->SetEvent();
  1457. }
  1458. else
  1459. {
  1460. m_ResponseEvent->ResetEvent();
  1461. }
  1462. //m_pPacketReceivedQue->UnLock();
  1463. //m_pPacketNotifyQue->Lock();
  1464. if (m_pPacketNotifyQue->size() > 0 )
  1465. {
  1466. ////mLog::FDEBUG("{$} have notify to process ..m_pPacketNotifyQue {$} .", m_strClientName, m_pPacketNotifyQue->size());
  1467. ret = true;
  1468. m_NotifyEvent->SetEvent();
  1469. }
  1470. else
  1471. {
  1472. m_NotifyEvent->ResetEvent();
  1473. }
  1474. //m_pPacketNotifyQue->UnLock();
  1475. return ret;
  1476. }
  1477. /*
  1478. * PACKET_CMD CommonLogicClient::ReadCmd(ResDataObject &CmdObject)
  1479. {
  1480. PACKET_CMD ret = PACKET_CMD_NONE;
  1481. while (IsDataArrived())
  1482. {
  1483. ret = LogicClient::ReadCmd(CmdObject);
  1484. if (ret != PACKET_CMD_NONE)
  1485. {
  1486. return ret;
  1487. }
  1488. }
  1489. return ret;
  1490. }
  1491. */
  1492. /// <summary>
  1493. /// 子类CommonLogicClient ReadCMD判断 需要检查逻辑,进行兼容性修改
  1494. /// </summary>
  1495. /// <param name="CmdObject"></param>
  1496. /// <returns></returns>
  1497. PACKET_CMD LogicClient::ReadCmd(ResDataObject &CmdObject)
  1498. {
  1499. PACKET_CMD cmd = PACKET_CMD_NONE;
  1500. //check connection
  1501. if (IsClosed() == true)
  1502. {
  1503. cmd = PACKET_CMD_CLOSE;
  1504. ClearQueue_Internal();
  1505. return cmd;
  1506. }
  1507. if(IsThreadSafe() == false)
  1508. {
  1509. ClearQueue_Internal();
  1510. return cmd;
  1511. }
  1512. if (Lock(100) == false)
  1513. {
  1514. std::cout << "LogicClient::ReadCmd " << m_strClientName << "] Lock timeout Receved Que has [" << m_pPacketReceivedQue->size() << "] Object" << endl;
  1515. return cmd;
  1516. }
  1517. //read receive que second
  1518. RET_STATUS ret = RET_NOSUPPORT;
  1519. PACKET_TYPE rType;
  1520. ResDataObject rMsg;
  1521. string key;
  1522. //read wait que first
  1523. while (DeQueuePacket_Internal(rMsg))
  1524. {
  1525. rType = PacketAnalizer::GetPacketType(&rMsg);
  1526. cmd = PacketAnalizer::GetPacketCmd(&rMsg);
  1527. key = PacketAnalizer::GetPacketKey(&rMsg);
  1528. std::cout << "ReadCmd [" << m_strClientName << "] Type: " << rType << "cmd: " << cmd << " Key: " << key << endl;
  1529. ////mLog::FINFO("{$} ReadCmd Type: {$} cmd: {$} Key: {$} ", m_strClientName, (int)rType, (int)cmd, key);
  1530. if (cmd > PACKET_CMD_NONE && cmd < PACKET_CMD_MAX)
  1531. {
  1532. if (rType == PACKET_TYPE_RES)
  1533. {
  1534. //1.response
  1535. CmdObject = rMsg;
  1536. //[open] need care
  1537. if (cmd == PACKET_CMD_OPEN)
  1538. {
  1539. ret = ProcessOpenResponse(rMsg);
  1540. if (ret >= RET_SUCCEED)
  1541. {
  1542. UnLock();
  1543. return cmd;
  1544. }
  1545. }
  1546. }
  1547. else if (rType == PACKET_TYPE_NOTIFY)
  1548. {
  1549. //2.notify
  1550. UpdateDeviceResource(cmd, rMsg);
  1551. CmdObject = rMsg;
  1552. UnLock();
  1553. return cmd;
  1554. }
  1555. else
  1556. {
  1557. //ignore it
  1558. cmd = PACKET_CMD_NONE;
  1559. }
  1560. }
  1561. rMsg.clear();
  1562. }
  1563. UnLock();
  1564. return cmd;
  1565. }
  1566. PACKET_CMD LogicClient::ReadNotify(ResDataObject& CmdObject)
  1567. {
  1568. CmdObject.clear();
  1569. //m_pPacketNotifyQue->Lock();
  1570. if (m_pPacketNotifyQue->DeQueue(CmdObject))
  1571. {
  1572. //m_pPacketNotifyQue->UnLock();
  1573. return PacketAnalizer::GetPacketCmd(&CmdObject);
  1574. }
  1575. m_NotifyEvent->ResetEvent();
  1576. if(m_pPacketReceivedQue->DeQueue(CmdObject))
  1577. {
  1578. return PacketAnalizer::GetPacketCmd(&CmdObject);
  1579. }
  1580. m_ResponseEvent->ResetEvent();
  1581. //m_pPacketNotifyQue->UnLock();
  1582. return PACKET_CMD_NONE;
  1583. }
  1584. RET_STATUS SYSTEM_CALL LogicClient::WaitForState(const char* state, DWORD timeout)
  1585. {
  1586. PACKET_CMD cmd;
  1587. ResDataObject data;
  1588. ResDataObject context;
  1589. string strKey, value;
  1590. DWORD firstTick = GetTickCount();
  1591. std::shared_ptr<LinuxEvent> notifyhandle = GetNotifyHandle();
  1592. while (true)
  1593. {
  1594. while (IsDataArrived())
  1595. {
  1596. data.clear();
  1597. cmd = ReadCmd(data);
  1598. strKey = PacketAnalizer::GetPacketKey(&data);
  1599. PacketAnalizer::GetPacketContext(&data, context);
  1600. value = (const char *)context;
  1601. if (strKey == "CurrentState")
  1602. {
  1603. if (value == string(state))
  1604. return RET_SUCCEED;
  1605. }
  1606. else if (strKey == string(state))
  1607. {
  1608. return RET_SUCCEED;
  1609. }
  1610. }
  1611. if ((GetTickCount() - firstTick) < timeout)
  1612. {
  1613. if (notifyhandle->Wait(timeout))
  1614. {
  1615. continue;
  1616. }
  1617. }
  1618. break;
  1619. }
  1620. return RET_FAILED;
  1621. }
  1622. RET_STATUS LogicClient::UpdateDeviceResource(int cmdType,ResDataObject &packet)
  1623. {
  1624. RET_STATUS ret = RET_SUCCEED;
  1625. Lock();
  1626. try {
  1627. string strKey = PacketAnalizer::GetPacketKey(&packet);
  1628. ResDataObject ResContext;
  1629. PacketAnalizer::GetPacketContext(&packet, ResContext);
  1630. if (cmdType == PACKET_CMD_UPDATE)
  1631. {
  1632. if (m_DeviceResource[ATTRIBUTE].GetFirstOf(strKey.c_str()) >= 0)
  1633. {
  1634. m_DeviceResource[ATTRIBUTE].update(strKey.c_str(), ResContext);
  1635. }
  1636. }
  1637. else if (cmdType == PACKET_CMD_ADD)
  1638. {
  1639. if (strKey=="ErrorList")
  1640. {
  1641. /*pLogicDeivce->*/GetDeviceResource(&m_DeviceResource);
  1642. }
  1643. else if (m_DeviceResource[ATTRIBUTE].GetFirstOf(strKey.c_str()) < 0)
  1644. {
  1645. m_DeviceResource[ATTRIBUTE].add(strKey.c_str(), ResContext);
  1646. }
  1647. else if (m_DeviceResource[ATTRIBUTE].GetFirstOf(strKey.c_str()) >= 0 &&
  1648. (string)ResContext != "" &&m_DeviceResource[ATTRIBUTE][strKey.c_str()].GetFirstOf(((string)ResContext).c_str()) < 0)
  1649. {
  1650. m_DeviceResource[ATTRIBUTE][strKey.c_str()].add(((string)ResContext).c_str(), "");
  1651. }
  1652. }
  1653. else if (cmdType == PACKET_CMD_DEL)
  1654. {
  1655. if (strKey == "ErrorList")
  1656. {
  1657. /*pLogicDeivce->*/GetDeviceResource(&m_DeviceResource);
  1658. }
  1659. else if (m_DeviceResource[ATTRIBUTE].GetFirstOf(strKey.c_str()) >= 0 )
  1660. {
  1661. if ((string)ResContext != "" && m_DeviceResource[ATTRIBUTE][strKey.c_str()].GetFirstOf(((string)ResContext).c_str()) >= 0)
  1662. {
  1663. m_DeviceResource[ATTRIBUTE][strKey.c_str()].eraseOneOf(((string)ResContext).c_str(), m_DeviceResource[ATTRIBUTE][strKey.c_str()].GetFirstOf(((string)ResContext).c_str()));
  1664. }
  1665. else
  1666. {
  1667. m_DeviceResource[ATTRIBUTE].eraseOneOf(strKey.c_str(), m_DeviceResource[ATTRIBUTE].GetFirstOf(strKey.c_str()));
  1668. }
  1669. }
  1670. }
  1671. }
  1672. catch (...)
  1673. {
  1674. ret = RET_FAILED;
  1675. }
  1676. UnLock();
  1677. return ret;
  1678. }