engine.io原理分析

exports = module.exports = function() {
  if (arguments.length && arguments[0] instanceof http.Server) {
    return attach.apply(this, arguments);
  }
  return new Server(arguments);
};

function attach(server, options) {
  const engine = new Server(options);
  engine.attach(server, options);
  return engine;
}

constructor(opts = {}) {
    // 忽略一系列参数处理
    this.init();
  }

init() {
    const wsModule = this.opts.wsEngine === "ws" ? require("ws") : require(this.opts.wsEngine);
    this.ws = new wsModule.Server({
      // 不使用websocket的服务器能力
      noServer: true,
      clientTracking: false,
      perMessageDeflate: this.opts.perMessageDeflate,
      maxPayload: this.opts.maxHttpBufferSize
    });
  }

attach(server, options) {
    const self = this;
    // 判断是不是自己的请求
    function check(req) {
      return path === req.url.substr(0, path.length);
    }
    // 监听upgrade事件
    server.on("upgrade", function(req, socket, head) {
        // 判断是不是自己的请求
        if (check(req)) {
          self.handleUpgrade(req, socket, head);
        }
  }

// 处理websocket协议升级请求
  handleUpgrade(req, socket, upgradeHead) {
    // 解析query参数
    this.prepare(req);

    const self = this;
    this.verify(req, true, function(err, success) {
      if (!success) {
        abortConnection(socket, err);
        return;
      }

      const head = Buffer.from(upgradeHead); // eslint-disable-line node/no-deprecated-api
      upgradeHead = null;

      /*
        使用ws模块完成websocket的升级，ws回复同意升级协议后执行onWebSocket,
        conn是一个ws模块的WebSocket对象，见websocke-server的completeUpgrade
      */
      self.ws.handleUpgrade(req, socket, head, function(conn) {
        self.onWebSocket(req, conn);
      });
    });
  }

onWebSocket(req, socket) {
    req.websocket = socket;
    this.handshake(req._query.transport, req);
  }

async handshake(transportName, req) {
    let id = await this.generateId(req);
    // 新建一个通道
    var transport = new transports[transportName](req);
    const socket = new Socket(id, 
                              // 对应的server
                              this, 
                              // 对应的通道，Socket只是封装底层的能力
                              transport, 
                              // 对应的websocket升级请求对象
                              req);
    const self = this;
    transport.onRequest(req);
    // 应用层拿到的socket对象
    this.emit("connection", socket);
  }

constructor(req) {
    super(req);
    // 保存ws模块的websocket对象
    this.socket = req.websocket;
    /*
      注册事件，由ws模块触发，然后再往上层触发，
      message事件是当解析到websocket协议中的数据部分时触发
    */
    this.socket.on("message", this.onData.bind(this));
    this.socket.once("close", this.onClose.bind(this));
    this.socket.on("error", this.onError.bind(this));
    this.socket.on("headers", headers => {
      this.emit("headers", headers);
    });
    this.writable = true;
    this.perMessageDeflate = null;
  }

constructor(id, server, transport, req) {
    super();
    this.id = id;
    // socket关联的服务器
    this.server = server;
    // 状态
    this.readyState = "opening";
    // 待写数据
    this.writeBuffer = [];
    this.packetsFn = [];
    this.sentCallbackFn = [];
    // 关闭数据通道时执行的清除函数集
    this.cleanupFn = [];
    // 对应的request
    this.request = req;
    this.checkIntervalTimer = null;
    // 检测是否完成协议切换的定时器id
    this.upgradeTimeoutTimer = null;
    // 心跳对应的定时器
    this.pingTimeoutTimer = null;
    // 发送ping包后，多久没有收到回复则断开连接对应的定时器id
    this.pingIntervalTimer = null;
    // 设置通信通道
    this.setTransport(transport);
    this.onOpen();
  }

setTransport(transport) {
    const onError = this.onError.bind(this);
    const onPacket = this.onPacket.bind(this);
    const flush = this.flush.bind(this);
    const onClose = this.onClose.bind(this, "transport close");
    // 通道有数据则通知socket等，socket再往上报
    this.transport = transport;
    // 监听websocket.js的对象的事件，通道出错
    this.transport.once("error", onError);
    this.transport.on("packet", onPacket);
    // 触发drain事件则继续发送数据
    this.transport.on("drain", flush);
    // 通道关闭
    this.transport.once("close", onClose);
    // ...
  }

// 发送由long polling升级到websocket的通知包 或者是直接建立websocket通信时发生的包
  onOpen() {
    this.readyState = "open";
    this.transport.sid = this.id;
    this.sendPacket(
      "open",
      JSON.stringify({
        sid: this.id,
        // 可以升级到这个（些）协议
        upgrades: this.getAvailableUpgrades(),
        // 心跳间隔
        pingInterval: this.server.opts.pingInterval,
        // 多久后发送心跳
        pingTimeout: this.server.opts.pingTimeout
      })
    );

    // 触发open事件
    this.emit("open");
    // 开启心跳检测
    this.schedulePing();
  }

this.socket.on("message", this.onData.bind(this));
  // 有数据到来时执行，根据engine.io协议进行解析，解析触发onpacket事件
  onData(data) {
    this.onPacket(parser.decodePacket(data));
  }
  // 对数据解析后执行
  onPacket(packet) {
    this.emit("packet", packet);
  }

onPacket(packet) {
    if ("open" === this.readyState) {
      this.emit("packet", packet);
      // 收到数据包后重置定时器，避免关闭连接
      this.resetPingTimeout(
        this.server.opts.pingInterval + this.server.opts.pingTimeout
      );
      // type见engine.io-parser的decode
      switch (packet.type) {
        // 收到pong之后，等待一段时间后，继续发送ping包
        case "pong":
          this.schedulePing();
          this.emit("heartbeat");
          break;
        // 解析出错，关闭socket
        case "error":

          this.onClose("parse error");
          break;

        case "message":
          // data是上层数据，比如socketio协议的数据包
          this.emit("data", packet.data);
          this.emit("message", packet.data);
          break;
      }
    } 
  }