基于net模块，从零实现websocket(ws模块)

Posted 2022-04-18 coderlin_

Websocket

• Websokcet是H5开始提供的一种浏览器与服务器进行全双工通讯的网络技术
• 通俗的讲，就是在客户端和服务器有一个持久的链接，两边可以在任意时间开始发送数据。
• 属于应用层协议，它是基于TCP传输协议的，并复用HTTP的握手通道。

websocket连接

• websocket服用了http的握手通道。具体就是，客户端通过http请求，与websocket服务端协商升级协议，协议升级完毕之后，后续的数据交换则遵守Websocket的协议。
  
  
  
• 请求体中Upgrade表示升级为websocket协议,Scc-WebSocket-Key发送key，Sec-WebSocket-Version表示版本号
• 响应体中，Connection表示同意升级，Sec-WebSocket-Accept是根据客户端发送的key计算出来的。
  
  先走HTTP协议，再走webSocket协议。

客户端，申请协议升级

• 首先客户端发起协议升级请求
• 请求采用的是标准的HTTP报文格式，且只支持GET方法.

GET ws://localhost:8080/ HTTP/1.1
Connection: Upgrade
Host: localhost:8080
Origin: http://localhost:3000
Sec-WebSocket-Key: hGDHklcMkYR51H2A17ikxw== //提供key，服务端根据key计算出sec-websocket-accept,提供基本的防护。比u防止恶意连接。
Sec-WebSocket-Version: 13 //版本
Upgrade: websocket //表示升级协议

服务端：响应协议升级

HTTP/1.1 101 Switching Protocols
Connection: Upgrade
Sec-WebSocket-Accept: vbhpq1GJDvx0/yZjL+KV8LArKlo=
Upgrade: websocket

状体码101表示协议切换。

Sec-Websocket-Accept计算

const crypto = require('crypto')
const CODE = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" //固定的code
function toAcceptKey(wsKey)
    return crypto.createHash('sha1').update(wsKey+CODE).digest('base64')

数据帧格式

• websocket客户端，服务端通信的最小单位是帧，由1个或者多个帧组成一条完整的消息。
• 发送端将消息切割成多个帧，并发送给服务端。
• 接收端，接收消息帧，并将关联的帧重新组装成完成的信息。
  

bit和byte

• 1比特就是一个位（bit），位是数据存储的最小单元。
• 一个字节，也就是一个byte，就等于8个bit。
• 一个英文字母是1字节
• 一个中文汉字是2字节。

位运算符

// 按位 与& 两个数同为1，结果才是1
const A = 0b11110000;
const B = 0b00001111;
const C = 0b11110000
console.log((A & C).toString(2));  11110000

// 按位 或|  两个数只要有一个为1，结果就是1
console.log((A | B).toString(2));   11111111
//按位 或^ 两个数有一个不同，结果就是1，都相同，结果就是0
console.log((A ^ B).toString(2));  11111111

一个数据帧格式

• 单位是bit，比如FIN RSV1各占1比特,opcode占4比特
  
• FIN，1个比特，如果是1，表示这是所有分片中最后一个分片了，如果是0，就表示后续还有分片。
• RSV ，各占1比特，保留位。一般是0
• opcode 4个比特，操作代码，由4个比特,.每一个值都有特殊的含义。
  
  这里主要注意%x1文本帧，%x2二进制帧。
• Mask: 1个比特，这个是指明“payload data”是否被计算掩码，客户端发送给服务端的时候需要掩码操作，Mask为1，并且有一个Masking-key。服务端发送给客户端的数据，Mask为0，不需要掩码。
• Payload len，数据的长度
• Masking-key：0或4给二字节。所有从客户端传送到服务端的数据帧，数据载荷都做了掩码操作，MASK为1，且携带了Masking-key。如果MASK为0，则表示没有Masking-key
• Payload data，帧真正要发送的数据，可以是任意长度，但尽管理论上帧的大小没有限制，但发送的数据不能太大，否则会导致无法高效利用网络带宽，正如上面所说Websocket提供分片。

Payload len

数据的长度计算，也是特殊的。
如

• pl = x,为0-125的时候，数据的长度就是x字节。
• pl = x，为126的时候，后续两个字节代表一个16位的无符号整数。该无符号整数的值为数据的长度。
• pl = x 为127，后续8个字节代表一个64的无符号整数。该无符号整数的值为数据的长度。
• pl如果占用了多个字节，pl的二进制表达采用网络序(bg endian,重要的位在前。)
• 大端序和小端序。大端序在前，小端序在后。
  
  

function getLength(buffer) 
  const byte = buffer.readUInt8(0); //Buffer對象讀取一個無符號的8位整數，0表示从0开始读取。就是将一个字节转为10进制，变成10进制
  const str = byte.toString(2); //变成2进制
  // 截取掉第一位, mark标记
  let length = parseInt(str.substring(1), 2);

  // 如果x < 125，那么x就是数据的长度，
  if (length < 125) 
   else if (length === 126) 
    // 126的话 后续两个字节代表一个16位的无符号整数。该无符号整数的值为数据的长度。
    length = buffer.readUInt16BE(1); //跳过第一位开始读取大端序16位整数，也就是参数后面两个 0b00000000, 0b00000001
   else 
    //最大值就只有127了，因为2的7次方-1是127，而payload length只有7个字节。
    // 后续8个字节代表一个64的无符号整数。该无符号整数的值为数据的长度。
    length = buffer.readBigUInt64BE(1);
  
  return length;


console.log(getLength(Buffer.from([0b00011110, 0b00000000, 0b0000001])));

getLength接受一个buffer，他会去掉前一个bit，也就是Mark标记，然后计算7个bit的长度，根据长度计算出payload length的数值。

掩码算法

当Mark为1的时候，就需要进行掩码操作。

• 掩码键也就是Masking-key是由客户端挑选出来的，32bit的随机数，掩码操作不会影响数据长度。
• 掩码和反掩码操作都采用了如下算法：对索引i模以4得到结果，并对原来的索引进行异或操作(简答地说就是每个4数比较）。
  如

数据： 				0101
masking-key: 		1010  
结果就是：			1111	(数据 异或^ masking-key)
masking-key: 		1010
还原:				0101	(结果 异或^ masking-key)

function unmask(buffer, mask) 
  for (let i = 0; i < buffer.length; i++) 
    buffer[i] ^= mask[i % 4];
  
  return buffer


const mask = Buffer.from([1, 0, 1, 0]);
const buffe1r = Buffer.from([0,1,0,1,0,1,0,1]);
console.log(unmask(buffe1r, mask));

实现ws模块。

• 基础TCP传输层协议，实现一个websocket服务器。

第一步，升级协议

class Server extends EventEmitter 
  constructor(options) 
    super(options);
    this.options = options;
    //创建TCP服务器，只管消息传输，不管解析
    this.server = net.createServer(this.listener);
    this.port = options.port || 8888;
    this.server.listen(this.port, () => 
      console.log(`the server is running $this.port`);
    );
  
  listener = (socket) => 
    // socket是一个套接字，就是用它来发送和接收消息。
    //保持长连接
    socket.setKeepAlive(true);
    // 实现ws的message和send事件
    socket.on("data", (chunk) => 
      chunk = chunk.toString();
      // Chunk是包含请求头和请求体的字符串
      /**
       * chunk 就是
       * GET / HTTP/1.1\\r\\n
       * Host: xxx \\r\\n
       * ...
       * Sec-WebSocket-Key: xxx \\r\\n
       * .... \\r\\n
       * \\r\\n
       *
       * 请求体
       */
      // 接手到客户端发送给服务器的数据之后
      //如果有Upgrade: websocket,表示客户端需要升级协议
      if (chunk.toString().match(/Upgrade: websocket/)) 
        this.upgradeProtocol(socket, chunk);
      
    );

    //连接成功之后触发connection事件，并且传递socket对象
    this.emit("connection", socket);
  ;

  // 升级协议
  upgradeProtocol = (socket, chunk) => 
    const rows = chunk.split("\\r\\n").slice(1, -2); //第一行不要，最后一个不要
    const headers = toHeaders(rows); //获取请求头对象
    const wsKey = headers["Sec-WebSocket-Key"];
    const acceptKey = toAcceptKey(wsKey);
    /** 响应体
        HTTP/1.1 101 Switching Protocols\\r\\n
        Connection: Upgrade\\r\\n
        Sec-WebSocket-Accept: vbhpq1GJDvx0/yZjL+KV8LArKlo=\\r\\n
        Upgrade: websocket\\r\\n
       */
    const response = [
      "HTTP/1.1 101 Switching Protocols",
      "Connection: Upgrade",
      `Sec-WebSocket-Accept: $acceptKey`,
      "Upgrade: websocket",
      "testHeader: myWs",
      "\\r\\n",
    ].join("\\r\\n");
    socket.write(response); //返回给客户端
  ;

通过net模块创建TCP连接，然后根据请求头是否包含Upgrade决定要不要升级协议，如果升级，调用upgradeProtocol，解析请求体，获取key，然后制造响应体，返回。结果：

正常连接成功，协议升级完毕。

第二部，解析ws数据帧，触发sockek.on(‘message’)事件

socket.on("data", (chunk) => 
      const firstChunk = chunk.toString();
      // firstChunk是包含请求头和请求体的字符串
      /**
       * chunk 就是
       * GET / HTTP/1.1\\r\\n
       * Host: xxx \\r\\n
       * ...
       * Sec-WebSocket-Key: xxx \\r\\n
       * .... \\r\\n
       * \\r\\n
       *
       * 请求体
       */
      // 接手到客户端发送给服务器的数据之后
      //如果有Upgrade: websocket,表示客户端需要升级协议
      if (firstChunk.match(/Upgrade: websocket/)) 
        this.upgradeProtocol(socket, firstChunk);
       else 
        // 如果没有，就是已经建立了ws连接，正常传送数据了
        this.onmessage(socket, chunk);
      
    );

如果没有Upgrade,表示已经建立起了ws连接，需要处理数据帧了。

 onmessage = (socket, chunk) => 
    // ws通信单位是数据帧,开始解析
    // 与是两个都是1才会是1，如果第一个是1，那么结果是0b00000000，如果第一个是0，那么结果是 0b00000001
    let FIN = (chunk[0] & 0b10000000) === 0b10000000;
    const opcode = chunk[0] & 0b00001111; // 取出后四位，得到操作码的十进制
    const masked = (chunk[1] & 0b10000000) === 0b10000000; //是否需要掩码
    const payloadLenght = chunk[1] & 0b01111111; // 取出后7为数据长度
    let payload;
    if (masked) 
      // 假设payloadLength长度小于126
      const maskingKey = chunk.slice(2, 6);
      payload = chunk.slice(6, 6 + payloadLenght);
      payload = unmask(payload, maskingKey); //经过反掩码拿到真实数据
     else 
      payload = chunk.slice(6, 6 + payloadLenght);
    
    if (FIN) 
      //如果为结束帧
      switch (opcode) 
        case 1:
          // 文本
          socket.emit("message", payload.toString("utf8"));
          break;
        case 2:
          socket.emit("message", payload);
          // 二进制帧
          break;
        default:
          break;
      
    
  ;

这里只对文本和二进制做处理，如上，先解析数据帧，得到对应的字段，然后通过判断，反掩码等等，获取到最终客户端发送的payload，然后触发message事件。
效果：

// 服务端
wsServer.on("connection", (socket) => 
  //socket是套接字;
  socket.on("message", (message) => 
    console.log("客户端发送的message", message);
  );
);

客户端

正常接收到了。

第三步，实现socket.send事件。

send事件会发送数据给客户端，所以需要自己拼接数据帧。

    socket.send = function (payload) 
      let opcode = Buffer.isBuffer(payload) ? 2 : 1;
      payload = Buffer.isBuffer(payload) ? payload : Buffer.from(payload);
      let length = payload.length;
      let buffer = Buffer.alloc(length + 2); //响应体长度，需要再加2个字节是因为，数据帧固定前面有两个字节做标识。就是FIN,opcode那些
      buffer[0] = 0b10000000 | opcode; // 第一个字节的内容共，FIN+opcode
      buffer[1] = length; //第二个字节放长度，因为第二个字节的第一个bit是Mark，服务端发送的mark为0.
      payload.copy(buffer, 2); // 将payload拷贝到buffer的第二位开始
      console.log('buffer',buffer.toString());
      socket.write(buffer);
    ;

• 创建数据帧较为容易，因为不需要MaskingKey，也不需要Mark，更不需要根据payloadLength考虑后面几个字节的事情。
• 直接创建一个buffer，拼死FIN和opcode，加上length，再拼上数据即可返回。效果：
  
  
  客户端正常收到。

ws全部代码:

const net = require("net"); //实现TCP协议的模块

const  EventEmitter  = require("events");
const 
  unmask,
  getLength,
  toAcceptKey,
  toHeaders,
 = require("../toAcceptKey.js");
const crypto = require("crypto");

// node很多库都是继承EventEmitter，实现事件订阅发布，解耦
class Server extends EventEmitter 
  constructor(options) 
    super(options);
    this.options = options;
    //创建TCP服务器，只管消息传输，不管解析
    this.server = net.createServer(this.listener);
    this.port = options.port || 8888;
    this.server.listen(this.port, () => 
      console.log(`the server is running $this.port`);
    );
  
  listener = (socket) => 
    // socket是一个套接字，就是用它来发送和接收消息。
    //保持长连接
    socket.setKeepAlive(true);
    // 实现ws的message和send事件
    socket.on("data", (chunk) => 
      const firstChunk = chunk.toString();
      // firstChunk是包含请求头和请求体的字符串
      /**
       * chunk 就是
       * GET / HTTP/1.1\\r\\n
       * Host: xxx \\r\\n
       * ...
       * Sec-WebSocket-Key: xxx \\r\\n
       * .... \\r\\n
       * \\r\\n
       *
       * 请求体
       */
      // 接手到客户端发送给服务器的数据之后
      //如果有Upgrade: websocket,表示客户端需要升级协议
      if (firstChunk.match(/Upgrade: websocket/)) 
        this.upgradeProtocol(socket, firstChunk);
       else 
        // 如果没有，就是已经建立了ws连接，正常传送数据了
        this.onmessage(socket, chunk);
      
    );

    //连接成功之后触发connection事件，并且传递socket对象
    this.emit("connection", socket);

    socket.send = function (payload) 
      let opcode = Buffer.isBuffer(payload) ? 2 : 1;
      payload = Buffer.isBuffer(payload) ? payload : Buffer.from(payload);
      let length = payload.length;
      let buffer = Buffer.alloc(length + 2); //响应体长度，需要再加2个字节是因为，数据帧固定前面有两个字节做标识。就是FIN,opcode那些
      buffer[0] = 0b10000000 | opcode; // 第一个字节的内容共，FIN+opcode
      buffer[1] = length; //第二个字节放长度，因为第二个字节的第一个bit是Mark，服务端发送的mark为0.
      payload.copy(buffer, 2); // 将payload拷贝到buffer的第二位开始
      console.log('buffer',buffer.toString());
      socket.write(buffer);
    ;
  ;

  onmessage = (socket, chunk) => 
    // ws通信单位是数据帧,开始解析
    // 与是两个都是1才会是1，如果第一个是1，那么结果是0b00000000，如果第一个是0，那么结果是 0b00000001
    let FIN = (chunk[0] & 0b10000000) === 0b10000000;
    const opcode = chunk[0] & 0b00001111; // 取出后四位，得到操作码的十进制
    const masked = (chunk[1] & 0b10000000) === 0b10000000; //是否需要掩码
    const payloadLenght = chunk[1] & 0b01111111; // 取出后7为数据长度
    let payload;
    if (masked) 
      // 假设payloadLength长度小于126
      const maskingKey = chunk.slice(2, 6);
      payload = chunk.slice(6, 6 + payloadLenght);
      payload = unmask(payload, maskingKey); //经过反掩码拿到真实数据
     else 
      payload = chunk.slice(6, 6 + payloadLenght);
    
    if (FIN) 
      //如果为结束帧
      switch (opcode) 
        case 1:
          // 文本
          socket.emit("message", payload.toString("utf8"));
          break;
        case 2:
          socket.emit("message", payload);
          // 二进制帧
          break;
        default:
          break;
      
    
  ;

  // 升级协议
  upgradeProtocol = (socket, chunk) => 
    const rows = chunk.split("\\r\\n").slice(1, -2); //第一行不要，最后一个不要
    const headers = toHeaders(rows); //获取请求头对象
    const wsKey = headers["Sec-WebSocket-Key"];
    const acceptKey = toAcceptKey(wsKey);
    /** 响应体
        HTTP/1.1 101 Switching Protocols\\r\\n
        Connection: Upgrade\\r\\n
        Sec-WebSocket-Accept: vbhpq1GJDvx0/yZjL+KV8LArKlo=\\r\\n
        Upgrade: websocket\\r\\n
       */
    const response = [
      "HTTP/1.1 101 Switching Protocols",
      "Connection: Upgrade",
      `Sec-WebSocket-Accept: $acceptKey`,
      "Upgrade: websocket",
      "testHeader: myWs", //用于测试的头部
      "\\r\\n",
    ].join("\\r\\n");
    socket.write(response); //返回给客户端
  ;


module.exports =  Server ;

代码仓库在：https://gitee.com/fine509/websocket