开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 多媒体 > 流媒体/Mpeg4/MP4 > 查看源码
mbus.c
资源名称：NETVIDEO.rar　[点击查看]
上传用户：sun1608
上传日期：2007-02-02
资源大小：6116k
文件大小：30k
源码类别：

流媒体/Mpeg4/MP4

开发平台：

Visual C++

mbus.c：源码内容
  1. /*
  2.  * FILE:     mbus.c
  3.  * AUTHOR:   Colin Perkins
  4.  * MODIFIED: Orion Hodson
  5.  *           Markus Germeier
  6.  * 
  7.  * Copyright (c) 1997-2000 University College London
  8.  * All rights reserved.
  9.  *
  10.  * Redistribution and use in source and binary forms, with or without
  11.  * modification, is permitted provided that the following conditions 
  12.  * are met:
  13.  * 1. Redistributions of source code must retain the above copyright
  14.  *    notice, this list of conditions and the following disclaimer.
  15.  * 2. Redistributions in binary form must reproduce the above copyright
  16.  *    notice, this list of conditions and the following disclaimer in the
  17.  *    documentation and/or other materials provided with the distribution.
  18.  * 3. All advertising materials mentioning features or use of this software
  19.  *    must display the following acknowledgement:
  20.  *      This product includes software developed by the Computer Science
  21.  *      Department at University College London
  22.  * 4. Neither the name of the University nor of the Department may be used
  23.  *    to endorse or promote products derived from this software without
  24.  *    specific prior written permission.
  25.  * THIS SOFTWARE IS PROVIDED BY THE AUTHORS AND CONTRIBUTORS ``AS IS'' AND
  26.  * ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  27.  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  28.  * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE
  29.  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  30.  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  31.  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  32.  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  33.  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  34.  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  35.  * SUCH DAMAGE.
  36.  */
  38. #include "config_unix.h"
  39. #include "config_win32.h"
  40. #include "debug.h"
  41. #include "memory.h"
  42. #include "net_udp.h"
  43. #include "hmac.h"
  44. #include "qfDES.h"
  45. #include "base64.h"
  46. #include "gettimeofday.h"
  47. #include "vsnprintf.h"
  48. #include "mbus.h"
  49. #include "mbus_config.h"
  50. #include "mbus_parser.h"
  51. #include "mbus_addr.h"
  53. #define MBUS_BUF_SIZE   1500
  54. #define MBUS_ACK_BUF_SIZE 1500
  55. #define MBUS_MAX_ADDR     10
  56. #define MBUS_MAX_QLEN     50 /* Number of messages we can queue with mbus_qmsg() */
  58. #define MBUS_MAGIC 0x87654321
  59. #define MBUS_MSG_MAGIC 0x12345678
  61. struct mbus_msg {
  62. struct mbus_msg *next;
  63. struct timeval  send_time; /* Time the message was sent, to trigger a retransmit */
  64. struct timeval  comp_time; /* Time the message was composed, the timestamp in the packet header */
  65. char *dest;
  66. int  reliable;
  67. int  complete; /* Indicates that we've finished adding cmds to this message */
  68. int  seqnum;
  69. int  retransmit_count;
  70. int  message_size;
  71. int  num_cmds;
  72. char *cmd_list[MBUS_MAX_QLEN];
  73. char *arg_list[MBUS_MAX_QLEN];
  74. uint32_t  idx_list[MBUS_MAX_QLEN];
  75. uint32_t  magic; /* For debugging... */
  76. };
  78. struct mbus {
  79. socket_udp    *s;
  80. char    *addr; /* Addresses we respond to.  */
  81. int     max_other_addr;
  82. int     num_other_addr;
  83. char **other_addr; /* Addresses of other entities on the mbus.  */
  84.         struct timeval          **other_hello;                  /* Time of last mbus.hello we received from other entities      */
  85. int     seqnum;
  86. struct mbus_msg    *cmd_queue; /* Queue of messages waiting to be sent */
  87. struct mbus_msg    *waiting_ack; /* The last reliable message sent, if we have not yet got the ACK */
  88. char    *hashkey;
  89. int     hashkeylen;
  90. char    *encrkey;
  91. int     encrkeylen;
  92. struct timeval     last_heartbeat; /* Last time we sent a heartbeat message */
  93. struct mbus_config  *cfg;
  94. void (*cmd_handler)(char *src, char *cmd, char *arg, void *dat);
  95. void (*err_handler)(int seqnum, int reason);
  96. uint32_t   magic; /* For debugging...                                             */
  97. uint32_t   index;
  98. uint32_t   index_sent;
  99. };
  101. static void mbus_validate(struct mbus *m)
  102. {
  103. #ifdef DEBUG
  104. int i;
  106. ASSERT(m->num_other_addr <= m->max_other_addr);
  107. ASSERT(m->num_other_addr >= 0);
  108. for (i = 0; i < m->num_other_addr; i++) {
  109. ASSERT(m->other_addr[i]  != NULL);
  110. ASSERT(m->other_hello[i] != NULL);
  111. }
  112. for (i = m->num_other_addr + 1; i < m->max_other_addr; i++) {
  113. ASSERT(m->other_addr[i]  == NULL);
  114. ASSERT(m->other_hello[i] == NULL);
  115. }
  116. #endif
  117. ASSERT(m->magic == MBUS_MAGIC);
  118. xmemchk();
  119. }
  121. static void mbus_msg_validate(struct mbus_msg *m)
  122. {
  123. #ifdef DEBUG
  124. int i;
  126. ASSERT((m->num_cmds < MBUS_MAX_QLEN) && (m->num_cmds >= 0));
  127. for (i = 0; i < m->num_cmds; i++) {
  128. ASSERT(m->cmd_list[i] != NULL);
  129. ASSERT(m->arg_list[i] != NULL);
  130. if (i > 0) {
  131. ASSERT(m->idx_list[i] > m->idx_list[i-1]);
  132. }
  133. }
  134. for (i = m->num_cmds + 1; i < MBUS_MAX_QLEN; i++) {
  135. ASSERT(m->cmd_list[i] == NULL);
  136. ASSERT(m->arg_list[i] == NULL);
  137. }
  138. ASSERT(m->dest != NULL);
  139. #endif
  140. ASSERT(m->magic == MBUS_MSG_MAGIC);
  141. }
  143. static void store_other_addr(struct mbus *m, char *a)
  144. {
  145. /* This takes the address a and ensures it is stored in the   */
  146. /* m->other_addr field of the mbus structure. The other_addr  */
  147. /* field should probably be a hash table, but for now we hope */
  148. /* that there are not too many entities on the mbus, so the   */
  149. /* list is small.                                             */
  150. int i;
  152. mbus_validate(m);
  154. for (i = 0; i < m->num_other_addr; i++) {
  155. if (mbus_addr_match(m->other_addr[i], a)) {
  156. /* Already in the list... */
  157. gettimeofday(m->other_hello[i],NULL);
  158. return;
  159. }
  160. }
  162. if (m->num_other_addr == m->max_other_addr) {
  163. /* Expand the list... */
  164. m->max_other_addr *= 2;
  165. m->other_addr = (char **) xrealloc(m->other_addr, m->max_other_addr * sizeof(char *));
  166. m->other_hello = (struct timeval **) xrealloc(m->other_hello, m->max_other_addr * sizeof(struct timeval *));
  167. }
  168. m->other_hello[m->num_other_addr]=(struct timeval *)xmalloc(sizeof(struct timeval));
  169. gettimeofday(m->other_hello[m->num_other_addr],NULL);
  170. m->other_addr[m->num_other_addr++] = xstrdup(a);
  171. }
  173. static void remove_other_addr(struct mbus *m, char *a)
  174. {
  175. /* Removes the address a from the m->other_addr field of the */
  176. /* mbus structure.                                           */
  177. int i, j;
  179. mbus_validate(m);
  181. for (i = 0; i < m->num_other_addr; i++) {
  182. if (mbus_addr_match(m->other_addr[i], a)) {
  183. xfree(m->other_addr[i]);
  184. xfree(m->other_hello[i]);
  185. for (j = i+1; j < m->num_other_addr; j++) {
  186. m->other_addr[j-1] = m->other_addr[j];
  187. m->other_hello[j-1] = m->other_hello[j];
  188. }
  189. m->other_addr[m->num_other_addr  - 1] = NULL;
  190. m->other_hello[m->num_other_addr - 1] = NULL;
  191. m->num_other_addr--;
  192. }
  193. }
  194. }
  196. static void remove_inactiv_other_addr(struct mbus *m, struct timeval t, int interval){
  197. /* Remove addresses we haven't heard from for about 5 * interval */
  198. /* Count backwards so it is safe to remove entries               */
  199. int i;
  200.     
  201. mbus_validate(m);
  203. for (i=m->num_other_addr-1; i>=0; i--){
  204. if ((t.tv_sec-(m->other_hello[i]->tv_sec)) > 5 * interval) {
  205. debug_msg("remove dead entity (%s)n", m->other_addr[i]);
  206. remove_other_addr(m, m->other_addr[i]);
  207. }
  208. }
  209. }
  211. int mbus_addr_valid(struct mbus *m, char *addr)
  212. {
  213. int i;
  215. mbus_validate(m);
  217. for (i = 0; i < m->num_other_addr; i++) {
  218. if (mbus_addr_match(m->other_addr[i], addr)) {
  219. return TRUE;
  220. }
  221. }
  222. return FALSE;
  223. }
  225. static int mbus_addr_unique(struct mbus *m, char *addr)
  226. {
  227. int     i, n = 0;
  229. mbus_validate(m);
  231. for (i = 0; i < m->num_other_addr; i++) {
  232. if (mbus_addr_match(m->other_addr[i], addr)) {
  233. n++;
  234. }
  235. }
  236. return n==1;
  237. }
  239. /* The mb_* functions are used to build an mbus message up in the */
  240. /* mb_buffer, and to add authentication and encryption before the */
  241. /* message is sent.                                               */
  242. char  mb_cryptbuf[MBUS_BUF_SIZE];
  243. char *mb_buffer;
  244. char *mb_bufpos;
  246. #define MBUS_AUTH_LEN 16
  248. static void mb_header(int seqnum, int ts, char reliable, const char *src, const char *dst, int ackseq)
  249. {
  250. xmemchk();
  251. mb_buffer   = (char *) xmalloc(MBUS_BUF_SIZE + 1);
  252. memset(mb_buffer,   0, MBUS_BUF_SIZE);
  253. memset(mb_buffer, ' ', MBUS_AUTH_LEN);
  254. mb_bufpos = mb_buffer + MBUS_AUTH_LEN;
  255. sprintf(mb_bufpos, "nmbus/1.0 %6d %9d %c (%s) %s ", seqnum, ts, reliable, src, dst);
  256. mb_bufpos += 33 + strlen(src) + strlen(dst);
  257. if (ackseq == -1) {
  258. sprintf(mb_bufpos, "()n");
  259. mb_bufpos += 3;
  260. } else {
  261. sprintf(mb_bufpos, "(%6d)n", ackseq);
  262. mb_bufpos += 9;
  263. }
  264. }
  266. static void mb_add_command(const char *cmnd, const char *args)
  267. {
  268. int offset = strlen(cmnd) + strlen(args) + 5;
  270. ASSERT((mb_bufpos + offset - mb_buffer) < MBUS_BUF_SIZE);
  272. sprintf(mb_bufpos, "%s (%s)n", cmnd, args);
  273. mb_bufpos += offset - 1; /* The -1 in offset means we're not NUL terminated - fix in mb_send */
  274. }
  276. static void mb_send(struct mbus *m)
  277. {
  278. char digest[16];
  279. int len;
  280. unsigned char initVec[8] = {0,0,0,0,0,0,0,0};
  281.  
  282. mbus_validate(m);
  284. *(mb_bufpos++) = '';
  285. ASSERT((mb_bufpos - mb_buffer) < MBUS_BUF_SIZE);
  286. ASSERT(strlen(mb_buffer) < MBUS_BUF_SIZE);
  288. /* Pad to a multiple of 8 bytes, so the encryption can work... */
  289. while (((mb_bufpos - mb_buffer) % 8) != 0) {
  290. *(mb_bufpos++) = '';
  291. }
  292. len = mb_bufpos - mb_buffer;
  293. ASSERT(len < MBUS_BUF_SIZE);
  294. ASSERT(strlen(mb_buffer) < MBUS_BUF_SIZE);
  296. xmemchk();
  297. if (m->hashkey != NULL) {
  298. /* Authenticate... */
  299. hmac_md5(mb_buffer + MBUS_AUTH_LEN+1, strlen(mb_buffer) - (MBUS_AUTH_LEN+1), m->hashkey, m->hashkeylen, digest);
  300. base64encode(digest, 12, mb_buffer, MBUS_AUTH_LEN);
  301. }
  302. xmemchk();
  303. if (m->encrkey != NULL) {
  304. /* Encrypt... */
  305. memset(mb_cryptbuf, 0, MBUS_BUF_SIZE);
  306. memcpy(mb_cryptbuf, mb_buffer, len);
  307. ASSERT((len % 8) == 0);
  308. ASSERT(len < MBUS_BUF_SIZE);
  309. ASSERT(m->encrkeylen == 8);
  310. xmemchk();
  311. qfDES_CBC_e(m->encrkey, mb_cryptbuf, len, initVec);
  312. xmemchk();
  313. memcpy(mb_buffer, mb_cryptbuf, len);
  314. }
  315. xmemchk();
  316. udp_send(m->s, mb_buffer, len);
  317. xfree(mb_buffer);
  318. }
  320. static void resend(struct mbus *m, struct mbus_msg *curr) 
  321. {
  322. /* Don't need to check for buffer overflows: this was done in mbus_send() when */
  323. /* this message was first transmitted. If it was okay then, it's okay now.     */
  324. int  i;
  326. mbus_validate(m);
  328. mb_header(curr->seqnum, curr->comp_time.tv_sec, (char)(curr->reliable?'R':'U'), m->addr, curr->dest, -1);
  329. for (i = 0; i < curr->num_cmds; i++) {
  330. mb_add_command(curr->cmd_list[i], curr->arg_list[i]);
  331. }
  332. mb_send(m);
  333. curr->retransmit_count++;
  334. }
  336. void mbus_retransmit(struct mbus *m)
  337. {
  338. struct mbus_msg *curr = m->waiting_ack;
  339. struct timeval time;
  340. long diff;
  342. mbus_validate(m);
  344. if (!mbus_waiting_ack(m)) {
  345. return;
  346. }
  348. mbus_msg_validate(curr);
  350. gettimeofday(&time, NULL);
  352. /* diff is time in milliseconds that the message has been awaiting an ACK */
  353. diff = ((time.tv_sec * 1000) + (time.tv_usec / 1000)) - ((curr->send_time.tv_sec * 1000) + (curr->send_time.tv_usec / 1000));
  354. if (diff > 10000) {
  355. debug_msg("Reliable mbus message failed!n");
  356. if (m->err_handler == NULL) {
  357. abort();
  358. }
  359. m->err_handler(curr->seqnum, MBUS_MESSAGE_LOST);
  360. /* if we don't delete this failed message, the error handler
  361.                    gets triggered every time we call mbus_retransmit */
  362. while (m->waiting_ack->num_cmds > 0) {
  363.     m->waiting_ack->num_cmds--;
  364.     xfree(m->waiting_ack->cmd_list[m->waiting_ack->num_cmds]);
  365.     xfree(m->waiting_ack->arg_list[m->waiting_ack->num_cmds]);
  366. }
  367. xfree(m->waiting_ack->dest);
  368. xfree(m->waiting_ack);
  369. m->waiting_ack = NULL;
  370. return;
  371. } 
  372. /* Note: We only send one retransmission each time, to avoid
  373.  * overflowing the receiver with a burst of requests...
  374.  */
  375. if ((diff > 750) && (curr->retransmit_count == 2)) {
  376. resend(m, curr);
  377. return;
  378. } 
  379. if ((diff > 500) && (curr->retransmit_count == 1)) {
  380. resend(m, curr);
  381. return;
  382. } 
  383. if ((diff > 250) && (curr->retransmit_count == 0)) {
  384. resend(m, curr);
  385. return;
  386. }
  387. }
  389. void mbus_heartbeat(struct mbus *m, int interval)
  390. {
  391. struct timeval curr_time;
  392. char *a = (char *) xmalloc(3);
  393. sprintf(a, "()");
  395. mbus_validate(m);
  397. gettimeofday(&curr_time, NULL);
  398. if (curr_time.tv_sec - m->last_heartbeat.tv_sec >= interval) {
  399. mb_header(++m->seqnum, (int) curr_time.tv_sec, 'U', m->addr, "()", -1);
  400. mb_add_command("mbus.hello", "");
  401. mb_send(m);
  403. m->last_heartbeat = curr_time;
  404. /* Remove dead sources */
  405. remove_inactiv_other_addr(m, curr_time, interval);
  406. }
  407. xfree(a);
  408. }
  410. int mbus_waiting_ack(struct mbus *m)
  411. {
  412. mbus_validate(m);
  413. return m->waiting_ack != NULL;
  414. }
  416. int mbus_sent_all(struct mbus *m)
  417. {
  418. mbus_validate(m);
  419. return (m->cmd_queue == NULL) && (m->waiting_ack == NULL);
  420. }
  422. struct mbus *mbus_init(void  (*cmd_handler)(char *src, char *cmd, char *arg, void *dat), 
  423.        void  (*err_handler)(int seqnum, int reason),
  424.        char  *addr)
  425. {
  426. struct mbus *m;
  427. struct mbus_key    k;
  428. struct mbus_parser *mp;
  429. int    i;
  430. char             *net_addr, *tmp;
  431. uint16_t           net_port;
  432. int                net_scope;
  434. m = (struct mbus *) xmalloc(sizeof(struct mbus));
  435. if (m == NULL) {
  436. debug_msg("Unable to allocate memory for mbusn");
  437. return NULL;
  438. }
  440. m->cfg = mbus_create_config();
  441. mbus_lock_config_file(m->cfg);
  442. net_addr = (char *) xmalloc(20);
  443. mbus_get_net_addr(m->cfg, net_addr, &net_port, &net_scope);
  444. m->s   = udp_init(net_addr, net_port, net_port, net_scope);
  445.         if (m->s == NULL) {
  446.                 debug_msg("Unable to initialize mbus addressn");
  447.                 xfree(m);
  448.                 return NULL;
  449.         }
  450. m->seqnum         = 0;
  451. m->cmd_handler    = cmd_handler;
  452. m->err_handler   = err_handler;
  453. m->num_other_addr = 0;
  454. m->max_other_addr = 10;
  455. m->other_addr     = (char **) xmalloc(sizeof(char *) * 10);
  456. m->other_hello    = (struct timeval **) xmalloc(sizeof(struct timeval *) * 10);
  457. for (i = 0; i < 10; i++) {
  458. m->other_addr[i]  = NULL;
  459. m->other_hello[i] = NULL;
  460. }
  461. m->cmd_queue   = NULL;
  462. m->waiting_ack   = NULL;
  463. m->magic          = MBUS_MAGIC;
  464. m->index          = 0;
  465. m->index_sent     = 0;
  467. mp = mbus_parse_init(xstrdup(addr));
  468. if (!mbus_parse_lst(mp, &tmp)) {
  469. debug_msg("Invalid mbus addressn");
  470. abort();
  471. }
  472. m->addr = xstrdup(tmp);
  473. mbus_parse_done(mp);
  474. ASSERT(m->addr != NULL);
  476. gettimeofday(&(m->last_heartbeat), NULL);
  478. mbus_get_encrkey(m->cfg, &k);
  479. m->encrkey    = k.key;
  480. m->encrkeylen = k.key_len;
  482. mbus_get_hashkey(m->cfg, &k);
  483. m->hashkey    = k.key;
  484. m->hashkeylen = k.key_len;
  486. mbus_unlock_config_file(m->cfg);
  488. xfree(net_addr);
  490. return m;
  491. }
  493. void mbus_cmd_handler(struct mbus *m, void  (*cmd_handler)(char *src, char *cmd, char *arg, void *dat))
  494. {
  495. mbus_validate(m);
  496. m->cmd_handler = cmd_handler;
  497. }
  499. static void mbus_flush_msgs(struct mbus_msg **queue)
  500. {
  501.         struct mbus_msg *curr, *next;
  502.         int i;
  504.         curr = *queue;
  505.         while(curr) {
  506.                 next = curr->next;
  507.                 xfree(curr->dest);
  508.                 for(i = 0; i < curr->num_cmds; i++) {
  509.                         xfree(curr->cmd_list[i]);
  510.                         xfree(curr->arg_list[i]);
  511.                 }
  512. xfree(curr);
  513.                 curr = next;
  514.         }
  515. *queue = NULL;
  516. }
  518. void mbus_exit(struct mbus *m) 
  519. {
  520.         int i;
  522.         ASSERT(m != NULL);
  523. mbus_validate(m);
  525. mbus_qmsg(m, "()", "mbus.bye", "", FALSE);
  526. mbus_send(m);
  528. /* FIXME: It should be a fatal error to call mbus_exit() if some messages are still outstanding. */
  529. /*        We will need an mbus_flush() call first though, to ensure nothing is waiting.          */
  530.         mbus_flush_msgs(&m->cmd_queue);
  531.         mbus_flush_msgs(&m->waiting_ack);
  533.         if (m->encrkey != NULL) {
  534.                 xfree(m->encrkey);
  535.         }
  536.         if (m->hashkey != NULL) {
  537.          xfree(m->hashkey);
  538. }
  540.         udp_exit(m->s);
  542. /* Clean up other_* */
  543. for (i=m->num_other_addr-1; i>=0; i--){
  544.     remove_other_addr(m, m->other_addr[i]);
  545. }
  547.         xfree(m->addr);
  548. xfree(m->other_addr);
  549. xfree(m->other_hello);
  550. xfree(m->cfg);
  551.         xfree(m);
  552. }
  554. void mbus_send(struct mbus *m)
  555. {
  556. /* Send one, or more, messages previosly queued with mbus_qmsg(). */
  557. /* Messages for the same destination are batched together. Stops  */
  558. /* when a reliable message is sent, until the ACK is received.    */
  559. struct mbus_msg *curr = m->cmd_queue;
  560. int  i;
  562. mbus_validate(m);
  563. if (m->waiting_ack != NULL) {
  564. return;
  565. }
  567. while (curr != NULL) {
  568. mbus_msg_validate(curr);
  569. /* It's okay for us to send messages which haven't been marked as complete - */
  570. /* that just means we're sending something which has the potential to have   */
  571. /* more data piggybacked. However, if it's not complete it MUST be the last  */
  572. /* in the list, or something has been reordered - which is bad.              */
  573. if (!curr->complete) {
  574. ASSERT(curr->next == NULL);
  575. }
  577. if (curr->reliable) {
  578.         if (!mbus_addr_valid(m, curr->dest)) {
  579.     debug_msg("Trying to send reliably to an unknown address...n");
  580.     if (m->err_handler == NULL) {
  581. abort();
  582.     }
  583.     m->err_handler(curr->seqnum, MBUS_DESTINATION_UNKNOWN);
  584. }
  585.         if (!mbus_addr_unique(m, curr->dest)) {
  586.     debug_msg("Trying to send reliably but address is not unique...n");
  587.     if (m->err_handler == NULL) {
  588. abort();
  589.     }
  590.     m->err_handler(curr->seqnum, MBUS_DESTINATION_NOT_UNIQUE);
  591. }
  592. }
  593. /* Create the message... */
  594. mb_header(curr->seqnum, curr->comp_time.tv_sec, (char)(curr->reliable?'R':'U'), m->addr, curr->dest, -1);
  595. for (i = 0; i < curr->num_cmds; i++) {
  596. ASSERT(m->index_sent == (curr->idx_list[i] - 1));
  597. m->index_sent = curr->idx_list[i];
  598. mb_add_command(curr->cmd_list[i], curr->arg_list[i]);
  599. }
  600. mb_send(m);
  602. m->cmd_queue = curr->next;
  603. if (curr->reliable) {
  604. /* Reliable message, wait for the ack... */
  605. gettimeofday(&(curr->send_time), NULL);
  606. m->waiting_ack = curr;
  607. curr->next = NULL;
  608. return;
  609. } else {
  610. while (curr->num_cmds > 0) {
  611. curr->num_cmds--;
  612. xfree(curr->cmd_list[curr->num_cmds]); curr->cmd_list[curr->num_cmds] = NULL;
  613. xfree(curr->arg_list[curr->num_cmds]); curr->arg_list[curr->num_cmds] = NULL;
  614. }
  615. xfree(curr->dest);
  616. xfree(curr);
  617. }
  618. curr = m->cmd_queue;
  619. }
  620. }
  622. void mbus_qmsg(struct mbus *m, const char *dest, const char *cmnd, const char *args, int reliable)
  623. {
  624. /* Queue up a message for sending. The message is not */
  625. /* actually sent until mbus_send() is called.         */
  626. struct mbus_msg *curr = m->cmd_queue;
  627. struct mbus_msg *prev = NULL;
  628. int  alen = strlen(cmnd) + strlen(args) + 4;
  629. int  i;
  631. mbus_validate(m);
  632. while (curr != NULL) {
  633. mbus_msg_validate(curr);
  634. if (!curr->complete) {
  635. /* This message is still open for new commands. It MUST be the last in the */
  636. /* cmd_queue, else commands will be reordered.                             */
  637. ASSERT(curr->next == NULL);
  638. if (mbus_addr_identical(curr->dest, dest) &&
  639.             (curr->num_cmds < MBUS_MAX_QLEN) && ((curr->message_size + alen) < (MBUS_BUF_SIZE - 500))) {
  640. curr->num_cmds++;
  641. curr->reliable |= reliable;
  642. curr->cmd_list[curr->num_cmds-1] = xstrdup(cmnd);
  643. curr->arg_list[curr->num_cmds-1] = xstrdup(args);
  644. curr->idx_list[curr->num_cmds-1] = ++(m->index);
  645. curr->message_size += alen;
  646. mbus_msg_validate(curr);
  647. return;
  648. } else {
  649. curr->complete = TRUE;
  650. }
  651. }
  652. prev = curr;
  653. curr = curr->next;
  654. }
  655. /* If we get here, we've not found an open message in the cmd_queue.  We */
  656. /* have to create a new message, and add it to the end of the cmd_queue. */
  657. curr = (struct mbus_msg *) xmalloc(sizeof(struct mbus_msg));
  658. curr->magic            = MBUS_MSG_MAGIC;
  659. curr->next             = NULL;
  660. curr->dest             = xstrdup(dest);
  661. curr->retransmit_count = 0;
  662. curr->message_size     = alen + 60 + strlen(dest) + strlen(m->addr);
  663. curr->seqnum           = ++m->seqnum;
  664. curr->reliable         = reliable;
  665. curr->complete         = FALSE;
  666. curr->num_cmds         = 1;
  667. curr->cmd_list[0]      = xstrdup(cmnd);
  668. curr->arg_list[0]      = xstrdup(args);
  669. curr->idx_list[curr->num_cmds-1] = ++(m->index);
  670. for (i = 1; i < MBUS_MAX_QLEN; i++) {
  671. curr->cmd_list[i] = NULL;
  672. curr->arg_list[i] = NULL;
  673. }
  674. if (prev == NULL) {
  675. m->cmd_queue = curr;
  676. } else {
  677. prev->next = curr;
  678. }
  679. gettimeofday(&(curr->send_time), NULL);
  680. gettimeofday(&(curr->comp_time), NULL);
  681. mbus_msg_validate(curr);
  682. }
  684. #define mbus_qmsgf(m, dest, reliable, cmnd, format, var) 
  685. { 
  686. char buffer[MBUS_BUF_SIZE]; 
  687. mbus_validate(m); 
  688. snprintf(buffer, MBUS_BUF_SIZE, format, var); 
  689. mbus_qmsg(m, dest, cmnd, buffer, reliable); 
  690. }
  692. #if 0
  693. void mbus_qmsgf(struct mbus *m, const char *dest, int reliable, const char *cmnd, const char *format, ...)
  694. {
  695. /* This is a wrapper around mbus_qmsg() which does a printf() style format into  */
  696. /* a buffer. Saves the caller from having to a a malloc(), write the args string */
  697. /* and then do a free(), and also saves worring about overflowing the buffer, so */
  698. /* removing a common source of bugs!                                             */
  699. char buffer[MBUS_BUF_SIZE];
  700. va_list ap;
  702. mbus_validate(m);
  703. va_start(ap, format);
  704. #ifdef WIN32
  705.         _vsnprintf(buffer, MBUS_BUF_SIZE, format, ap);
  706. #else
  707.         vsnprintf(buffer, MBUS_BUF_SIZE, format, ap);
  708. #endif
  709. va_end(ap);
  710. mbus_qmsg(m, dest, cmnd, buffer, reliable);
  711. }
  712. #endif
  714. int mbus_recv(struct mbus *m, void *data, struct timeval *timeout)
  715. {
  716. char *auth, *ver, *src, *dst, *ack, *r, *cmd, *param, *npos;
  717. char    buffer[MBUS_BUF_SIZE];
  718. int    buffer_len, seq, a, rx, ts, authlen, loop_count;
  719. char    ackbuf[MBUS_ACK_BUF_SIZE];
  720. char    digest[16];
  721. unsigned char  initVec[8] = {0,0,0,0,0,0,0,0};
  722. struct timeval  t;
  723. struct mbus_parser *mp, *mp2;
  725. mbus_validate(m);
  727. rx = FALSE;
  728. loop_count = 0;
  729. while (loop_count++ < 10) {
  730. memset(buffer, 0, MBUS_BUF_SIZE);
  731.                 ASSERT(m->s != NULL);
  732. udp_fd_zero();
  733. udp_fd_set(m->s);
  734. t.tv_sec  = timeout->tv_sec;
  735. t.tv_usec = timeout->tv_usec;
  736.                 if ((udp_select(&t) > 0) && udp_fd_isset(m->s)) {
  737. buffer_len = udp_recv(m->s, buffer, MBUS_BUF_SIZE);
  738. if (buffer_len > 0) {
  739. rx = TRUE;
  740. } else {
  741. return rx;
  742. }
  743. } else {
  744. return FALSE;
  745. }
  747. if (m->encrkey != NULL) {
  748. /* Decrypt the message... */
  749. if ((buffer_len % 8) != 0) {
  750. debug_msg("Encrypted message not a multiple of 8 bytes in lengthn");
  751. continue;
  752. }
  753. memcpy(mb_cryptbuf, buffer, buffer_len);
  754. memset(initVec, 0, 8);
  755. qfDES_CBC_d(m->encrkey, mb_cryptbuf, buffer_len, initVec);
  756. memcpy(buffer, mb_cryptbuf, buffer_len);
  757. }
  759. /* Sanity check that this is a vaguely sensible format message... Should prevent */
  760. /* problems if we're fed complete garbage, but won't prevent determined hackers. */
  761. if (strncmp(buffer + MBUS_AUTH_LEN + 1, "mbus/1.0", 8) != 0) {
  762. continue;
  763. }
  765. mp = mbus_parse_init(buffer);
  766. /* remove trailing 0 bytes */
  767. npos = (char *) strchr(buffer,'');
  768. if(npos!=NULL) {
  769. buffer_len=npos-buffer;
  770. }
  771. /* Parse the authentication header */
  772. if (!mbus_parse_sym(mp, &auth)) {
  773. debug_msg("Failed to parse authentication headern");
  774. mbus_parse_done(mp);
  775. continue;
  776. }
  778. /* Check that the packet authenticates correctly... */
  779. authlen = strlen(auth);
  780. hmac_md5(buffer + authlen + 1, buffer_len - authlen - 1, m->hashkey, m->hashkeylen, digest);
  781. base64encode(digest, 12, ackbuf, 16);
  782. if ((strlen(auth) != 16) || (strncmp(auth, ackbuf, 16) != 0)) {
  783. debug_msg("Failed to authenticate message...n");
  784. mbus_parse_done(mp);
  785. continue;
  786. }
  788. /* Parse the header */
  789. if (!mbus_parse_sym(mp, &ver)) {
  790. mbus_parse_done(mp);
  791. debug_msg("Parser failed version (1): %sn",ver);
  792. continue;
  793. }
  794. if (strcmp(ver, "mbus/1.0") != 0) {
  795. mbus_parse_done(mp);
  796. debug_msg("Parser failed version (2): %sn",ver);
  797. continue;
  798. }
  799. if (!mbus_parse_int(mp, &seq)) {
  800. mbus_parse_done(mp);
  801. debug_msg("Parser failed seqn");
  802. continue;
  803. }
  804. if (!mbus_parse_int(mp, &ts)) {
  805. mbus_parse_done(mp);
  806. debug_msg("Parser failed tsn");
  807. continue;
  808. }
  809. if (!mbus_parse_sym(mp, &r)) {
  810. mbus_parse_done(mp);
  811. debug_msg("Parser failed reliablen");
  812. continue;
  813. }
  814. if (!mbus_parse_lst(mp, &src)) {
  815. mbus_parse_done(mp);
  816. debug_msg("Parser failed srcn");
  817. continue;
  818. }
  819. if (!mbus_parse_lst(mp, &dst)) {
  820. mbus_parse_done(mp);
  821. debug_msg("Parser failed dstn");
  822. continue;
  823. }
  824. if (!mbus_parse_lst(mp, &ack)) {
  825. mbus_parse_done(mp);
  826. debug_msg("Parser failed ackn");
  827. continue;
  828. }
  830. store_other_addr(m, src);
  832. /* Check if the message was addressed to us... */
  833. if (mbus_addr_match(m->addr, dst)) {
  834. /* ...if so, process any ACKs received... */
  835. mp2 = mbus_parse_init(ack);
  836. while (mbus_parse_int(mp2, &a)) {
  837. if (mbus_waiting_ack(m)) {
  838. if (m->waiting_ack->seqnum == a) {
  839. while (m->waiting_ack->num_cmds > 0) {
  840. m->waiting_ack->num_cmds--;
  841. xfree(m->waiting_ack->cmd_list[m->waiting_ack->num_cmds]);
  842. xfree(m->waiting_ack->arg_list[m->waiting_ack->num_cmds]);
  843. }
  844. xfree(m->waiting_ack->dest);
  845. xfree(m->waiting_ack);
  846. m->waiting_ack = NULL;
  847. } else {
  848. debug_msg("Got ACK %d but wanted %dn", a, m->waiting_ack->seqnum);
  849. }
  850. } else {
  851. debug_msg("Got ACK %d but wasn't expecting itn", a);
  852. }
  853. }
  854. mbus_parse_done(mp2);
  855. /* ...if an ACK was requested, send one... */
  856. if (strcmp(r, "R") == 0) {
  857. char  *newsrc = (char *) xmalloc(strlen(src) + 3);
  858. struct timeval  t;
  860. sprintf(newsrc, "(%s)", src); /* Yes, this is a kludge. */
  861. gettimeofday(&t, NULL);
  862. mb_header(++m->seqnum, (int) t.tv_sec, 'U', m->addr, newsrc, seq);
  863. mb_send(m);
  864. xfree(newsrc);
  865. } else if (strcmp(r, "U") == 0) {
  866. /* Unreliable message.... not need to do anything */
  867. } else {
  868. debug_msg("Message with invalid reliability field "%s" ignoredn", r);
  869. }
  870. /* ...and process the commands contained in the message */
  871. while (mbus_parse_sym(mp, &cmd)) {
  872. if (mbus_parse_lst(mp, &param)) {
  873. char  *newsrc = (char *) xmalloc(strlen(src) + 3);
  874. sprintf(newsrc, "(%s)", src); /* Yes, this is a kludge. */
  875. /* Finally, we snoop on the message we just passed to the application, */
  876. /* to do housekeeping of our list of known mbus sources...             */
  877. if (strcmp(cmd, "mbus.bye") == 0) {
  878. remove_other_addr(m, newsrc);
  879. } 
  880. if (strcmp(cmd, "mbus.hello") == 0) {
  881. /* Mark this source as activ. We remove dead sources in mbus_heartbeat */
  882. store_other_addr(m, newsrc);
  883. }
  884. m->cmd_handler(newsrc, cmd, param, data);
  885. xfree(newsrc);
  886. } else {
  887. debug_msg("Unable to parse mbus command:n");
  888. debug_msg("cmd = %sn", cmd);
  889. debug_msg("arg = %sn", param);
  890. break;
  891. }
  892. }
  893. }
  894. mbus_parse_done(mp);
  895. }
  896. return rx;
  897. }
  899. #define RZ_HANDLE_WAITING 1
  900. #define RZ_HANDLE_GO      2
  902. struct mbus_rz {
  903. char *peer;
  904. char *token;
  905. struct mbus *m;
  906. void *data;
  907. int  mode;
  908. void (*cmd_handler)(char *src, char *cmd, char *args, void *data);
  909. };
  911. static void rz_handler(char *src, char *cmd, char *args, void *data)
  912. {
  913. struct mbus_rz *r = (struct mbus_rz *) data;
  914. struct mbus_parser *mp;
  916. if ((r->mode == RZ_HANDLE_WAITING) && (strcmp(cmd, "mbus.waiting") == 0)) {
  917. char *t;
  919. mp = mbus_parse_init(args);
  920. mbus_parse_str(mp, &t);
  921. if (strcmp(mbus_decode_str(t), r->token) == 0) {
  922.                         if (r->peer != NULL) xfree(r->peer);
  923. r->peer = xstrdup(src);
  924. }
  925. mbus_parse_done(mp);
  926. } else if ((r->mode == RZ_HANDLE_GO) && (strcmp(cmd, "mbus.go") == 0)) {
  927. char *t;
  929. mp = mbus_parse_init(args);
  930. mbus_parse_str(mp, &t);
  931. if (strcmp(mbus_decode_str(t), r->token) == 0) {
  932.                         if (r->peer != NULL) xfree(r->peer);
  933. r->peer = xstrdup(src);
  934. }
  935. mbus_parse_done(mp);
  936. } else {
  937. r->cmd_handler(src, cmd, args, r->data);
  938. }
  939. }
  941. char *mbus_rendezvous_waiting(struct mbus *m, char *addr, char *token, void *data)
  942. {
  943. /* Loop, sending mbus.waiting(token) to "addr", until we get mbus.go(token) */
  944. /* back from our peer. Any other mbus commands received whilst waiting are  */
  945. /* processed in the normal manner, as if mbus_recv() had been called.       */
  946. char *token_e, *peer;
  947. struct timeval  timeout;
  948. struct mbus_rz *r;
  950. mbus_validate(m);
  952. r = (struct mbus_rz *) xmalloc(sizeof(struct mbus_rz));
  953. r->peer        = NULL;
  954. r->token       = token;
  955. r->m           = m;
  956. r->data        = data;
  957. r->mode        = RZ_HANDLE_GO;
  958. r->cmd_handler = m->cmd_handler;
  959. m->cmd_handler = rz_handler;
  960. token_e        = mbus_encode_str(token);
  961. while (r->peer == NULL) {
  962. timeout.tv_sec  = 0;
  963. timeout.tv_usec = 100000;
  964. mbus_heartbeat(m, 1);
  965. mbus_qmsgf(m, addr, FALSE, "mbus.waiting", "%s", token_e);
  966. mbus_send(m);
  967. mbus_recv(m, r, &timeout);
  968. mbus_retransmit(m);
  969. }
  970. m->cmd_handler = r->cmd_handler;
  971. peer = r->peer;
  972. xfree(r);
  973. xfree(token_e);
  974. return peer;
  975. }
  977. char *mbus_rendezvous_go(struct mbus *m, char *token, void *data)
  978. {
  979. /* Wait until we receive mbus.waiting(token), then send mbus.go(token) back to   */
  980. /* the sender of that message. Whilst waiting, other mbus commands are processed */
  981. /* in the normal manner as if mbus_recv() had been called.                       */
  982. char *token_e, *peer;
  983. struct timeval  timeout;
  984. struct mbus_rz *r;
  986. mbus_validate(m);
  988. r = (struct mbus_rz *) xmalloc(sizeof(struct mbus_rz));
  989. r->peer        = NULL;
  990. r->token       = token;
  991. r->m           = m;
  992. r->data        = data;
  993. r->mode        = RZ_HANDLE_WAITING;
  994. r->cmd_handler = m->cmd_handler;
  995. m->cmd_handler = rz_handler;
  996. token_e        = mbus_encode_str(token);
  997. while (r->peer == NULL) {
  998. timeout.tv_sec  = 0;
  999. timeout.tv_usec = 100000;
  1000. mbus_heartbeat(m, 1);
  1001. mbus_send(m);
  1002. mbus_recv(m, r, &timeout);
  1003. mbus_retransmit(m);
  1004. }
  1006. mbus_qmsgf(m, r->peer, TRUE, "mbus.go", "%s", token_e);
  1007. do {
  1008. mbus_heartbeat(m, 1);
  1009. mbus_retransmit(m);
  1010. mbus_send(m);
  1011. timeout.tv_sec  = 0;
  1012. timeout.tv_usec = 100000;
  1013. mbus_recv(m, r, &timeout);
  1014. } while (!mbus_sent_all(m));
  1016. m->cmd_handler = r->cmd_handler;
  1017. peer = r->peer;
  1018. xfree(r);
  1019. xfree(token_e);
  1020. return peer;
  1021. }