2.1.4websocket协议与服务器实现

前言

本文基于websocket协议，在2.1.3http/https服务器的实现中TCP服务器代码的基础上实现websocket服务器。

websocket

websocket的主要应用场景

• 服务器主动推送数据给浏览器

websocket握手流程（在tcp三次握手之后）

• 浏览器发送请求
• 服务器接收请求并获取Sec-WebSocket-Key的值并进行相应的计算得到Sec-WebSocket-Accept的值，然后返回固定格式的信息给浏览器
• 浏览器接收信息并将Sec-WebSocket-Accept值与浏览器自己计算的值进行比较，若相同则握手成功

服务器接收到的浏览器的websocket请求为：

GET / HTTP/1.1
Host: 192.168.210.132:9999
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:105.0) Gecko/20100101 Firefox/105.0
Accept: */*
Accept-Language: zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2
Accept-Encoding: gzip, deflate
Sec-WebSocket-Version: 13
Origin: null
Sec-WebSocket-Extensions: permessage-deflate
Sec-WebSocket-Key: GFjXgXWdcYvUfA/wMx/kig==
Connection: keep-alive, Upgrade
Pragma: no-cache
Cache-Control: no-cache
Upgrade: websocket

服务器返回给浏览器的信息为：

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: value

其中value需通过以下计算得到：

// 请求中的Sec-WebSocket-Key
key = GFjXgXWdcYvUfA/wMx/kig==
// websocket协议定义的GUID
GUID = 258EAFA5-E914-47DA-95CA-C5AB0DC85B11  
// 将上面两段字符合并
str =  GFjXgXWdcYvUfA/wMx/kig==258EAFA5-E914-47DA-95CA-C5AB0DC85B11 
// 进行SHA-1加密算法处理
sha = SHA-1(str)
// 进行base64编码
value = base64-encoded(sha)

握手信息为字符格式，而传输的数据为二进制格式，数据的格式如下图：

基于websocket协议的服务器

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>

#include <openssl/sha.h>
#include <openssl/pem.h>
#include <openssl/bio.h>
#include <openssl/evp.h>

#define BUFFER_LENGTH       1024
#define MAX_EPOLL_EVENTS    1024

#define SERVER_PORT         9999
#define PORT_COUNT          1

typedef int NCALLBACK(int, int, void*);

// --- websocket
#define GUID                "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"

#define ACCEPT_KEY_LENGTH   64

enum  // 状态机
{
    WS_END = 0,  // wsstatus初始值为WS_END，故 WS_END = 0
    WS_HANDSHARK = 1,
    WS_TRANSMISSION = 2,
    WS_COUNT
};

// websocket数据头
typedef struct ws_ophdr
{
    // 小端
    unsigned char opcode:4,
                  rsv3:1,
                  rsv2:1,
                  rsv1:1,
                  fin:1;

    unsigned char pl_len:7,
                  mask:1;
}ws_ophdr;
// ---

typedef struct ntyevent
{
    int fd;
    int events;  // EPOLLIN, EPOLLOUT等
    void* arg;
    int (*callback)(int fd, int events, void* arg);

    int status;  // 0代表fd未被epoll监听，1代表fd已被epoll监听
    char rbuffer[BUFFER_LENGTH];
    char wbuffer[BUFFER_LENGTH];

    int rlength;
    int wlength;
    // long last_active;

    char sec_accept[ACCEPT_KEY_LENGTH];

    int wsstatus;  // 0, 1, 2, 3

}ntyevent;

typedef struct eventblock
{
    struct ntyevent* events;
    struct eventblock* next;
}eventblock;

// 反应堆，监听事件，管理fd相关内容（查找fd对应的ntyevent等）
typedef struct ntyreactor
{
    int epfd;
    int blkcnt;

    struct eventblock* evblks;
}ntyreactor;

int recv_cb(int fd, int events, void* arg);
int send_cb(int fd, int events, void* arg);

// --- ntyevent相关操作
// 设置ntyevent
void nty_event_set(ntyevent* ev, int fd, NCALLBACK callback, void* arg)
{
    ev->fd = fd;
    ev->callback = callback;
    ev->events = 0;
    ev->arg = arg;
    // ev->last_active = time(NULL);

    return;
}

// 将ntyevent对应fd及事件添加/修改到epoll监听列表
int nty_event_add(int epfd, int events, ntyevent* ev)
{
    struct epoll_event ep_ev = {0, {0}};
    ep_ev.data.ptr = ev;
    ep_ev.events = ev->events = events;

    int op;
    // 判断fd是否已经被epoll监听
    // 若是，则修改该fd被监听的事件，否则，将该fd添加到epoll监听列表
    if(ev->status == 1)
    {
        op = EPOLL_CTL_MOD;
    }
    else
    {
        op = EPOLL_CTL_ADD;
        ev->status = 1;
    }

    if(epoll_ctl(epfd, op, ev->fd, &ep_ev) < 0)
    {
        printf("event add failed [fd=%d], events[%d]\n",ev->fd, events);
        return -1;
    }

    return 0;
}

// 将ntyevent对应fd及事件从epoll监听列表中删除
int nty_event_del(int epfd, ntyevent* ev)
{
    struct epoll_event ep_ev = {0, {0}};

    // 若fd未被添加到epoll监听列表
    if(ev->status != 1)
    {
        return -1;
    }

    ep_ev.data.ptr = ev;
    ev->status = 0;
    epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, &ep_ev);

    return 0;
}
// ---

// --- ntyreactor相关操作
// 初始化ntyreactor
int ntyreactor_init(ntyreactor* reactor)
{
    if(!reactor) return -1;
    memset(reactor, 0, sizeof(ntyreactor));

    reactor->epfd = epoll_create(1);
    if(reactor->epfd <= 0)
    {
        printf("create epfd in %s err %s\n", __func__, strerror(errno));
        return -2;
    }

    ntyevent* evs = (ntyevent*)malloc(MAX_EPOLL_EVENTS*sizeof(ntyevent));
    if(!evs)
    {
        printf("create epfd in %s err %s\n", __func__, strerror(errno));
        close(reactor->epfd);
        return -3;
    }
    memset(evs, 0, (MAX_EPOLL_EVENTS)*sizeof(ntyevent));

    eventblock* block = (eventblock*)malloc(sizeof(eventblock));
    if(!block)
    {
        printf("create epfd in %s err %s\n", __func__, strerror(errno));
        free(evs);
        close(reactor->epfd);
        return -3;
    }

    block->events = evs;
    block->next = NULL;

    reactor->evblks = block;
    reactor->blkcnt = 1;

    return 0;
}

// 析构ntyreactor
int ntyreactor_destory(ntyreactor* reactor)
{
    close(reactor->epfd);

    eventblock* blk = reactor->evblks;
    eventblock* blk_next;
    while(blk)
    {
        blk_next = blk->next;

        free(blk->events);
        free(blk);

        blk = blk_next;
    }

    return 0;
}

// 为ntyreactor的链表(evblks)创建一个新结点，并分配内存
int ntyreactor_alloc(ntyreactor* reactor)
{
    if(!reactor) return -1;
    if(!reactor->evblks) return -1;

    eventblock* blk = reactor->evblks;
    while(blk->next)
        blk = blk->next;

    ntyevent* evs = (ntyevent*)malloc((MAX_EPOLL_EVENTS)*sizeof(ntyevent));
    if(!evs)
    {
        printf("%s ntyevent failed\n", __func__);
        return -2;
    }
    memset(evs, 0, (MAX_EPOLL_EVENTS)*sizeof(ntyevent));

    eventblock* block = (eventblock*)malloc(sizeof(eventblock));
    if(!block)
    {
        printf("%s eventblock failed\n", __func__);
        free(evs);
        return -3;
    }

    block->events = evs;
    block->next = NULL;

    blk->next = block;
    ++(reactor->blkcnt);

    return 0;
}

// 在ntyreactor中查找sockfd对应的ntyevent
ntyevent* ntyreactor_idx(ntyreactor* reactor, int sockfd)
{
    if(!reactor) return NULL;
    if(!reactor->evblks) return NULL;

    int blkidx = sockfd / MAX_EPOLL_EVENTS;
    while(blkidx >= reactor->blkcnt)
    {
        ntyreactor_alloc(reactor);
    }

    int i = 0;
    eventblock* blk = reactor->evblks;
    while(i++ != blkidx && blk)
    {
        blk = blk->next;
    }

    return &blk->events[sockfd % MAX_EPOLL_EVENTS];
}

// 向ntyreactor添加listenfd
int ntyreactor_addlistener(ntyreactor* reactor, int sockfd, NCALLBACK* acceptor)
{
    if(!reactor) return -1;
    if(!reactor->evblks) return -1;

    ntyevent* event = ntyreactor_idx(reactor, sockfd);
    if(!event) return -1;

    nty_event_set(event, sockfd, acceptor, reactor);
    nty_event_add(reactor->epfd, EPOLLIN, event);

    return 0;
}

// 运行ntyreactor，即循环执行epoll_wait，对可读可写IO执行相应回调函数
int ntyreactor_run(ntyreactor* reactor)
{
    if(!reactor) return -1;
    if(reactor->epfd < 0) return -1;
    if(!reactor->evblks) return -1;

    struct epoll_event events[MAX_EPOLL_EVENTS+1];

    int checkpos = 0, i;
    
    while(1)
    {
        int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
        if(nready < 0)
        {
            printf("epoll_wait error, exit\n");
            continue;
        }

        for(i = 0;i < nready; i++)
        {
            ntyevent* ev = (ntyevent*)events[i].data.ptr;

            if((events[i].events & EPOLLIN) && (ev->events & EPOLLIN))    // 若IO可输入
            {
                ev->callback(ev->fd, events[i].events, ev->arg);
            }
            if((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT))  // 若IO可输出
            {
                ev->callback(ev->fd, events[i].events, ev->arg);
            }
        }
    }
}
// ---

// --- websocket
// 读取一行
int readline(char* allbuf, int idx, char* linebuf)
{
    int len = strlen(allbuf);
    for(;idx < len;idx++)
    {
        if(allbuf[idx] == '\r' && allbuf[idx+1] == '\n')
            return idx + 2;
        else
            *(linebuf++) = allbuf[idx]; 
    }

    return -1;
}

// base64编码
int base64_encode(char *in_str, int in_len, char *out_str)
{    
	BIO *b64, *bio;    
	BUF_MEM *bptr = NULL;    
	size_t size = 0;    

	if (in_str == NULL || out_str == NULL)        
		return -1;    

	b64 = BIO_new(BIO_f_base64());    
	bio = BIO_new(BIO_s_mem());    
	bio = BIO_push(b64, bio);
	
	BIO_write(bio, in_str, in_len);    
	BIO_flush(bio);    

	BIO_get_mem_ptr(bio, &bptr);    
	memcpy(out_str, bptr->data, bptr->length);    
	out_str[bptr->length-1] = '\0';    
	size = bptr->length;    

	BIO_free_all(bio);    
	return size;
}

// websocket握手
int ws_handshark(ntyevent* ev)
{
    int idx = 0;
    char sec_data[128] = {0};
    char sec_accept[128] = {0};

    do
    {
        char linebuf[BUFFER_LENGTH] = {0};
        idx = readline(ev->rbuffer, idx, linebuf);

        if(strstr(linebuf, "Sec-WebSocket-Key"))  // 获取Sec-WebSocket-Key
        {
            strcat(linebuf, GUID);

            SHA1(linebuf+strlen("Sec-WebSocket-Key: "), strlen(linebuf+strlen("Sec-WebSocket-Key: ")), sec_data);

            base64_encode(sec_data, strlen(sec_data), sec_accept);

            printf("idx: %d, line: %s\n", idx, sec_accept);

            memcpy(ev->sec_accept, sec_accept, ACCEPT_KEY_LENGTH);

            return 1;
        }
    }
    while((ev->rbuffer[idx] != '\r' || ev->rbuffer[idx+1] != '\n') && idx != -1);

    return 0;
}

// 将密文转为明文
void unmask(char* payload, int length, char* mask_key)
{
    int i = 0;

    for(i = 0;i < length;i++)
    {
        payload[i] ^= mask_key[i % 4];
    }
}

// 
int ws_transmission(ntyevent* ev)
{
    ws_ophdr* hdr = (ws_ophdr*)ev->rbuffer;

    printf("%s\n", __func__);

    if(hdr->pl_len < 126)
    {
        unsigned char* payload = NULL;
        if(hdr->mask)  // 有Masking-key(4 bytes)，为密文
        {
            payload = ev->rbuffer + 6;
            unmask(payload, hdr->pl_len, ev->rbuffer + 2);
        }
        else  // 无Masking-key(0 bytes)，为明文
        {
            payload = ev->rbuffer + 2;
        }

        printf("payload: %s\n", payload);

        return 1;
    }
    else if(hdr->pl_len == 126)
    {

    }
    else if(hdr->pl_len == 127)
    {

    }
    else
    {
        return 0;
        //assert(0);
    }
}

// 解析websocket请求
int ws_request(ntyevent* ev)
{
    if(ev->wsstatus == WS_HANDSHARK)
    {
        if(ws_handshark(ev))
            ev->wsstatus = WS_TRANSMISSION;
    }
    else if(ev->wsstatus == WS_TRANSMISSION)
    {
        if(ws_transmission(ev))
            ev->wsstatus = WS_END;
    }
}

// 响应websocket请求
int ws_response(ntyevent* ev)
{
    ev->wlength = sprintf(ev->wbuffer, "HTTP/1.1 101 Switching Protocols\r\n"
                                       "Upgrade: websocket\r\n"
                                       "Connection: Upgrade\r\n"
                                       "Sec-WebSocket-Accept: %s\r\n\r\n",
                                       ev->sec_accept);

    return ev->wlength;
}
// ---

// --- 回调函数
// recv回调函数
int recv_cb(int fd, int events, void* arg)
{
    ntyreactor* reactor = (ntyreactor*)arg;
    ntyevent* ev = ntyreactor_idx(reactor, fd);

    if(!ev) return -1;

    int len = recv(fd, ev->rbuffer, BUFFER_LENGTH, 0);
    nty_event_del(reactor->epfd, ev);

    if(len > 0)  // 已接收数据
    {
        ev->rlength = len;
        ev->rbuffer[len] = '\0';  // 去除脏数据

        //printf("recv [%d]:%s\n", fd, ev->rbuffer);

        // --- websocket
        printf("wsstatus: %d\n", ev->wsstatus);
        if(ev->wsstatus == WS_END)
        {
            ev->wsstatus = WS_HANDSHARK;
        }
        printf("request\n");
        ws_request(ev);  // 解析websocket请求
        // ---

        // 接收数据后，将fd设置为待发送模式
        nty_event_set(ev, fd, send_cb, reactor);
        nty_event_add(reactor->epfd, EPOLLOUT, ev);
    }
    else if(len == 0)  // 客户端断开连接
    {
        nty_event_del(reactor->epfd, ev);
        printf("recv_cb --> disconnect\n");
        close(ev->fd);
    }
    else  // 发生异常
    {
        if(errno == EAGAIN && errno == EWOULDBLOCK)
        {

        }
        else if(errno == ECONNRESET)
        {
            nty_event_del(reactor->epfd, ev);
            close(ev->fd);
        }
        printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));
        // strerror() 获取指向错误消息字符串的指针
    }

    return len;
}

// send回调函数
int send_cb(int fd, int events, void* arg)
{
    ntyreactor* reactor = (ntyreactor*)arg;
    ntyevent* ev = ntyreactor_idx(reactor, fd);

    if(!ev) return -1;

    // --- websocket
    ws_response(ev);  // 响应websocket请求
    // ---

    int len = send(fd, ev->wbuffer, ev->wlength, 0);
    if(len > 0)  // 已发送数据
    {
        //printf("send[fd=%d], [%d]%s\n",fd, len, ev->wbuffer);
        
        // 发送数据后，将fd设置为待接收模式
        nty_event_del(reactor->epfd, ev);
        nty_event_set(ev, fd, recv_cb, reactor);
        nty_event_add(reactor->epfd, EPOLLIN, ev);
    }
    else  // 发送失败
    {
        nty_event_del(reactor->epfd, ev);
        close(ev->fd);

        printf("send[fd=%d] error %s\n", fd, strerror(errno));
    }

    return len;
}

// accept回调函数
int accept_cb(int fd, int events, void* arg)
{
    ntyreactor* reactor = (ntyreactor*)arg;
    if(!reactor) return -1;

    struct sockaddr_in client_addr;
    socklen_t len = sizeof(client_addr);

    int clientfd = 0;

    if((clientfd = accept(fd, (struct sockaddr*)&client_addr, &len)) == -1)
    {
        if(errno != EAGAIN && errno != EINTR)
        {

        }
        printf("accept[fd=%d] error %s\n", fd, strerror(errno));
        return -1;
    }

    // 将clientfd设置为非阻塞
    int flag = 0;
    if((flag = fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0)
    {
        printf("%s: fcntl nonblocking failed, %d\n", __func__, MAX_EPOLL_EVENTS);
        return -1;
    }

    ntyevent* event = ntyreactor_idx(reactor, clientfd);

    if(!event) return -1;

    // 初始化客户端fd为接收模式
	nty_event_set(event, clientfd, recv_cb, reactor);
	nty_event_add(reactor->epfd, EPOLLIN, event);

    printf("new connect [%s:%d], pos[%d]\n",
        inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), clientfd);

    return 0;
}
// ---

// 创建服务器端口（listenfd）
int init_sock(short port)
{
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    fcntl(fd, F_SETFL, O_NONBLOCK);

    struct sockaddr_in server_addr;
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons(port);

    bind(fd, (struct sockaddr*)&server_addr, sizeof(server_addr));

    if(listen(fd, 20) < 0)
    {
        printf("listen failed : %s\n", strerror(errno));
        return -1;
    }

    printf("listen server port : %d\n", port);
    return fd;
}

int main(int argc, char* argv[])
{
    ntyreactor* reactor = (ntyreactor*)malloc(sizeof(ntyreactor));
    ntyreactor_init(reactor);

    unsigned short port = SERVER_PORT;
    if(argc == 2)
    {
        port = atoi(argv[1]);
    }

    int i = 0;
    int sockfds[PORT_COUNT] = {0};

    for(i = 0; i < PORT_COUNT; i++)
    {
        sockfds[i] = init_sock(port + i);
        ntyreactor_addlistener(reactor, sockfds[i], accept_cb);
    }

    ntyreactor_run(reactor);

    ntyreactor_destory(reactor);

    for(i = 0; i < PORT_COUNT; i++)
    {
        close(sockfds[i]);
    }
    free(reactor);

    return 0;
}

使用SHA-1加密算法及实现base64编码需安装openssl库

sudo apt-get install libssl-dev

编译时需链接动态库ssl及crypto

gcc -o websocket websocket.c -lssl -lcrypto

参考博客：

C语言 | 位域的使用详解

弄懂大端小端含义文章来源地址https://www.toymoban.com/news/detail-431936.html

到了这里，关于2.1.4websocket协议与服务器实现的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！