luat_websocket.c 18 KB

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