简单websocket多进程服务器(附带客户端)

#include <sys/types.h> #include <sys/stat.h> #include <sys/socket.h> #include <sys/types.h> #include <arpa/inet.h> #include <signal.h> #include <fcntl.h> #include <unistd.h> #include <stdio.h> #include <sys/epoll.h> #include <string.h> #include <stdlib.h> #include <sys/resource.h> #include <syslog.h> #include <pthread.h> #include "base64.h" #define MAXEVENTS 100 typedef struct { u_char fin; u_char opencode; u_char mask; u_long payload_length; u_char masking_key[4]; } websocket_head; void strreverse(char *str, size_t length) { if (str == NULL || length <= 1) return; char *beg, *end; char tmp; for (beg = str, end = str + length - 1; beg < end; beg++, end--) { tmp = *beg; *beg = *end; *end = tmp; } } //守护进程创建 int daemonize(const char *pro_name) { int pid, i; int fd0, fd1, fd2; struct rlimit rl; struct sigaction act; umask(0); if ((pid = fork()) < 0) { perror("fork error"); return -1; } else if (pid > 0) exit(0); setsid(); act.sa_handler = SIG_IGN; act.sa_flags = 0; sigemptyset(&act.sa_mask); if (sigaction(SIGHUP, &act, NULL) < 0) { perror("sigaction SIGHUP error"); return -2; } if ((pid = fork()) < 0) { perror("fork error"); return -3; } else if (pid > 0) exit(0); if (chdir("/") < 0) { perror("chdir error"); return -4; } getrlimit(RLIMIT_NOFILE, &rl); if (rl.rlim_max == RLIM_INFINITY) rl.rlim_max = 1024; for (i = 0; i < rl.rlim_max; i++) close(i); fd0 = open("/dev/null", O_RDWR); fd1 = dup(0); fd2 = dup(0); openlog(pro_name, LOG_CONS, LOG_DAEMON); if (fd0 != 0 || fd1 != 1 || fd2 != 2) { syslog(LOG_ERR, "unexpected file descriptors: %d %d %d", fd0, fd1, fd2); return -5; } return 0; } //线程函数，等待子进程退出，避免产生僵尸进程 void *_wait(void *arg) { waitpid(*(pid_t *)arg, NULL, 0); } void set_noblock(int fd) { fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK); } void sig_fun(int sig) { printf("\nclose server......\n"); while (wait(NULL) > 0); //wait for child processes..... exit(EXIT_SUCCESS); } /* * 功能：初始化服务器，生成socket并绑定端口地址 * 参数：socket， 服务器IP， 端口 * */ int init_server(int *sockfd, const char *addr, char *port) { struct sockaddr_in server_addr; server_addr.sin_family = AF_INET; server_addr.sin_port = htons(atoi(port)); server_addr.sin_addr.s_addr = inet_addr(addr); if ((*sockfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) { perror("socket() error: "); return -1; } if (bind(*sockfd, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) { perror("bind() error: "); return -1; } if (listen(*sockfd, 10) == -1) { perror("listen() error: "); return -1; } return *sockfd; } void *ping(void *args) { int clt_fd = *(int *)args; char packet[2]; packet[0] = 0x89; packet[1] = 0x00; while (1) { write(clt_fd, packet, 2); sleep(20); } } /*get key from msg, than get server key~*/ char *get_ser_key(const char *msg, char *ser_key) { int index, len; char *begin; char client_key[256] = { '\0' }; unsigned char sha1_data[SHA_DIGEST_LENGTH + 1] = { '\0' }; const char *GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; if (msg == NULL || ser_key == NULL) return NULL; begin = strstr(msg, "Sec-WebSocket-Key: ") + strlen("Sec-WebSocket-Key: "); for (index = 0; ; index++) { if (begin[index] == '\r') break; client_key[index] = begin[index]; } //we have got the key.... //start get the ser_key index += sprintf(client_key + index, GUID); SHA1(client_key, index, sha1_data); base64_encode(sha1_data, strlen(sha1_data), ser_key, 256); return ser_key; } /*shake hand with client...*/ int shake_hand(int client_fd) { char msg_client[BUFSIZ] = { '\0' }; char ser_key[256] = { '\0' }; char buff[BUFSIZ]; int index = 0; if (read(client_fd, msg_client, BUFSIZ) <= 0) return -1; printf("client request hand: %s", msg_client); get_ser_key(msg_client, ser_key); memset(buff, '\0', BUFSIZ); index += sprintf(buff, "HTTP/1.1 101 Switching Protocols\r\n"); index += sprintf(buff + index, "Upgrade: websocket\r\n"); index += sprintf(buff + index, "Connection: Upgrade\r\n"); index += sprintf(buff + index, "Sec-WebSocket-Accept: %s\r\n\r\n", ser_key); printf("server: %s", buff); send(client_fd, buff, index, MSG_DONTWAIT); return 0; } //read client mseage head, than read data... int receive_and_parse(int fd) { char *data; char tmp; int index = 0; int nbytes = 0; websocket_head head; //get fin and opencode if (read(fd, &tmp, 1) <= 0) return -1; head.fin = ((tmp & 0x80) == 0x80); head.opencode = (tmp & 0x0F); // get mask and plyload length if (read(fd, &tmp, 1) <= 0) return -1; head.mask = ((tmp & 0x80) == 0X80); head.payload_length = (tmp & 0x7F); // the next two bytes is true payload length if (head.payload_length == 126) { char extern_len[2]; if (read(fd, extern_len, 2) <= 0) return -1; head.payload_length = (extern_len[0] & 0xff) << 8 | (extern_len[1] & 0xff); } // the next eight bytes is the true payload length else if (head.payload_length == 127) { char extern_len[8]; if (read(fd, &extern_len, 8) <= 0) return -1; strreverse(extern_len, 8); memcpy(&(head.payload_length), extern_len, 8); } //处理特殊帧 if (head.opencode == 8) //断开连接 { char packet[2]; packet[0] = 0x88; packet[1] = 0x00; send(fd, packet, 2, MSG_DONTWAIT); return -2; } else if (head.opencode == 0xA) //pong { return -3; } // get masking key if (read(fd, head.masking_key, 4) <= 0) return -1; data = malloc(sizeof(char) * head.payload_length); memset(data, 0, head.payload_length); nbytes = read(fd, data, head.payload_length); for (index = 0; index < nbytes; index++) *(data + index) ^= *(head.masking_key + (index % 4)); //t测试 printf("client: %s\n", data); free(data); return 0; } /*发送数据*/ int send_message(int clt_fd, const char *message, size_t length) { char *packet; int packet_length; if (length < 126) { packet = malloc(sizeof(char) * (2 + length)); memset(packet, 0, sizeof(char) * (2 + length)); packet[0] = 0x81; packet[1] = 0x7F & length; packet_length = length + 2; memcpy(packet + 2, message, length); } else if (length < 0xffff) { packet = malloc(sizeof(char) * (4 + length)); packet[0] = 0x81; packet[1] = 126; packet[2] = (length >> 8 & 0xff); packet[3] = (length & 0xff); packet_length = length + 4; memcpy(packet + 4, message, length); } else { packet = malloc(sizeof(char) * (12 + length)); packet[0] = 0x81; packet[1] = 127; memcpy(packet + 2, &length, sizeof(unsigned long long)); strreverse(packet + 2, sizeof(unsigned long long)); packet_length = length + 12; memcpy(packet + 12, message, length); } if (send(clt_fd, packet, packet_length, MSG_DONTWAIT) == -1) { perror("write error"); return -1; } free(packet); return 0; } /*服务器工作函数*/ /*void worker(int ser_fd) { int epollfd, nfds, index; int msg_lenth; int clt_fd; pid_t pid; char data[BUFSIZ] = { '\0' }; char flags[BUFSIZ] = { '\0' }; //mark client shake hand or not pthread_t thread; pthread_attr_t attr; struct epoll_event events[MAXEVENTS], ev; pthread_attr_init(&attr); pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); if ((epollfd = epoll_create(1)) == -1) { perror("epoll_create error: "); return; } ev.events = EPOLLIN; ev.data.fd = ser_fd; if (epoll_ctl(epollfd, EPOLL_CTL_ADD, ser_fd, &ev) == -1) { perror("epoll_ctl add server's fd error:"); return; } while (1) { if ((nfds = epoll_wait(epollfd, events, MAXEVENTS, -1)) == -1) {