libcm is a C development framework with an emphasis on audio signal processing applications.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

cmRtNet.c 19KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789
  1. #include "cmGlobal.h"
  2. #include "cmRpt.h"
  3. #include "cmErr.h"
  4. #include "cmCtx.h"
  5. #include "cmMem.h"
  6. #include "cmMallocDebug.h"
  7. #include "cmUdpPort.h"
  8. #include "cmRtNet.h"
  9. #include "cmTime.h"
  10. #include "cmRtSysMsg.h"
  11. enum
  12. {
  13. kLocalNodeNetFl = 0x01,
  14. kValidNodeNetFl = 0x02
  15. };
  16. typedef enum
  17. {
  18. kHelloSelNetId,
  19. kNodeSelNetId,
  20. kEndpointSelNetId,
  21. kEndpointAckSelNetId,
  22. kDoneSelNetId,
  23. } cmRtNetSelId_t;
  24. typedef struct cmRtNetEnd_str
  25. {
  26. cmChar_t* label;
  27. unsigned id;
  28. struct cmRtNetEnd_str* link;
  29. } cmRtNetEnd_t;
  30. typedef struct cmRtNetNode_str
  31. {
  32. cmChar_t* label;
  33. struct sockaddr_in sockaddr;
  34. cmChar_t* addr;
  35. cmUdpPort_t port;
  36. unsigned flags;
  37. unsigned endPtIdx; // tracks the next endpoint to send during sync-mode
  38. unsigned endPtCnt; // local-node=actual cnt of endpt's remote-node:expected cnt of endpt's
  39. cmTimeSpec_t lastSendTime;
  40. cmRtNetEnd_t* ends;
  41. struct cmRtNetNode_str* link;
  42. } cmRtNetNode_t;
  43. typedef struct
  44. {
  45. cmErr_t err;
  46. cmUdpH_t udpH;
  47. cmUdpCallback_t cbFunc;
  48. void* cbArg;
  49. cmRtNetNode_t* nodes;
  50. cmRtNetNode_t* localNode;
  51. unsigned udpRecvBufByteCnt;
  52. unsigned udpTimeOutMs;
  53. unsigned interSyncSendTimeMs;
  54. } cmRtNet_t;
  55. typedef struct
  56. {
  57. cmRtSysMsgHdr_t hdr;
  58. cmRtNetSelId_t selId;
  59. const cmChar_t* label; // node or endpoint label
  60. unsigned id; // endptCnt or endpoint id
  61. } cmRtNetSyncMsg_t;
  62. cmRtNetH_t cmRtNetNullHandle = cmSTATIC_NULL_HANDLE;
  63. cmRtNet_t* _cmRtNetHandleToPtr( cmRtNetH_t h )
  64. {
  65. cmRtNet_t* p = (cmRtNet_t*)h.h;
  66. assert( p != NULL );
  67. return p;
  68. }
  69. void _cmRtNetVRpt( cmRtNet_t* p, const cmChar_t* fmt, va_list vl )
  70. {
  71. cmRptVPrintf(p->err.rpt,fmt,vl);
  72. }
  73. void _cmRtNetRpt( cmRtNet_t* p, const cmChar_t* fmt, ... )
  74. {
  75. va_list vl;
  76. va_start(vl,fmt);
  77. _cmRtNetVRpt(p,fmt,vl);
  78. va_end(vl);
  79. }
  80. cmRtNetNode_t* _cmRtNetFindNode( cmRtNet_t* p, const cmChar_t* label )
  81. {
  82. if( label == NULL )
  83. return NULL;
  84. cmRtNetNode_t* np = p->nodes;
  85. for(; np!=NULL; np=np->link)
  86. if( strcmp(label,np->label)==0)
  87. return np;
  88. return NULL;
  89. }
  90. cmRtNetNode_t* _cmRtNetFindNodeFromSockAddr( cmRtNet_t* p, const struct sockaddr_in* saddr )
  91. {
  92. if( saddr == NULL )
  93. return NULL;
  94. cmRtNetNode_t* np = p->nodes;
  95. for(; np!=NULL; np=np->link)
  96. if( np->sockaddr.sin_addr.s_addr == saddr->sin_addr.s_addr && np->sockaddr.sin_port == saddr->sin_port )
  97. return np;
  98. return NULL;
  99. }
  100. void _cmRtNetFreeNode( cmRtNetNode_t* np )
  101. {
  102. cmRtNetEnd_t* ep = np->ends;
  103. while( ep != NULL )
  104. {
  105. cmRtNetEnd_t* nep = ep->link;
  106. cmMemFree(ep->label);
  107. cmMemFree(ep);
  108. ep = nep;
  109. }
  110. cmMemFree(np->label);
  111. cmMemFree(np->addr);
  112. cmMemFree(np);
  113. }
  114. void _cmRtNetReleaseNodes( cmRtNet_t* p )
  115. {
  116. cmRtNetNode_t* np = p->nodes;
  117. while( np != NULL )
  118. {
  119. cmRtNetNode_t* nnp = np->link;
  120. _cmRtNetFreeNode(np);
  121. np = nnp;
  122. }
  123. p->nodes = NULL;
  124. p->localNode = NULL;
  125. }
  126. cmRtNetRC_t _cmRtNetReleaseNode( cmRtNet_t* p, cmRtNetNode_t* np )
  127. {
  128. cmRtNetNode_t* cnp = p->nodes;
  129. cmRtNetNode_t* pnp = NULL;
  130. while( cnp != NULL )
  131. {
  132. cmRtNetNode_t* nnp = cnp->link;
  133. if( np == cnp )
  134. {
  135. if( pnp == NULL )
  136. p->nodes = np->link;
  137. else
  138. pnp->link = np->link;
  139. _cmRtNetFreeNode(np);
  140. return kOkNetRC;
  141. }
  142. pnp = np;
  143. cnp = nnp;
  144. }
  145. assert(0);
  146. return cmErrMsg(&p->err,kNodeNotFoundNetRC,"Node to release not found.");
  147. }
  148. cmRtNetRC_t _cmRtNetCreateNode( cmRtNet_t* p, const cmChar_t* label, const cmChar_t* addr, cmUdpPort_t port, const struct sockaddr_in* saddr, unsigned flags, unsigned endPtCnt )
  149. {
  150. cmRtNetRC_t rc = kOkNetRC;
  151. cmRtNetNode_t* np;
  152. if( label == NULL )
  153. return cmErrMsg(&p->err,kInvalidLabelNetRC,"A null or blank node label was encountered.");
  154. if((np = _cmRtNetFindNode(p,label)) != NULL )
  155. return cmErrMsg(&p->err,kDuplLabelNetRC,"The node label '%s' is already in use.",cmStringNullGuard(label));
  156. np = cmMemAllocZ(cmRtNetNode_t,1);
  157. np->label = cmMemAllocStr(label);
  158. np->sockaddr = *saddr;
  159. np->addr = addr==NULL ? NULL : cmMemAllocStr(addr);
  160. np->port = port;
  161. np->flags = flags;
  162. np->endPtCnt = endPtCnt;
  163. np->link = p->nodes;
  164. p->nodes = np;
  165. return rc;
  166. }
  167. cmRtNetEnd_t* _cmRtNetFindNodeEnd(cmRtNetNode_t* np, const cmChar_t* endPtLabel )
  168. {
  169. cmRtNetEnd_t* ep = np->ends;
  170. for(; ep!=NULL; ep=ep->link)
  171. if( strcmp(ep->label,endPtLabel)==0 )
  172. return ep;
  173. return NULL;
  174. }
  175. cmRtNetEnd_t* _cmRtNetIndexToEndpoint( cmRtNet_t* p, cmRtNetNode_t* np, unsigned endIndex )
  176. {
  177. cmRtNetEnd_t* ep = np->ends;
  178. unsigned i;
  179. for(i=0; ep!=NULL; ep=ep->link)
  180. {
  181. if( i == endIndex )
  182. return ep;
  183. ++i;
  184. }
  185. return NULL;
  186. }
  187. cmRtNetRC_t _cmRtNetCreateEndpoint( cmRtNet_t* p, cmRtNetNode_t* np, const cmChar_t* endPtLabel, unsigned endPtId )
  188. {
  189. if( endPtLabel == NULL )
  190. return cmErrMsg(&p->err,kInvalidLabelNetRC,"A null or blank node label was encountered.");
  191. if( _cmRtNetFindNodeEnd( np, endPtLabel) != NULL)
  192. return cmErrMsg(&p->err,kDuplEndNetRC,"A duplicate endpoint ('%s') was encountered on node '%s'.",endPtLabel,np->label);
  193. cmRtNetRC_t rc = kOkNetRC;
  194. cmRtNetEnd_t* ep = cmMemAllocZ(cmRtNetEnd_t,1);
  195. ep->label = cmMemAllocStr(endPtLabel);
  196. ep->id = endPtId;
  197. ep->link = np->ends;
  198. np->ends = ep;
  199. return rc;
  200. }
  201. unsigned _cmRtNetSyncMsgSerialByteCount( const cmRtNetSyncMsg_t* m )
  202. { return sizeof(cmRtNetSyncMsg_t) + (m->label==NULL ? 1 : strlen(m->label) + 1); }
  203. cmRtNetRC_t _cmRtNetSerializeSyncMsg( cmRtNet_t* p, const cmRtNetSyncMsg_t* m, void* buf, unsigned n )
  204. {
  205. unsigned bn = _cmRtNetSyncMsgSerialByteCount(m);
  206. char* b = (char*)buf;
  207. if( bn > n )
  208. return cmErrMsg(&p->err,kBufToSmallNetRC,"Serialize buffer too small.");
  209. memcpy(b,m,sizeof(*m));
  210. strcpy(b + sizeof(*m),m->label==NULL ? "" : m->label);
  211. return kOkNetRC;
  212. }
  213. cmRtNetRC_t _cmRtNetDeserializeSyncMsg( const void* buf, unsigned n, cmRtNetSyncMsg_t* m )
  214. {
  215. assert( n > sizeof(*m));
  216. memcpy(m,buf,sizeof(*m));
  217. const cmRtNetSyncMsg_t* mp = (const cmRtNetSyncMsg_t*)buf;
  218. const cmChar_t* s = (const cmChar_t*)(mp+1);
  219. m->label = cmMemAllocStr(s);
  220. return kOkNetRC;
  221. }
  222. cmRtNetRC_t _cmRtNetSendSyncMsg( cmRtNet_t* p, cmRtNetNode_t* np, cmRtNetSelId_t selId, const cmChar_t* msgLabel, unsigned msgId )
  223. {
  224. cmRtNetSyncMsg_t m;
  225. cmRtNetRC_t rc = kOkNetRC;
  226. cmUdpRC_t udpRC = kOkUdpRC;
  227. m.hdr.rtSubIdx = cmInvalidIdx;
  228. m.hdr.selId = kNetSyncSelRtId;
  229. m.selId = selId;
  230. m.label = msgLabel;
  231. m.id = msgId;
  232. // determine size of msg to send
  233. unsigned n = _cmRtNetSyncMsgSerialByteCount(&m);
  234. cmChar_t buf[n];
  235. // serialize msg into buf[]
  236. if((rc = _cmRtNetSerializeSyncMsg(p,&m,buf,n)) != kOkNetRC )
  237. return rc;
  238. // send the msg
  239. if( np==p->localNode )
  240. udpRC = cmUdpSend2(p->udpH, buf, n, "255.255.255.255", np->port );
  241. else
  242. udpRC = cmUdpSendTo(p->udpH, buf, n, &np->sockaddr );
  243. // check for send errors
  244. if( udpRC != kOkUdpRC )
  245. {
  246. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"Sync msg. send on UDP port failed.");
  247. }
  248. else
  249. {
  250. // record the last send time
  251. cmTimeGet(&np->lastSendTime);
  252. }
  253. return rc;
  254. }
  255. cmRtNetRC_t _cmRtNetFree( cmRtNet_t* p )
  256. {
  257. cmRtNetRC_t rc = kOkNetRC;
  258. if( cmUdpFree(&p->udpH) != kOkUdpRC )
  259. cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP Port free failed.");
  260. _cmRtNetReleaseNodes(p);
  261. cmMemFree(p);
  262. return rc;
  263. }
  264. cmRtNetRC_t cmRtNetAlloc( cmCtx_t* ctx, cmRtNetH_t* hp, cmUdpCallback_t cbFunc, void* cbArg )
  265. {
  266. cmRtNetRC_t rc;
  267. if((rc = cmRtNetFree(hp)) != kOkNetRC )
  268. return rc;
  269. cmRtNet_t* p = cmMemAllocZ(cmRtNet_t,1);
  270. cmErrSetup(&p->err,&ctx->rpt,"cmRtNet");
  271. // allocate the UDP port
  272. if(cmUdpAlloc(ctx,&p->udpH) != kOkUdpRC )
  273. {
  274. cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP Port allocate failed.");
  275. goto errLabel;
  276. }
  277. p->udpTimeOutMs = 50;
  278. p->udpRecvBufByteCnt = 8192;
  279. p->interSyncSendTimeMs = 10;
  280. p->cbFunc = cbFunc;
  281. p->cbArg = cbArg;
  282. hp->h = p;
  283. errLabel:
  284. if(rc != kOkNetRC )
  285. _cmRtNetFree(p);
  286. return rc;
  287. }
  288. cmRtNetRC_t cmRtNetFree( cmRtNetH_t* hp )
  289. {
  290. cmRtNetRC_t rc = kOkNetRC;
  291. if( hp==NULL || cmRtNetIsValid(*hp)==false )
  292. return rc;
  293. cmRtNet_t* p = _cmRtNetHandleToPtr(*hp);
  294. if((rc = _cmRtNetFree(p)) != kOkNetRC )
  295. return rc;
  296. hp->h = NULL;
  297. return rc;
  298. }
  299. const cmChar_t* cmRtNetLocalHostName( cmRtNetH_t h )
  300. {
  301. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  302. return cmUdpHostName(p->udpH);
  303. }
  304. bool cmRtNetIsValid( cmRtNetH_t h )
  305. { return h.h !=NULL; }
  306. cmUdpH_t cmRtNetUdpPortHandle( cmRtNetH_t h )
  307. {
  308. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  309. return p->udpH;
  310. }
  311. cmRtNetRC_t _cmRtNetSendEndpointReplyMsg( cmRtNet_t* p, cmRtNetNode_t* np )
  312. {
  313. cmRtNetRC_t rc = kOkNetRC;
  314. cmRtNetEnd_t* ep;
  315. const cmChar_t* msgLabel = NULL;
  316. unsigned msgId = cmInvalidId;
  317. cmRtNetSelId_t selId = kEndpointSelNetId;
  318. const cmChar_t* rptLabel = "endpoint";
  319. if( np == NULL )
  320. return cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with an endpoint reply was not found.");
  321. // if all of the endpoints have been sent to this node ...
  322. if((ep = _cmRtNetIndexToEndpoint(p,p->localNode,np->endPtIdx)) == NULL )
  323. {
  324. if( np->endPtIdx == p->localNode->endPtCnt )
  325. {
  326. selId = kDoneSelNetId;
  327. rptLabel = "done";
  328. }
  329. else
  330. {
  331. selId = kEndpointAckSelNetId;
  332. rptLabel = "ep ack";
  333. }
  334. }
  335. else
  336. {
  337. msgLabel = ep->label;
  338. msgId = ep->id;
  339. }
  340. // notify the remote node that all endpoints have been sent
  341. if((rc = _cmRtNetSendSyncMsg(p,np,selId,msgLabel,msgId )) != kOkNetRC )
  342. rc = cmErrMsg(&p->err,rc,"Send '%s' to %s:%s:%i failed.",rptLabel,cmStringNullGuard(np->label),cmStringNullGuard(np->addr),np->port);
  343. else
  344. _cmRtNetRpt(p,"Sent %s.\n",cmStringNullGuard(rptLabel));
  345. np->endPtIdx += 1;
  346. return rc;
  347. }
  348. // When the network message recieve function (See cmRtNetAlloc() 'cbFunc')
  349. // receives a message with the cmRtSysMsgHdr_t.selId == kNetSyncSelRtId
  350. // it should call this function to update the current sync state of the
  351. // cmRtNet.
  352. cmRtNetRC_t _cmRtNetSyncModeRecv( cmRtNet_t* p, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
  353. {
  354. cmRtNetRC_t rc = kOkNetRC;
  355. cmRtNetSyncMsg_t m;
  356. m.label = NULL;
  357. assert( cmRtNetIsSyncModeMsg(data,dataByteCnt));
  358. if( _cmRtNetDeserializeSyncMsg(data,dataByteCnt,&m) != kOkNetRC )
  359. {
  360. rc = cmErrMsg(&p->err,rc,"Net sync. receive failed due to deserialize fail.");
  361. goto errLabel;
  362. }
  363. assert( m.hdr.selId == kNetSyncSelRtId );
  364. // attempt to locate the remote node which sent the msg
  365. cmRtNetNode_t* np = _cmRtNetFindNodeFromSockAddr(p,fromAddr);
  366. switch( m.selId )
  367. {
  368. case kHelloSelNetId:
  369. case kNodeSelNetId:
  370. {
  371. // if the node already exists ...
  372. if( np != NULL )
  373. {
  374. // ... delete it because we are about to get new info. about it.
  375. if((rc = _cmRtNetReleaseNode(p,np )) != kOkNetRC )
  376. goto errLabel;
  377. }
  378. // create a node proxy to represent the remote node
  379. // (Note:m.id == remote node endpoint count (i.e. the count of endpoints expected for the remote node.))
  380. if(( rc = _cmRtNetCreateNode(p,m.label,NULL,0,fromAddr,0,m.id)) != kOkNetRC )
  381. goto errLabel;
  382. // send response
  383. switch( m.selId )
  384. {
  385. case kHelloSelNetId:
  386. _cmRtNetRpt(p,"rcv hello\n"); // reply with local node
  387. rc = _cmRtNetSendSyncMsg( p, np, kNodeSelNetId, NULL, p->localNode->endPtCnt );
  388. break;
  389. case kNodeSelNetId:
  390. _cmRtNetRpt(p,"rcv node\n");
  391. _cmRtNetSendEndpointReplyMsg( p, np ); // reply with first endpoint
  392. break;
  393. default:
  394. assert(0);
  395. }
  396. }
  397. break;
  398. case kEndpointAckSelNetId:
  399. case kDoneSelNetId:
  400. rc = _cmRtNetSendEndpointReplyMsg(p,np);
  401. break;
  402. case kEndpointSelNetId:
  403. {
  404. cmRtNetEnd_t* ep;
  405. // verify the remote node exists.
  406. if( np == NULL )
  407. {
  408. rc = cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with an endpoint receive was not found.");
  409. goto errLabel;
  410. }
  411. // attempt to find the end point
  412. if((ep = _cmRtNetFindNodeEnd(np,m.label)) != NULL )
  413. ep->id = m.id; // the endpoint was found update the endPtId
  414. else
  415. {
  416. // create a local proxy for the endpoint
  417. if((rc = _cmRtNetCreateEndpoint(p,np,m.label,m.id)) != kOkNetRC )
  418. goto errLabel;
  419. }
  420. // reply with a local endpoint or 'done' msg
  421. rc = _cmRtNetSendEndpointReplyMsg( p, np );
  422. }
  423. break;
  424. default:
  425. assert(0);
  426. break;
  427. }
  428. errLabel:
  429. cmMemFree((cmChar_t*)m.label);
  430. return rc;
  431. }
  432. void _cmRtNetRecv( void* cbArg, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
  433. {
  434. cmRtNet_t* p = (cmRtNet_t*)cbArg;
  435. if( cmRtNetIsSyncModeMsg(data,dataByteCnt))
  436. _cmRtNetSyncModeRecv(p,data,dataByteCnt,fromAddr);
  437. else
  438. p->cbFunc(p->cbArg,data,dataByteCnt,fromAddr);
  439. }
  440. cmRtNetRC_t cmRtNetRegisterLocalNode( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* ipAddr, cmUdpPort_t port )
  441. {
  442. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  443. cmRtNetRC_t rc;
  444. struct sockaddr_in sockaddr;
  445. // release the local node
  446. if( p->localNode != NULL )
  447. {
  448. _cmRtNetReleaseNode(p,p->localNode);
  449. p->localNode = NULL;
  450. }
  451. // if this is the local node then initialze the local socket
  452. if( cmUdpInit(p->udpH,port,kNonBlockingUdpFl | kBroadcastUdpFl,_cmRtNetRecv,p,NULL,0,p->udpRecvBufByteCnt,p->udpTimeOutMs) != kOkUdpRC )
  453. {
  454. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"The UDP port initialization failed.");
  455. goto errLabel;
  456. }
  457. // get the socket address
  458. if( cmUdpInitAddr(p->udpH, ipAddr, port, &sockaddr ) != kOkUdpRC )
  459. {
  460. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"IP::port to socket address conversion failed.");
  461. goto errLabel;
  462. }
  463. // create the local node
  464. if((rc = _cmRtNetCreateNode(p,nodeLabel, ipAddr, port, &sockaddr, kLocalNodeNetFl, 0)) != kOkNetRC )
  465. goto errLabel;
  466. // the last created node is always the first node on the list
  467. p->localNode = p->nodes;
  468. // begin listening on the local port
  469. if( cmUdpEnableListen(p->udpH, true ) != kOkUdpRC )
  470. {
  471. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"The UDP port failed to enter 'listen' mode.");
  472. goto errLabel;
  473. }
  474. errLabel:
  475. return rc;
  476. }
  477. cmRtNetRC_t cmRtNetRegisterEndPoint( cmRtNetH_t h, const cmChar_t* endPtLabel, unsigned endPtId )
  478. {
  479. cmRtNetRC_t rc = kOkNetRC;
  480. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  481. if( p->localNode == NULL )
  482. return cmErrMsg(&p->err,kLocalNodeNetRC,"Local endpoints may not be added if a local node has not been defined.");
  483. if((rc = _cmRtNetCreateEndpoint(p, p->localNode,endPtLabel,endPtId )) == kOkNetRC )
  484. p->localNode->endPtCnt += 1;
  485. return rc;
  486. }
  487. cmRtNetRC_t cmRtNetClearAll( cmRtNetH_t h )
  488. {
  489. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  490. _cmRtNetReleaseNodes(p);
  491. return kOkNetRC;
  492. }
  493. cmRtNetRC_t cmRtNetBeginSyncMode( cmRtNetH_t h )
  494. {
  495. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  496. // broadcast 'node' msg
  497. return _cmRtNetSendSyncMsg( p, p->localNode, kHelloSelNetId, NULL, p->localNode->endPtCnt );
  498. }
  499. cmRtNetRC_t cmRtNetReceive( cmRtNetH_t h )
  500. {
  501. cmRtNetRC_t rc = kOkNetRC;
  502. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  503. if( cmUdpGetAvailData(p->udpH, NULL, NULL, NULL ) != kOkUdpRC )
  504. {
  505. cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP port query failed.");
  506. goto errLabel;
  507. }
  508. errLabel:
  509. return rc;
  510. }
  511. bool cmRtNetIsSyncModeMsg( const void* data, unsigned dataByteCnt )
  512. {
  513. cmRtNetSyncMsg_t* m = (cmRtNetSyncMsg_t*)data;
  514. return dataByteCnt >= sizeof(cmRtNetSyncMsg_t) && m->hdr.selId == kNetSyncSelRtId;
  515. }
  516. unsigned cmRtNetEndPointIndex( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* endPtLabel )
  517. {
  518. //cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  519. return cmInvalidIdx;
  520. }
  521. cmRtNetRC_t cmRtNetSend( cmRtNetH_t h, unsigned endPointIndex, const void* msg, unsigned msgByteCnt )
  522. {
  523. cmRtNetRC_t rc = kOkNetRC;
  524. //cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  525. return rc;
  526. }
  527. void cmRtNetReport( cmRtNetH_t h )
  528. {
  529. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  530. cmRpt_t* rpt = p->err.rpt;
  531. cmRtNetNode_t* np = p->nodes;
  532. for(; np!=NULL; np=np->link)
  533. {
  534. cmRptPrintf(rpt,"Node: %s ",np->label);
  535. if( np->addr != NULL )
  536. cmRptPrintf(rpt,"%s ",np->addr );
  537. if( cmIsFlag(np->flags,kLocalNodeNetFl) )
  538. cmRptPrintf(rpt,"LOCAL ");
  539. cmRptPrintf(rpt,"%s ",cmStringNullGuard(cmUdpAddrToString(p->udpH,&np->sockaddr)));
  540. if( np->port != cmInvalidId )
  541. cmRptPrintf(rpt,"%i ",np->port );
  542. cmRptPrintf(rpt,"\n");
  543. cmRtNetEnd_t* ep = np->ends;
  544. for(; ep!=NULL; ep=ep->link)
  545. {
  546. cmRptPrintf(rpt," endpt: %i %s\n",ep->id,cmStringNullGuard(ep->label ));
  547. }
  548. }
  549. }
  550. //==========================================================================
  551. #include "cmThread.h"
  552. typedef struct
  553. {
  554. cmThreadH_t thH;
  555. cmRtNetH_t netH;
  556. } _cmRtNetTest_t;
  557. void _cmRtNetTestRecv( void* cbArg, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
  558. {
  559. //_cmRtNetTest_t* p = (_cmRtNetTest_t*)cbArg;
  560. }
  561. bool _cmRtNetTestThreadFunc(void* param)
  562. {
  563. _cmRtNetTest_t* p = (_cmRtNetTest_t*)param;
  564. if( cmRtNetIsValid(p->netH) )
  565. {
  566. cmRtNetReceive(p->netH);
  567. }
  568. cmSleepMs(40);
  569. return true;
  570. }
  571. void cmRtNetTest( cmCtx_t* ctx, bool mstrFl )
  572. {
  573. char c;
  574. _cmRtNetTest_t t;
  575. const cmChar_t* hostNameStr;
  576. cmUdpPort_t port = 5876;
  577. _cmRtNetTest_t* p = &t;
  578. cmRtNetRC_t rc = kOkNetRC;
  579. memset(&t,0,sizeof(t));
  580. if( cmThreadCreate(&p->thH,_cmRtNetTestThreadFunc,p,&ctx->rpt) != kOkThRC )
  581. goto errLabel;
  582. if((rc = cmRtNetAlloc(ctx,&p->netH,_cmRtNetTestRecv,p)) != kOkNetRC )
  583. goto errLabel;
  584. hostNameStr = cmRtNetLocalHostName(p->netH);
  585. if( hostNameStr == NULL )
  586. hostNameStr = "<no-host-name>";
  587. if((rc = cmRtNetRegisterLocalNode(p->netH, hostNameStr, NULL, port )) != kOkNetRC)
  588. goto errLabel;
  589. if( mstrFl )
  590. {
  591. if((rc = cmRtNetRegisterEndPoint(p->netH,"thunk_ep0", 0 )) != kOkNetRC )
  592. goto errLabel;
  593. if(( rc = cmRtNetBeginSyncMode(p->netH)) != kOkNetRC )
  594. goto errLabel;
  595. }
  596. else
  597. {
  598. if((rc = cmRtNetRegisterEndPoint(p->netH,"whirl_ep0", 0 )) != kOkNetRC )
  599. goto errLabel;
  600. }
  601. if( cmThreadPause(p->thH,0) != kOkThRC )
  602. goto errLabel;
  603. cmRptPrintf(&ctx->rpt,"%s q=quit\n", mstrFl ? "Master: " : "Slave: ");
  604. while( (c=getchar()) != 'q' )
  605. {
  606. switch(c)
  607. {
  608. case 'r':
  609. cmRtNetReport(p->netH);
  610. break;
  611. }
  612. }
  613. errLabel:
  614. cmThreadDestroy(&p->thH);
  615. cmRtNetFree(&p->netH);
  616. return;
  617. }