首页 > 代码库 > engine.io分析2--socket.io的基石

engine.io分析2--socket.io的基石

2024-08-07 23:25:46   218人阅读

转载请注明： TheViper http://www.cnblogs.com/TheViper

源码分析

var engine = require(‘./node_modules/engine.io/lib/engine.io.js‘);var server = engine.listen(8000,{    transports:[‘polling‘]});server.on(‘connection‘, function(socket){  socket.send(‘utf 8 string‘);});

之前提到过engine.io只有websocket,polling两种传输方式。对于websocket，engine.io用了第三方库ws,这个从源码目录看的出来，调用很简单。所以本文只说polling这种方式.

另外精力有限，所以只分析服务端，就当是抛砖引玉，源码比较简单。

首先是加载engine.io.js.

exports = module.exports = function() {  // backwards compatible use as `.attach`  // if first argument is an http server  if (arguments.length && arguments[0] instanceof http.Server) {    return attach.apply(this, arguments);  }  // if first argument is not an http server, then just make a regular eio server  return exports.Server.apply(null, arguments);};

注释上说的很清楚，if是为了兼容以前的版本，判断传入的是不是http.然后初始化Server.

Server初始化只是将参数写入，没有指定参数的，使用默认参数，还有就是定义后面用到的变量。最后，如果参数方式中有websocket,就初始化WebSocketServer。

  if (~this.transports.indexOf(‘websocket‘)) {    this.ws = new WebSocketServer({ noServer: true, clientTracking: false });  }

接着是engine.listen().

function listen(port, options, fn) {  if (‘function‘ == typeof options) {    fn = options;    options = {};  }  var server = http.createServer(function (req, res) {    res.writeHead(501);    res.end(‘Not Implemented‘);  });  server.listen(port, fn);  // create engine server  var engine = exports.attach(server, options);  engine.httpServer = server;  return engine;};

先判断是不是重载的方法，如果是，则没有指定端口，使用默认端口，并且参数移位。后面是将我这种定义方式，变成

var engine = require(‘engine.io‘);var http = require(‘http‘).createServer().listen(3000);var server = engine.attach(http);

这是文档上另一种定义方式。

接着是attach()

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

转到server里面的attach(),里面做的事就比较多了。简单说就是

  // cache and clean up listeners  var listeners = server.listeners(‘request‘).slice(0);  server.removeAllListeners(‘request‘);  server.on(‘close‘, self.close.bind(self));  // add request handler  server.on(‘request‘, function(req, res){    if (check(req)) {      debug(‘intercepting request for path "%s"‘, path);      self.handleRequest(req, res);    } else {      for (var i = 0, l = listeners.length; i < l; i++) {        listeners[i].call(server, req, res);      }    }  });

清空监听器，有请求传入时用server.on(‘request‘)监听。

  function check (req) {    return path == req.url.substr(0, path.length);  }

一般情况下，check()返回的是true,搞不懂什么时候返回false.看后面好像是负载均衡，需要配置多个监听的时候用。不明白！

接着是self.handleRequest(req, res);

Server.prototype.handleRequest = function(req, res){  debug(‘handling "%s" http request "%s"‘, req.method, req.url);  this.prepare(req);  req.res = res;  var self = this;  this.verify(req, false, function(err, success) {    if (!success) {      sendErrorMessage(req, res, err);      return;    }    if (req._query.sid) {      debug(‘setting new request for existing client‘);      self.clients[req._query.sid].transport.onRequest(req);    } else {      self.handshake(req._query.transport, req);    }  });};

this.prepare(req)是将请求里面的查询参数封装到req._query上。然后是verify()

Server.prototype.verify = function(req, upgrade, fn){  // transport check  var transport = req._query.transport;  if (!~this.transports.indexOf(transport)) {    debug(‘unknown transport "%s"‘, transport);    return fn(Server.errors.UNKNOWN_TRANSPORT, false);  }  // sid check  var sid = req._query.sid;  if (sid) {    if (!this.clients.hasOwnProperty(sid))      return fn(Server.errors.UNKNOWN_SID, false);    if (!upgrade && this.clients[sid].transport.name !== transport) {      debug(‘bad request: unexpected transport without upgrade‘);      return fn(Server.errors.BAD_REQUEST, false);    }  } else {    // handshake is GET only    if (‘GET‘ != req.method) return fn(Server.errors.BAD_HANDSHAKE_METHOD, false);    if (!this.allowRequest) return fn(null, true);    return this.allowRequest(req, fn);  }  fn(null, true);};

记下请求中要求的传输方式，例如http://localhost:3000/socket.io/?EIO=3&transport=polling&t=1418545776090-1&sid=M5e5hB6NAOmh7YW1AAAB。可以看到由transport参数指定。然后记下sid,这个请求中也有。注意，第一次请求的时候，由于没有实例socket,所以没有socket的id。这时判断有没有sid,确定是不是第一次请求。

下面我们假定是第一次请求，然后走else.

 if (‘GET‘ != req.method),这里要判断下，因为engine.io会通过请求的方法的不同确定后面的执行流程，默认get方法维持长连接，当然发出握手请求的也是get方法，这个后面会说到。

 this.allowRequest像是拦截器，拦截握手请求，和传输升级的请求（websocket）.

例子没有设置allowRequest方法，就直接fn(null,true),进入回调函数.

  this.verify(req, false, function(err, success) {    if (!success) {      sendErrorMessage(req, res, err);      return;    }    if (req._query.sid) {      debug(‘setting new request for existing client‘);      self.clients[req._query.sid].transport.onRequest(req);    } else {      self.handshake(req._query.transport, req);    }  });

如果是第一次请求，后面执行握手；如果不是，则选定已经存在的socket,调用socket上面的transport(传输方式，已经绑定到socket)的onRequest方法，这个后面会说到。现在我们还是第一次请求，所以开始握手。

Server.prototype.handshake = function(transport, req){  var id = base64id.generateId();  debug(‘handshaking client "%s"‘, id);  var transportName = transport;  try {    var transport = new transports[transport](req);    if (‘polling‘ == transportName) {      transport.maxHttpBufferSize = this.maxHttpBufferSize;    }    if (req._query && req._query.b64) {      transport.supportsBinary = false;    } else {      transport.supportsBinary = true;    }  }  catch (e) {    sendErrorMessage(req, req.res, Server.errors.BAD_REQUEST);    return;  }  var socket = new Socket(id, this, transport, req);  var self = this;  if (false !== this.cookie) {    transport.on(‘headers‘, function(headers){      headers[‘Set-Cookie‘] = self.cookie + ‘=‘ + id;    });  }  transport.onRequest(req);  this.clients[id] = socket;  this.clientsCount++;  socket.once(‘close‘, function(){    delete self.clients[id];    self.clientsCount--;  });  this.emit(‘connection‘, socket);};

 var transport = new transports[transport](req);这里的transports是transports = require(‘./transports‘)。加载的是transprots文件夹下面的index.js,不是transprots.js.

var XHR = require(‘./polling-xhr‘);var JSONP = require(‘./polling-jsonp‘);module.exports = exports = {  polling: polling,  websocket: require(‘./websocket‘)};exports.polling.upgradesTo = [‘websocket‘];function polling (req) {  if (‘string‘ == typeof req._query.j) {    return new JSONP(req);  } else {    return new XHR(req);  }}

由于transport由请求中的transprot参数决定，这里是polling。所以会执行polling(req),再看请求参数里有没有j,最终确定具体的传输方式。这里是xhr.

function XHR(req){  Polling.call(this, req);}XHR.prototype.__proto__ = Polling.prototype;

这里XHR继承了Polling.然后实例化Polling.

function Polling (req) {  Transport.call(this, req);}Polling.prototype.__proto__ = Transport.prototype;

又去实例化Transport.Transport继承EventEmitter。

 回到主线，var socket = new Socket(id, this, transport, req);这里开始实例化socket了。socket里面做了什么，后面会说到。然后在header上设置cookie,值是上面产生的随机id.方便开发者做权限控制。

接着是transport.onRequest(req); 进入polling-xhr.js

XHR.prototype.onRequest = function (req) {  if (‘OPTIONS‘ == req.method) {    var res = req.res;    var headers = this.headers(req);    headers[‘Access-Control-Allow-Headers‘] = ‘Content-Type‘;    res.writeHead(200, headers);    res.end();  } else {    Polling.prototype.onRequest.call(this, req);  }};

 这里会先判断请求方法是不是options,这个好像只有高级浏览器中才会出现。

第一次请求是get方法，走后面。

Polling.prototype.onRequest = function (req) {  var res = req.res;    if (‘GET‘ == req.method) {    this.onPollRequest(req, res);  } else if (‘POST‘ == req.method) {    this.onDataRequest(req, res);  } else {    res.writeHead(500);    res.end();  }};

又对请求方法进行判断。从方法名字就看的出，get方法用来维持长连接的，当然广播返回的数据也是通过这个get方法。post方法用来传输客户端的数据，比如客户端聊天的文字，如果没有数据，就会发出get请求。

我们还在握手请求，get方法。

Polling.prototype.onPollRequest = function (req, res) {  if (this.req) {    debug(‘request overlap‘);    // assert: this.res, ‘.req and .res should be (un)set together‘    this.onError(‘overlap from client‘);    res.writeHead(500);    return;  }  debug(‘setting request‘);  this.req = req;  this.res = res;  var self = this;  function onClose () {    self.onError(‘poll connection closed prematurely‘);  }  function cleanup () {    req.removeListener(‘close‘, onClose);    self.req = self.res = null;  }  req.cleanup = cleanup;  req.on(‘close‘, onClose);  this.writable = true;  this.emit(‘drain‘);  // if we‘re still writable but had a pending close, trigger an empty send  if (this.writable && this.shouldClose) {    debug(‘triggering empty send to append close packet‘);    this.send([{ type: ‘noop‘ }]);  }};

onPollRequest（）主要是对相应对象，如request,绑定事件，回调函数，后面传输数据时会用到。

回到前面看实例化socket,然后做了什么。

function Socket (id, server, transport, req) {  this.id = id;  this.server = server;  this.upgraded = false;  this.readyState = ‘opening‘;  this.writeBuffer = [];  this.packetsFn = [];  this.sentCallbackFn = [];  this.request = req;  // Cache IP since it might not be in the req later  this.remoteAddress = req.connection.remoteAddress;  this.checkIntervalTimer = null;  this.upgradeTimeoutTimer = null;  this.pingTimeoutTimer = null;  this.setTransport(transport);  this.onOpen();}

变量初始化后，this.setTransport(transport);

Socket.prototype.setTransport = function (transport) {  this.transport = transport;  this.transport.once(‘error‘, this.onError.bind(this));  this.transport.on(‘packet‘, this.onPacket.bind(this));  this.transport.on(‘drain‘, this.flush.bind(this));  this.transport.once(‘close‘, this.onClose.bind(this, ‘transport close‘));  //this function will manage packet events (also message callbacks)  this.setupSendCallback();};

对前面实例化的transport绑定事件，回调函数。然后是onOpen()

Socket.prototype.onOpen = function () {  this.readyState = ‘open‘;  // sends an `open` packet  this.transport.sid = this.id;  this.sendPacket(‘open‘, JSON.stringify({      sid: this.id    , upgrades: this.getAvailableUpgrades()    , pingInterval: this.server.pingInterval    , pingTimeout: this.server.pingTimeout  }));  this.emit(‘open‘);  this.setPingTimeout();};

sendPacket()开始传输针对握手请求的响应，然后是设置超时定时器。

Socket.prototype.setPingTimeout = function () {  var self = this;  clearTimeout(self.pingTimeoutTimer);  self.pingTimeoutTimer = setTimeout(function () {    self.onClose(‘ping timeout‘);  }, self.server.pingInterval + self.server.pingTimeout);};

回到sendPacket()

Socket.prototype.sendPacket = function (type, data, callback) {  if (‘closing‘ != this.readyState) {    debug(‘sending packet "%s" (%s)‘, type, data);    var packet = { type: type };    if (data) packet.data =http://www.mamicode.com/ data;    // exports packetCreate event    this.emit(‘packetCreate‘, packet);    this.writeBuffer.push(packet);    //add send callback to object    this.packetsFn.push(callback);    this.flush();  }};

this.writeBuffer是暂时保存响应数据的缓冲。packet是用来包装响应数据的。然后flush输出。

Socket.prototype.flush = function () {  if (‘closed‘ != this.readyState && this.transport.writable    && this.writeBuffer.length) {    debug(‘flushing buffer to transport‘);    this.emit(‘flush‘, this.writeBuffer);    this.server.emit(‘flush‘, this, this.writeBuffer);    var wbuf = this.writeBuffer;    this.writeBuffer = [];    if (!this.transport.supportsFraming) {      this.sentCallbackFn.push(this.packetsFn);    } else {      this.sentCallbackFn.push.apply(this.sentCallbackFn, this.packetsFn);    }    this.packetsFn = [];    this.transport.send(wbuf);    this.emit(‘drain‘);    this.server.emit(‘drain‘, this);  }};

触发各种绑定的事件，最后调用this.transport.send(wbuf);

Polling.prototype.send = function (packets) {  if (this.shouldClose) {    debug(‘appending close packet to payload‘);    packets.push({ type: ‘close‘ });    this.shouldClose();    this.shouldClose = null;  }  var self = this;  parser.encodePayload(packets, this.supportsBinary, function(data) {    self.write(data);  });};

对buffer解析后，write()

Polling.prototype.write = function (data) {  debug(‘writing "%s"‘, data);  this.doWrite(data);  this.req.cleanup();  this.writable = false;};

doWrite()由子类实现。

XHR.prototype.doWrite = function(data){  // explicit UTF-8 is required for pages not served under utf  var isString = typeof data =http://www.mamicode.com/= ‘string‘;  var contentType = isString    ? ‘text/plain; charset=UTF-8‘    : ‘application/octet-stream‘;  var contentLength = ‘‘ + (isString ? Buffer.byteLength(data) : data.length);  var headers = {    ‘Content-Type‘: contentType,    ‘Content-Length‘: contentLength  };  // prevent XSS warnings on IE  // https://github.com/LearnBoost/socket.io/pull/1333  var ua = this.req.headers[‘user-agent‘];  if (ua && (~ua.indexOf(‘;MSIE‘) || ~ua.indexOf(‘Trident/‘))) {    headers[‘X-XSS-Protection‘] = ‘0‘;  }  this.res.writeHead(200, this.headers(this.req, headers));  this.res.end(data);};

engine.io分析2--socket.io的基石


联系
我们