luat_websocket.c 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690
  1. #include "luat_base.h"
  2. #include "luat_network_adapter.h"
  3. #include "luat_rtos.h"
  4. #include "luat_mem.h"
  5. #include "luat_websocket.h"
  6. #define LUAT_LOG_TAG "websocket"
  7. #include "luat_log.h"
  8. #define LUAT_USE_WEBSOCKET_DEBUG 0
  9. #if LUAT_USE_WEBSOCKET_DEBUG == 0
  10. #undef LLOGD
  11. #define LLOGD(...)
  12. #endif
  13. #if LUAT_USE_WEBSOCKET_DEBUG
  14. static void print_pkg(const char *tag, char *buff, luat_websocket_pkg_t *pkg)
  15. {
  16. if (pkg == NULL)
  17. {
  18. LLOGD("pkg is NULL");
  19. return;
  20. }
  21. LLOGD("%s pkg %02X%02X", tag, buff[0], buff[1]);
  22. LLOGD("%s pkg FIN %d R %d OPT %d MARK %d PLEN %d", tag, pkg->FIN, pkg->R, pkg->OPT_CODE, pkg->mark, pkg->plen);
  23. }
  24. #else
  25. #define print_pkg(...)
  26. #endif
  27. static int32_t luat_websocket_callback(void *data, void *param);
  28. int luat_websocket_payload(char *buf, luat_websocket_pkg_t *pkg, size_t limit)
  29. {
  30. uint32_t pkg_len = 0;
  31. // 先处理FIN
  32. pkg->FIN = (buf[0] >> 7) & 0x01;
  33. // 处理操作码
  34. pkg->OPT_CODE = buf[0] & 0xF;
  35. // 然后处理plen
  36. pkg->plen = buf[1] & 0x7F;
  37. print_pkg("downlink", buf, pkg);
  38. // websocket的payload长度支持3种情况:
  39. // 0字节(小于126,放在头部)
  40. // 2个字节 126 ~ 0xFFFF
  41. // 6个字节 0x1000 ~ 0xFFFFFFFF, 不打算支持.
  42. if (pkg->plen == 126)
  43. {
  44. if (limit < 4)
  45. {
  46. // 还缺1个字节,等吧
  47. LLOGD("wait more data offset, limit %d < 4", limit);
  48. return 0;
  49. }
  50. pkg->plen = (buf[2] & 0xFF) << 8;
  51. pkg->plen += (buf[3] & 0xFF);
  52. pkg_len = 4 + pkg->plen;
  53. pkg->payload = buf + 4;
  54. }
  55. else if (pkg->plen == 127)
  56. {
  57. // 后续还要8个字节,但这个包也太大了吧!!!
  58. LLOGE("websocket payload is too large!!!");
  59. return -1;
  60. }
  61. else
  62. {
  63. pkg_len = 2 + pkg->plen;
  64. pkg->payload = buf + 2;
  65. }
  66. LLOGD("payload size %ld pkg %ld", pkg->plen, pkg_len);
  67. if (pkg->plen > WEBSOCKET_PAYLOAD_MAX)
  68. {
  69. LLOGE("websocket payload is too large!!! %ld , must < %ld", pkg->plen, WEBSOCKET_RECV_BUF_LEN_MAX);
  70. return -1;
  71. }
  72. if (limit < pkg_len)
  73. {
  74. LLOGD("wait more data, limit %ld want %ld", limit, pkg_len);
  75. return 0;
  76. }
  77. return 1;
  78. }
  79. int luat_websocket_send_packet(void *socket_info, const void *buf, unsigned int count)
  80. {
  81. luat_websocket_ctrl_t *websocket_ctrl = (luat_websocket_ctrl_t *)socket_info;
  82. uint32_t tx_len = 0;
  83. int ret = network_tx(websocket_ctrl->netc, buf, count, 0, NULL, 0, &tx_len, 0);
  84. if (ret < 0)
  85. {
  86. LLOGI("network_tx %d , close socket", ret);
  87. luat_websocket_close_socket(websocket_ctrl);
  88. return 0;
  89. }
  90. return count;
  91. }
  92. void luat_websocket_ping(luat_websocket_ctrl_t *websocket_ctrl)
  93. {
  94. if (websocket_ctrl->websocket_state == 0)
  95. return;
  96. luat_websocket_pkg_t pkg = {
  97. .FIN = 1,
  98. .OPT_CODE = WebSocket_OP_PING,
  99. .plen = 0};
  100. luat_websocket_send_frame(websocket_ctrl, &pkg);
  101. }
  102. void luat_websocket_pong(luat_websocket_ctrl_t *websocket_ctrl)
  103. {
  104. luat_websocket_pkg_t pkg = {
  105. .FIN = 1,
  106. .OPT_CODE = WebSocket_OP_PONG,
  107. .plen = 0};
  108. luat_websocket_send_frame(websocket_ctrl, &pkg);
  109. }
  110. void luat_websocket_reconnect(luat_websocket_ctrl_t *websocket_ctrl) {
  111. int ret = luat_websocket_connect(websocket_ctrl);
  112. if (ret)
  113. {
  114. LLOGI("reconnect init socket ret=%d\n", ret);
  115. luat_websocket_close_socket(websocket_ctrl);
  116. }
  117. }
  118. int luat_websocket_set_cb(luat_websocket_ctrl_t *websocket_ctrl, luat_websocket_cb_t websocket_cb){
  119. if (websocket_ctrl == NULL || websocket_ctrl->netc == NULL)
  120. return -1;
  121. websocket_ctrl->websocket_cb = websocket_cb;
  122. return 0;
  123. }
  124. #ifdef __LUATOS__
  125. int l_luat_websocket_msg_cb(luat_websocket_ctrl_t *websocket_ctrl, int arg1, int arg2);
  126. #endif
  127. static int luat_websocket_msg_cb(luat_websocket_ctrl_t *websocket_ctrl, int arg1, int arg2){
  128. #ifdef __LUATOS__
  129. l_luat_websocket_msg_cb(websocket_ctrl,arg1,arg2);
  130. #else
  131. if (websocket_ctrl->websocket_cb){
  132. luat_websocket_cb_t websocket_cb = websocket_ctrl->websocket_cb;
  133. websocket_cb(websocket_ctrl, arg1,arg2);
  134. }else{
  135. LLOGE("websocket need set cb\n");
  136. }
  137. #endif
  138. return 0;
  139. }
  140. LUAT_RT_RET_TYPE luat_websocket_timer_callback(LUAT_RT_CB_PARAM)
  141. {
  142. luat_websocket_ctrl_t *websocket_ctrl = (luat_websocket_ctrl_t *)param;
  143. luat_websocket_msg_cb(websocket_ctrl, WEBSOCKET_MSG_TIMER_PING, 0);
  144. }
  145. static void reconnect_timer_cb(LUAT_RT_CB_PARAM)
  146. {
  147. luat_websocket_ctrl_t *websocket_ctrl = (luat_websocket_ctrl_t *)param;
  148. luat_websocket_msg_cb(websocket_ctrl, WEBSOCKET_MSG_RECONNECT, 0);
  149. }
  150. int luat_websocket_init(luat_websocket_ctrl_t *websocket_ctrl, int adapter_index)
  151. {
  152. memset(websocket_ctrl, 0, sizeof(luat_websocket_ctrl_t));
  153. websocket_ctrl->adapter_index = adapter_index;
  154. websocket_ctrl->netc = network_alloc_ctrl(adapter_index);
  155. if (!websocket_ctrl->netc)
  156. {
  157. LLOGW("network_alloc_ctrl fail");
  158. return -1;
  159. }
  160. network_init_ctrl(websocket_ctrl->netc, NULL, luat_websocket_callback, websocket_ctrl);
  161. websocket_ctrl->websocket_state = 0;
  162. websocket_ctrl->netc->is_debug = 0;
  163. websocket_ctrl->keepalive = 60;
  164. network_set_base_mode(websocket_ctrl->netc, 1, 10000, 0, 0, 0, 0);
  165. network_set_local_port(websocket_ctrl->netc, 0);
  166. websocket_ctrl->reconnect_timer = luat_create_rtos_timer(reconnect_timer_cb, websocket_ctrl, NULL);
  167. websocket_ctrl->ping_timer = luat_create_rtos_timer(luat_websocket_timer_callback, websocket_ctrl, NULL);
  168. return 0;
  169. }
  170. int luat_websocket_autoreconn(luat_websocket_ctrl_t *websocket_ctrl, uint8_t reconnect,uint32_t reconnect_time){
  171. websocket_ctrl->reconnect = reconnect;
  172. if (reconnect){
  173. websocket_ctrl->reconnect_time = reconnect_time<1000?1000:reconnect_time;
  174. }
  175. return 0;
  176. }
  177. int luat_websocket_set_connopts(luat_websocket_ctrl_t *websocket_ctrl, luat_websocket_connopts_t* opts)
  178. {
  179. int is_tls = 0;
  180. const char *tmp = opts->url;
  181. LLOGD("url %s", tmp);
  182. // TODO 支持基本授权的URL ws://wendal:123@wendal.cn:8080/abc
  183. websocket_ctrl->host[0] = 0;
  184. char port_tmp[6] = {0};
  185. uint16_t port = 0;
  186. if (!memcmp(tmp, "wss://", strlen("wss://")))
  187. {
  188. // LLOGD("using WSS");
  189. is_tls = 1;
  190. tmp += strlen("wss://");
  191. }
  192. else if (!memcmp(tmp, "ws://", strlen("ws://")))
  193. {
  194. // LLOGD("using ws");
  195. is_tls = 0;
  196. tmp += strlen("ws://");
  197. }
  198. // LLOGD("tmp %s", tmp);
  199. size_t uri_start_index = 0;
  200. for (size_t i = 0; i < strlen(tmp); i++)
  201. {
  202. if (tmp[i] == '/')
  203. {
  204. uri_start_index = i;
  205. break;
  206. }
  207. }
  208. if (uri_start_index < 2) {
  209. uri_start_index = strlen(tmp);
  210. }
  211. for (size_t j = 0; j < uri_start_index; j++)
  212. {
  213. if (tmp[j] == ':')
  214. {
  215. memcpy(websocket_ctrl->host, tmp, j);
  216. websocket_ctrl->host[j] = 0;
  217. memcpy(port_tmp, tmp + j + 1, uri_start_index - j - 1);
  218. port = atoi(port_tmp);
  219. //LLOGD("port str %s %d", port_tmp, port);
  220. // LLOGD("found custom host %s port %d", websocket_ctrl->host, port);
  221. break;
  222. }
  223. }
  224. // 没有自定义host
  225. if (websocket_ctrl->host[0] == 0)
  226. {
  227. memcpy(websocket_ctrl->host, tmp, uri_start_index);
  228. websocket_ctrl->host[uri_start_index] = 0;
  229. // LLOGD("found custom host %s", websocket_ctrl->host);
  230. }
  231. memcpy(websocket_ctrl->uri, tmp + uri_start_index, strlen(tmp) - uri_start_index);
  232. websocket_ctrl->uri[strlen(tmp) - uri_start_index] = 0;
  233. if (port == 0) {
  234. port = is_tls ? 443 : 80;
  235. }
  236. if (websocket_ctrl->uri[0] == 0)
  237. {
  238. websocket_ctrl->uri[0] = '/';
  239. websocket_ctrl->uri[1] = 0x00;
  240. }
  241. // LLOGD("host %s port %d uri %s", host, port, uri);
  242. // memcpy(websocket_ctrl->host, host, strlen(host) + 1);
  243. websocket_ctrl->remote_port = port;
  244. // memcpy(websocket_ctrl->uri, uri, strlen(uri) + 1);
  245. LLOGD("host %s port %d uri %s", websocket_ctrl->host, port, websocket_ctrl->uri);
  246. if (is_tls)
  247. {
  248. if (network_init_tls(websocket_ctrl->netc, 0)){
  249. return -1;
  250. }
  251. }
  252. else
  253. {
  254. network_deinit_tls(websocket_ctrl->netc);
  255. }
  256. if (opts->keepalive > 0) {
  257. websocket_ctrl->keepalive = opts->keepalive;
  258. }
  259. if (opts->use_ipv6) {
  260. network_connect_ipv6_domain(websocket_ctrl->netc, 1);
  261. }
  262. return 0;
  263. }
  264. static void websocket_reconnect(luat_websocket_ctrl_t *websocket_ctrl)
  265. {
  266. LLOGI("reconnect after %dms", websocket_ctrl->reconnect_time);
  267. websocket_ctrl->buffer_offset = 0;
  268. //websocket_ctrl->reconnect_timer = luat_create_rtos_timer(reconnect_timer_cb, websocket_ctrl, NULL);
  269. luat_stop_rtos_timer(websocket_ctrl->reconnect_timer);
  270. luat_start_rtos_timer(websocket_ctrl->reconnect_timer, websocket_ctrl->reconnect_time, 0);
  271. }
  272. void luat_websocket_close_socket(luat_websocket_ctrl_t *websocket_ctrl)
  273. {
  274. LLOGI("websocket closing socket");
  275. if (websocket_ctrl->netc)
  276. {
  277. network_force_close_socket(websocket_ctrl->netc);
  278. }
  279. luat_websocket_msg_cb(websocket_ctrl, WEBSOCKET_MSG_DISCONNECT, 0);
  280. luat_stop_rtos_timer(websocket_ctrl->ping_timer);
  281. websocket_ctrl->websocket_state = 0;
  282. if (websocket_ctrl->reconnect) {
  283. websocket_reconnect(websocket_ctrl);
  284. }
  285. }
  286. void luat_websocket_release_socket(luat_websocket_ctrl_t *websocket_ctrl)
  287. {
  288. luat_websocket_msg_cb(websocket_ctrl, WEBSOCKET_MSG_RELEASE, 0);
  289. if (websocket_ctrl->ping_timer) {
  290. luat_release_rtos_timer(websocket_ctrl->ping_timer);
  291. websocket_ctrl->ping_timer = NULL;
  292. }
  293. if (websocket_ctrl->reconnect_timer) {
  294. luat_release_rtos_timer(websocket_ctrl->reconnect_timer);
  295. websocket_ctrl->reconnect_timer = NULL;
  296. }
  297. if (websocket_ctrl->headers) {
  298. luat_heap_free(websocket_ctrl->headers);
  299. websocket_ctrl->headers = NULL;
  300. }
  301. if (websocket_ctrl->netc)
  302. {
  303. network_release_ctrl(websocket_ctrl->netc);
  304. websocket_ctrl->netc = NULL;
  305. }
  306. }
  307. static const char* ws_headers =
  308. "Upgrade: websocket\r\n"
  309. "Connection: Upgrade\r\n"
  310. "Sec-WebSocket-Key: w4v7O6xFTi36lq3RNcgctw==\r\n"
  311. "Sec-WebSocket-Version: 13\r\n"
  312. "\r\n";
  313. static int websocket_connect(luat_websocket_ctrl_t *websocket_ctrl)
  314. {
  315. LLOGD("request host %s port %d uri %s", websocket_ctrl->host, websocket_ctrl->remote_port, websocket_ctrl->uri);
  316. int ret = 0;
  317. // 借用pkg_buff
  318. if (0) {
  319. ret = snprintf_((char*)websocket_ctrl->pkg_buff,
  320. WEBSOCKET_RECV_BUF_LEN_MAX,
  321. "GET %s HTTP/1.1\r\n"
  322. "Host: %s\r\n",
  323. websocket_ctrl->uri, websocket_ctrl->host);
  324. }
  325. else {
  326. ret = snprintf_((char*)websocket_ctrl->pkg_buff,
  327. WEBSOCKET_RECV_BUF_LEN_MAX,
  328. "GET %s HTTP/1.1\r\n"
  329. "Host: %s:%d\r\n",
  330. websocket_ctrl->uri, websocket_ctrl->host, websocket_ctrl->remote_port);
  331. }
  332. LLOGD("Request %s", websocket_ctrl->pkg_buff);
  333. ret = luat_websocket_send_packet(websocket_ctrl, websocket_ctrl->pkg_buff, ret);
  334. if (websocket_ctrl->headers) {
  335. luat_websocket_send_packet(websocket_ctrl, websocket_ctrl->headers, strlen(websocket_ctrl->headers));
  336. }
  337. luat_websocket_send_packet(websocket_ctrl, ws_headers, strlen(ws_headers));
  338. LLOGD("websocket_connect ret %d", ret);
  339. return ret;
  340. }
  341. int luat_websocket_send_frame(luat_websocket_ctrl_t *websocket_ctrl, luat_websocket_pkg_t *pkg)
  342. {
  343. char *dst = luat_heap_malloc(pkg->plen + 8);
  344. if (dst == NULL) {
  345. LLOGE("out of memory when send_frame");
  346. return -2;
  347. }
  348. memset(dst, 0, pkg->plen + 8);
  349. size_t offset = 0;
  350. size_t ret = 0;
  351. // first byte, FIN and OPTCODE
  352. dst[0] = pkg->FIN << 7;
  353. dst[0] |= (pkg->OPT_CODE & 0xF);
  354. if (pkg->plen < 126)
  355. {
  356. dst[1] = pkg->plen;
  357. offset = 2;
  358. }
  359. else if (pkg->plen < 0xFFFF)
  360. {
  361. dst[1] = 126;
  362. dst[2] = pkg->plen >> 8;
  363. dst[3] = pkg->plen & 0xFF;
  364. offset = 4;
  365. }
  366. else {
  367. LLOGE("pkg too large %d", pkg->plen);
  368. luat_heap_free(dst);
  369. return -1;
  370. }
  371. dst[1] |= 1 << 7;
  372. print_pkg("uplink", dst, pkg);
  373. // 添加mark, TODO 改成随机?
  374. char mark[] = {0, 1, 2, 3};
  375. memcpy(dst + offset, mark, 4);
  376. offset += 4;
  377. if (pkg->plen > 0)
  378. {
  379. for (size_t i = 0; i < pkg->plen; i++)
  380. {
  381. dst[offset + i] = pkg->payload[i] ^ (mark[i % 4]);
  382. }
  383. }
  384. ret = luat_websocket_send_packet(websocket_ctrl, dst, offset + pkg->plen);
  385. luat_heap_free(dst);
  386. return ret > 0 ? 0 : -8;
  387. }
  388. static int websocket_parse(luat_websocket_ctrl_t *websocket_ctrl)
  389. {
  390. int ret = 0;
  391. char *buf = (char*)websocket_ctrl->pkg_buff;
  392. LLOGD("websocket_parse offset %d %d", websocket_ctrl->buffer_offset, websocket_ctrl->websocket_state);
  393. if (websocket_ctrl->websocket_state == 0)
  394. {
  395. if (websocket_ctrl->buffer_offset < strlen("HTTP/1.1 101"))
  396. { // 最起码得等5个字符
  397. LLOGD("wait more data offset %d", websocket_ctrl->buffer_offset);
  398. return 0;
  399. }
  400. // 前3个字符肯定是101, 否则必然是不合法的
  401. if (memcmp("HTTP/1.1 101", buf, strlen("HTTP/1.1 101")))
  402. {
  403. buf[websocket_ctrl->buffer_offset] = 0;
  404. LLOGE("server not support websocket? resp code %s", buf);
  405. return -1;
  406. }
  407. // 然后找\r\n\r\n
  408. for (size_t i = 4; i < websocket_ctrl->buffer_offset; i++)
  409. {
  410. if (!memcmp("\r\n\r\n", buf + i, 4))
  411. {
  412. // LLOGD("Found \\r\\n\\r\\n");
  413. // 找到了!! 但貌似完全不用处理呢
  414. websocket_ctrl->buffer_offset = 0;
  415. LLOGI("ready!!");
  416. websocket_ctrl->websocket_state = 1;
  417. luat_stop_rtos_timer(websocket_ctrl->ping_timer);
  418. luat_start_rtos_timer(websocket_ctrl->ping_timer, 30000, 1);
  419. luat_websocket_msg_cb(websocket_ctrl, WEBSOCKET_MSG_CONNACK, 0);
  420. return 1;
  421. }
  422. }
  423. // LLOGD("Not Found \\r\\n\\r\\n %s", buf);
  424. return 0;
  425. }
  426. if (websocket_ctrl->buffer_offset < 2)
  427. {
  428. LLOGD("wait more data offset %d", websocket_ctrl->buffer_offset);
  429. return 0;
  430. }
  431. // 判断数据长度, 前几个字节能判断出够不够读出websocket的头
  432. luat_websocket_pkg_t pkg = {0};
  433. ret = luat_websocket_payload(buf, &pkg, websocket_ctrl->buffer_offset);
  434. if (ret == 0) {
  435. LLOGD("wait more data offset %d", websocket_ctrl->buffer_offset);
  436. return 0;
  437. }
  438. if (ret < 0) {
  439. LLOGI("payload too large!!!");
  440. return -1;
  441. }
  442. switch (pkg.OPT_CODE)
  443. {
  444. case 0x00: // 连续帧
  445. break;
  446. case 0x01: // 文本帧
  447. break;
  448. case 0x02: // 二进制帧
  449. break;
  450. case 0x08:
  451. // 主动断开? 我擦
  452. LLOGD("server say CLOSE");
  453. return -1;
  454. case 0x09:
  455. // ping->pong
  456. luat_websocket_pong(websocket_ctrl);
  457. break;
  458. case 0x0A:
  459. break;
  460. default:
  461. LLOGE("unkown optcode 0x%02X", pkg.OPT_CODE);
  462. return -1;
  463. }
  464. size_t pkg_len = pkg.plen >= 126 ? pkg.plen + 4 : pkg.plen + 2;
  465. if (pkg.OPT_CODE <= 0x02)
  466. {
  467. char *buff = luat_heap_malloc(pkg_len);
  468. if (buff == NULL)
  469. {
  470. LLOGE("out of memory when malloc websocket buff");
  471. return -1;
  472. }
  473. memcpy(buff, buf, pkg_len);
  474. luat_websocket_msg_cb(websocket_ctrl, WEBSOCKET_MSG_PUBLISH, (int)buff);
  475. }
  476. // 处理完成后, 如果还有数据, 移动数据, 继续处理
  477. websocket_ctrl->buffer_offset -= pkg_len;
  478. if (websocket_ctrl->buffer_offset > 0)
  479. {
  480. memmove(websocket_ctrl->pkg_buff, websocket_ctrl->pkg_buff + pkg_len, websocket_ctrl->buffer_offset);
  481. return 1;
  482. }
  483. return 0;
  484. }
  485. int luat_websocket_read_packet(luat_websocket_ctrl_t *websocket_ctrl)
  486. {
  487. uint32_t total_len = 0;
  488. uint32_t rx_len = 0;
  489. int result = network_rx(websocket_ctrl->netc, NULL, 0, 0, NULL, NULL, &total_len);
  490. if (total_len == 0)
  491. {
  492. LLOGW("rx event but NO data wait for recv");
  493. return 0;
  494. }
  495. int recv_want = 0;
  496. while (((WEBSOCKET_RECV_BUF_LEN_MAX + 8) - websocket_ctrl->buffer_offset) > 0)
  497. {
  498. if (WEBSOCKET_MAX_READ > (WEBSOCKET_RECV_BUF_LEN_MAX - websocket_ctrl->buffer_offset))
  499. {
  500. recv_want = WEBSOCKET_RECV_BUF_LEN_MAX - websocket_ctrl->buffer_offset;
  501. }
  502. else
  503. {
  504. recv_want = WEBSOCKET_MAX_READ;
  505. }
  506. // 从网络接收数据
  507. result = network_rx(websocket_ctrl->netc, websocket_ctrl->pkg_buff + websocket_ctrl->buffer_offset, recv_want, 0, NULL, NULL, &rx_len);
  508. if (rx_len == 0 || result != 0)
  509. {
  510. LLOGD("rx_len %d result %d", rx_len, result);
  511. break;
  512. }
  513. // 收到数据了, 传给处理函数继续处理
  514. // 数据的长度变更, 触发传递
  515. websocket_ctrl->buffer_offset += rx_len;
  516. LLOGD("data recv %d offset %d", rx_len, websocket_ctrl->buffer_offset);
  517. further:
  518. result = websocket_parse(websocket_ctrl);
  519. if (result == 0)
  520. {
  521. // OK
  522. }
  523. else if (result == 1)
  524. {
  525. if (websocket_ctrl->buffer_offset > 0)
  526. goto further;
  527. else
  528. {
  529. continue;
  530. }
  531. }
  532. else
  533. {
  534. LLOGW("websocket_parse ret %d", result);
  535. //luat_websocket_close_socket(websocket_ctrl);
  536. return -1;
  537. }
  538. }
  539. if (WEBSOCKET_RECV_BUF_LEN_MAX + 8 < websocket_ctrl->buffer_offset) {
  540. LLOGD("pkg maybe too large");
  541. return -1;
  542. }
  543. return 0;
  544. }
  545. static int32_t luat_websocket_callback(void *data, void *param)
  546. {
  547. OS_EVENT *event = (OS_EVENT *)data;
  548. luat_websocket_ctrl_t *websocket_ctrl = (luat_websocket_ctrl_t *)param;
  549. int ret = 0;
  550. // LLOGD("LINK %d ON_LINE %d EVENT %d TX_OK %d CLOSED %d",EV_NW_RESULT_LINK & 0x0fffffff,EV_NW_RESULT_CONNECT & 0x0fffffff,EV_NW_RESULT_EVENT & 0x0fffffff,EV_NW_RESULT_TX & 0x0fffffff,EV_NW_RESULT_CLOSE & 0x0fffffff);
  551. //LLOGD("network websocket cb %8X %s %8X", event->ID & 0x0ffffffff, network_ctrl_callback_event_string(event->ID), event->Param1);
  552. if (event->ID == EV_NW_RESULT_LINK)
  553. {
  554. return 0; // 这里应该直接返回, 不能往下调用network_wait_event
  555. }
  556. else if (event->ID == EV_NW_RESULT_CONNECT)
  557. {
  558. if (event->Param1 == 0) {
  559. ret = websocket_connect(websocket_ctrl);
  560. if (ret < 0) {
  561. return 0; // 发送失败, 那么
  562. }
  563. }
  564. else {
  565. // 连接失败, 重连吧.
  566. }
  567. }
  568. else if (event->ID == EV_NW_RESULT_EVENT)
  569. {
  570. if (event->Param1 == 0)
  571. {
  572. ret = luat_websocket_read_packet(websocket_ctrl);
  573. if (ret < 0) {
  574. luat_websocket_close_socket(websocket_ctrl);
  575. return ret;
  576. }
  577. // LLOGD("luat_websocket_read_packet ret:%d",ret);
  578. luat_stop_rtos_timer(websocket_ctrl->ping_timer);
  579. luat_start_rtos_timer(websocket_ctrl->ping_timer, websocket_ctrl->keepalive * 1000, 1);
  580. }
  581. }
  582. else if (event->ID == EV_NW_RESULT_TX)
  583. {
  584. luat_stop_rtos_timer(websocket_ctrl->ping_timer);
  585. luat_start_rtos_timer(websocket_ctrl->ping_timer, websocket_ctrl->keepalive * 1000, 1);
  586. if (websocket_ctrl->frame_wait) {
  587. websocket_ctrl->frame_wait --;
  588. luat_websocket_msg_cb(websocket_ctrl, WEBSOCKET_MSG_SENT, 0);
  589. }
  590. }
  591. else if (event->ID == EV_NW_RESULT_CLOSE)
  592. {
  593. }
  594. if (event->Param1)
  595. {
  596. LLOGW("websocket_callback param1 %d, closing socket", event->Param1);
  597. luat_websocket_close_socket(websocket_ctrl);
  598. return 0;
  599. }
  600. ret = network_wait_event(websocket_ctrl->netc, NULL, 0, NULL);
  601. if (ret < 0)
  602. {
  603. LLOGW("network_wait_event ret %d, closing socket", ret);
  604. luat_websocket_close_socket(websocket_ctrl);
  605. return -1;
  606. }
  607. return 0;
  608. }
  609. int luat_websocket_connect(luat_websocket_ctrl_t *websocket_ctrl)
  610. {
  611. int ret = 0;
  612. const char *hostname = websocket_ctrl->host;
  613. uint16_t port = websocket_ctrl->remote_port;
  614. LLOGI("connect host %s port %d", hostname, port);
  615. network_close(websocket_ctrl->netc, 0);
  616. ret = network_connect(websocket_ctrl->netc, hostname, strlen(hostname), (!network_ip_is_vaild(&websocket_ctrl->ip_addr)) ? NULL : &(websocket_ctrl->ip_addr), port, 0) < 0;
  617. LLOGD("network_connect ret %d", ret);
  618. if (ret < 0)
  619. {
  620. network_close(websocket_ctrl->netc, 0);
  621. return -1;
  622. }
  623. return 0;
  624. }
  625. int luat_websocket_set_headers(luat_websocket_ctrl_t *websocket_ctrl, char *headers) {
  626. if (websocket_ctrl == NULL)
  627. return 0;
  628. if (websocket_ctrl->headers != NULL) {
  629. luat_heap_free(websocket_ctrl->headers);
  630. websocket_ctrl->headers = NULL;
  631. }
  632. if (headers) {
  633. websocket_ctrl->headers = headers;
  634. }
  635. return 0;
  636. }