luat_websocket.c 19 KB

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