luat_websocket.c 19 KB

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