node index.js

var ws = new WebSocket("ws://127.0.0.1:3000");
ws.onmessage = function(evt) {
  console.log( "Received Message: " + evt.data);
};

ws.send('hello world');

// HTTP服务器部分
var server = http.createServer(function(req, res) {
  res.end('websocket test\r\n');
});

// Upgrade请求处理
server.on('upgrade', function(req, socket, upgradeHead){
  // 初始化 ws
  var ws = new WebSocket(req, socket, upgradeHead);
  
  // ... ws 监听 data、error 的逻辑等

});

class WebSocket extends EventEmitter {
  constructor(req, socket, upgradeHead){
    super(); // 调用 EventEmitter 构造函数
    
    // 1. 构造响应头 resHeaders 部分
    
    // 2. 监听 socket 的 data 事件，以及 error 事件
    
    // 3. 初始化成员属性
    
  }
}

    var resKey = hashWebSocketKey(req.headers['sec-websocket-key']);

    // 构造响应头
    var resHeaders = [
      'HTTP/1.1 101 Switching Protocols',
      'Upgrade: websocket',
      'Connection: Upgrade',
      'Sec-WebSocket-Accept: ' + resKey
    ]
      .concat('', '')
      .join('\r\n');
    socket.write(resHeaders);

    socket.on('data', data => {
      this.buffer = Buffer.concat([this.buffer, data]);
      while (this._processBuffer()) {} // 循环处理返回的 data 数据
    });

    socket.on('close', had_error => {
      if (!this.closed) {
        this.emit('close', 1006);
        this.closed = true;
      }
    });

var opcode = byte1 & 0x0f; //截取第一个字节的后 4 位，即 opcode 码

this._handleFrame(opcode, payload); // 处理操作码

_handleFrame(opcode, buffer) {
    var payload;
    switch (opcode) {
      case OPCODES.TEXT:
        payload = buffer.toString('utf8'); //如果是文本需要转化为utf8的编码
        this.emit('data', opcode, payload); //Buffer.toString()默认utf8 这里是故意指示的
        break;
      case OPCODES.BINARY: //二进制文件直接交付
        payload = buffer;
        this.emit('data', opcode, payload);
        break;
      case OPCODES.PING: // 发送 pong 做响应
        this._doSend(OPCODES.PONG, buffer);
        break;
      case OPCODES.PONG: //不做处理
        console.log('server receive pong');
        break;
      case OPCODES.CLOSE: // close有很多关闭码
        let code, reason; // 用于获取关闭码和关闭原因
        if (buffer.length >= 2) {
          code = buffer.readUInt16BE(0);
          reason = buffer.toString('utf8', 2);
        }
        this.close(code, reason);
        this.emit('close', code, reason);
        break;
      default:
        this.close(1002, 'unhandle opcode:' + opcode);
    }
  }

    var FIN = byte1 & 0x80; // 如果为0x80，则标志传输结束，获取高位 bit
    // 如果是 0 的话，说明是延续帧，需要保存好 opCode
    if (!FIN) {
      this.frameOpcode = opcode || this.frameOpcode; // 确保不为 0;
    }
    
    //....
    // 有可能是分帧，需要拼接数据
    this.frames = Buffer.concat([this.frames, payload]); // 保存到 frames 中

    if (FIN) {
      payload = this.frames.slice(0); // 获取所有拼接完整的数据
      opcode = opcode || this.frameOpcode; // 如果是 0 ，则保持获取之前保存的 code
      this.frames = Buffer.alloc(0); // 清空 frames
      this.frameOpcode = 0; // 清空 opcode
      this._handleFrame(opcode, payload); // 处理操作码
    }

  _doSend(opcode, payload) {
    // 1. 考虑数据分片
    this.socket.write(
      encodeMessage(count > 0 ? OPCODES.CONTINUE : opcode, payload)
    ); //编码后直接通过socket发送

    // ...
    var len = Buffer.byteLength(payload);
    // 分片的距离逻辑
    var count = 0;
    // 这里可以针对 payload 的长度做分片
    while (len > MAX_FRAME_SIZE) {
      var framePayload = payload.slice(0, MAX_FRAME_SIZE);
      payload = payload.slice(MAX_FRAME_SIZE);
      this.socket.write(
        encodeMessage(
          count > 0 ? OPCODES.CONTINUE : opcode,
          framePayload,
          false
        )
      ); //编码后直接通过socket发送
      count++;
      len = Buffer.byteLength(payload);
    }
  // ...

var ws = new WebSocket("ws://127.0.0.1:3000");
ws.onmessage = function(evt) {
  console.log( "Received Message: " + evt.data);
};
var myArray = new ArrayBuffer(131072 * 2 + 1);
ws.send(myArray);

server detect fragment, sizeof payload: 131072
server detect fragment, sizeof payload: 131072
receive data: 2 262145

Received Message: good job