?

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?webcomm與前端的通訊-websocket服務(wù)器操作流程

?

#include <libwebsockets.h> #include <pthread.h> #include <string.h>#define MAX_PAYLOAD_SIZE (4096)/** 開啟確認(rèn)才有重發(fā)的機(jī)制 */ typedef struct comm_packet {void *content; /* is malloc'd */size_t content_len;uint8_t enable_ack; /* 是否需要響應(yīng) 0：不需要響應(yīng) 1：需要響應(yīng) */uint8_t retry_cnt; /* 重發(fā)的次數(shù)(ms) */ } comm_packet_t;/* one of these created for each message */ typedef struct msg {comm_packet_t *payload; /* is malloc'd */size_t len; } msg_t;/* one of these is created for each client connecting to us */ struct per_session_data__cetwebsocket {struct per_session_data__cetwebsocket *pss_list;struct lws *wsi;uint32_t tail;struct lws_ring *ring; /* ringbuffer holding unsent messages */ };/* one of these is created for each vhost our protocol is used with */ struct per_vhost_data__cetwebsocket {struct lws_context *context;struct lws_vhost *vhost;const struct lws_protocols *protocol;struct per_session_data__cetwebsocket *pss_list; /* linked-list of live pss*/ };static int callback_protocol_cetwebsocket_server(struct lws *wsi, enum lws_callback_reasons reason,void *user, void *in, size_t len);#define LWS_PLUGIN_PROTOCOL_CETWEBSOCKET \{ \"cet-webserver", /* "cet-webserver" */ \callback_protocol_cetwebsocket_server, \sizeof(struct per_session_data__cetwebsocket), \2048, \0, NULL, 0 \}static struct lws_protocols sg_protocols[] = {{ "http", lws_callback_http_dummy, 0, 0 }, LWS_PLUGIN_PROTOCOL_CETWEBSOCKET,{ NULL, NULL, 0, 0 } /* terminator */ };static int sg_interrupted = 0;/* destroys the message when everyone has had a copy of it */ static void cetwebsocket_destroy_message(void *_msg) {msg_t *message = _msg;comm_packet_t *ppacket = (comm_packet_t *)message->payload;free(ppacket->content);ppacket->content = NULL;free(message->payload);message->payload = NULL;message->len = 0; }static struct per_session_data__cetwebsocket ** cetwebsocket_traverse_find(struct per_vhost_data__cetwebsocket *vhd, struct per_session_data__cetwebsocket *pss) { ASSERT_PARAM_CHECK(vhd, NULL);ASSERT_PARAM_CHECK(pss, NULL);struct per_session_data__cetwebsocket **ppss = NULL;for (ppss = &(vhd->pss_list); *ppss; ppss = &((*ppss)->pss_list)) {if ((*ppss) == pss) {return ppss;}}return NULL; }static struct per_session_data__cetwebsocket * cetwebsocket_client_remove(struct per_vhost_data__cetwebsocket *vhd, struct per_session_data__cetwebsocket *pss) { ASSERT_PARAM_CHECK(vhd, NULL);ASSERT_PARAM_CHECK(pss, NULL);struct per_session_data__cetwebsocket **ppss = NULL;if (NULL != (ppss = cetwebsocket_traverse_find(vhd, pss))) {lws_ring_destroy(pss->ring);*ppss = pss->pss_list; return *ppss;}return pss; }static int cetwebsocket_insert_ring(struct per_session_data__cetwebsocket *pss, const char *func_id, const char *mid, uint8_t enable_ack, const void *content, int contentlen) { int n = 0;msg_t amsg;ASSERT_PARAM_CHECK(pss, -1);ASSERT_PARAM_CHECK(content, -1);n = (int)lws_ring_get_count_free_elements(pss->ring); if (!n) {return -1;}memset(&amsg, 0x00, sizeof(amsg));/* notice we over-allocate by LWS_PRE */amsg.payload = calloc(sizeof(char), sizeof(comm_packet_t));if (!amsg.payload) {return -1;}/* notice we over-allocate by LWS_PRE */amsg.payload->content = calloc(sizeof(char), LWS_PRE + contentlen + 1);if (!amsg.payload->content) {return -1;}amsg.payload->enable_ack = enable_ack;amsg.payload->retry_cnt = 0;amsg.payload->content_len = contentlen;amsg.len = amsg.payload->content_len + sizeof(comm_packet_t);memcpy((char *)amsg.payload->content + LWS_PRE, content, contentlen);/** 將實(shí)時消息插入到環(huán)形緩存中 */if (!lws_ring_insert(pss->ring, &amsg, 1)) {cetwebsocket_destroy_message(&amsg);return -1;}return 0; }static comm_packet_t * cetwebsocket_get_packet(struct lws_ring *ring, uint32_t *tail) {msg_t *pmsg = NULL;ASSERT_PARAM_CHECK(ring, NULL);pmsg = (msg_t *)lws_ring_get_element(ring, tail);return (NULL == pmsg)? NULL : (comm_packet_t *)pmsg->payload; }static int cetwebsocket_response_handler(struct per_session_data__cetwebsocket *pss, const char *mid) { comm_packet_t *ppacket;ASSERT_PARAM_CHECK(pss, -1);ASSERT_PARAM_CHECK(mid, -1);ppacket = cetwebsocket_get_packet(pss->ring, &pss->tail); if (NULL == ppacket)return -1;if (ppacket->enable_ack) { // ack respenseif (0 == strcmp(mid, ppacket->mid)) {lws_ring_consume_single_tail(pss->ring, &pss->tail, 1); }} else {; // not ack respense}return 0; }static int cetwebsocket_server_create(struct lws *wsi) { struct per_vhost_data__cetwebsocket *vhd = NULL;vhd = (struct per_vhost_data__cetwebsocket *)lws_protocol_vh_priv_zalloc(lws_get_vhost(wsi),lws_get_protocol(wsi), sizeof(struct per_vhost_data__cetwebsocket));vhd->context = lws_get_context(wsi);vhd->protocol = lws_get_protocol(wsi);vhd->vhost = lws_get_vhost(wsi);vhd->pss_debug_count = 0; vhd->pss_other_count = 0;vhd->pss_list = NULL;sg_ptr_websockets->vhost = vhd;return 0; }static int cetwebsocket_session_create(struct lws *wsi, struct per_vhost_data__cetwebsocket *vhd, struct per_session_data__cetwebsocket *pss) { char *cookie_id = NULL;char buffer[FIELDS_SIZE] = {0};ASSERT_PARAM_CHECK(wsi, -1);ASSERT_PARAM_CHECK(vhd, -1);ASSERT_PARAM_CHECK(pss, -1);pss->tail = 0;pss->wsi = wsi;pss->ring = lws_ring_create(sizeof(msg_t), (FIELDS_SIZE), cetwebsocket_destroy_message); if (!pss->ring) { return -1;}lws_ll_fwd_insert(pss, pss_list, vhd->pss_list);return 0; }static int cetwebsocket_server_write(struct lws *wsi, struct per_vhost_data__cetwebsocket *vhd, struct per_session_data__cetwebsocket *pss) { int len;comm_packet_t *ppacket;ASSERT_PARAM_CHECK(wsi, -1);ASSERT_PARAM_CHECK(vhd, -1);ASSERT_PARAM_CHECK(pss, -1);if (NULL == cetwebsocket_traverse_find(vhd, pss)) {return -1;}if (NULL == (ppacket = cetwebsocket_get_packet(pss->ring, &pss->tail))) {return -1;}len = lws_write(wsi, ((unsigned char *)ppacket->content) + LWS_PRE, ppacket->content_len, LWS_WRITE_TEXT);if (len < (int)ppacket->content_len) {return -1;}if (!ppacket->enable_ack || ppacket->retry_cnt > RETRY_MAX_CNT) { lws_ring_consume_single_tail(pss->ring, &pss->tail, 1);}return 0; }static int callback_protocol_cetwebsocket_server(struct lws *wsi, enum lws_callback_reasons reason,void *user, void *in, size_t len) {struct per_session_data__cetwebsocket *pss = (struct per_session_data__cetwebsocket *)user;struct per_vhost_data__cetwebsocket *vhd = sg_ptr_websockets->vhost;switch (reason) {case LWS_CALLBACK_PROTOCOL_INIT: cetwebsocket_server_create(wsi);break;case LWS_CALLBACK_PROTOCOL_DESTROY:break;case LWS_CALLBACK_ESTABLISHED:{ cetwebsocket_session_create(wsi, vhd, pss);break;}case LWS_CALLBACK_CLOSED: cetwebsocket_client_remove(vhd, pss);break;case LWS_CALLBACK_SERVER_WRITEABLE:{ cetwebsocket_server_write(wsi, vhd, pss);break;}case LWS_CALLBACK_RECEIVE:{ if (vhd->pss_list) { cetwebsocket_server_receive(wsi, vhd, pss, in, len);} break;}default:break;}return 0; }static int cetwebsocket_retry_writable_handler(struct per_vhost_data__cetwebsocket *vhost) {boolean is_writable = true;comm_packet_t *ppacket = NULL; struct per_session_data__cetwebsocket *pss = NULL;ASSERT_PARAM_CHECK(vhost, -1);for (pss = vhost->pss_list; pss; pss = pss->pss_list) {ppacket = cetwebsocket_get_packet(pss->ring, &pss->tail); if (NULL == ppacket)continue;if (ppacket->enable_ack) {if (ppacket->retry_cnt <= RETRY_MAX_CNT) {ppacket->retry_cnt += 1;} else {is_writable = false;}}if (true == is_writable) {lws_callback_on_writable(pss->wsi);}}return 0; }int main(int argc, char * const argv[]) {struct lws_context_creation_info info;struct lws_context *context; int n = 0/* , logs = LLL_USER | LLL_ERR | LLL_WARN | LLL_NOTICE| LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT | LLL_LATENCY | LLL_DEBUG *//* for LLL_ verbosity above NOTICE to be built into lws,* lws must have been configured and built with* -DCMAKE_BUILD_TYPE=DEBUG instead of =RELEASE *//* | LLL_INFO */ /* | LLL_PARSER */ /* | LLL_HEADER *//* | LLL_EXT */ /* | LLL_CLIENT */ /* | LLL_LATENCY *//* | LLL_DEBUG */;lws_set_log_level(0, NULL);memset(&info, 0, sizeof info); /* otherwise uninitialized garbage */info.port = 7681;info.protocols = sg_protocols;info.vhost_name = "localhost";info.ws_ping_pong_interval = 10;info.options = LWS_SERVER_OPTION_HTTP_HEADERS_SECURITY_BEST_PRACTICES_ENFORCE;info.options |= LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;info.ssl_cert_filepath = ME_GOAHEAD_SSL_CERTIFICATE;info.ssl_private_key_filepath = ME_GOAHEAD_SSL_KEY;context = lws_create_context(&info);if (!context) {return NULL;}while (n >= 0 && !sg_interrupted) { n = lws_service(context, 0);lws_cancel_service(context); }lws_context_destroy(context);return NULL; }

總結(jié)

以上是生活随笔為你收集整理的使用libwebsocket搭建websocket服务器实例的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。