engine.io客户端分析1--socket.io的基石

engine.io客户端分析1--socket.io的基石

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

    var socket = eio('http://localhost:8000');    socket.on('open', function(){        socket.on('message', function(data){            console.log(data);        });        socket.on('close', function(){});    });

源码很简单。

socket.js

function Socket(uri, opts){  if (!(this instanceof Socket)) return new Socket(uri, opts);....  this.open();}

先是各种传递参数，初始化。然后open().

Socket.PRototype.open = function () {    var  transport = this.transports[0];  this.readyState = 'opening';  // Retry with the next transport if the transport is disabled (jsonp: false)  var transport;  try {    transport = this.createTransport(transport);  } catch (e) {    this.transports.shift();    this.open();    return;  }  transport.open();  this.setTransport(transport);};

设置readyState,这个是表示全局状态的变量，值有opening,open,close.然后createTransport()

Socket.prototype.createTransport = function (name) {  debug('creating transport "%s"', name);  var query = clone(this.query);  // append engine.io protocol identifier  query.EIO = parser.protocol;  // transport name  query.transport = name;  // session id if we already have one  if (this.id) query.sid = this.id;  var transport = new transports[name]({    agent: this.agent, 。。。  });  return transport;};

这里我们不考虑websocket.

new transprots[name]在/transports/index.js.

function polling(opts){ ....  xhr = new xmlHttpRequest(opts);  if ('open' in xhr && !opts.forceJSONP) {    return new XHR(opts);  } else {    if (!jsonp) throw new Error('JSONP disabled');    return new JSONP(opts);  }}

如果要强制执行jsonp传输的话，要这样设置。

    var socket = eio('http://localhost:8000',{        forceJSONP:true    });

没有设置的话，默认走xhr(Ajax).如果跨域的话，会发出options请求，请求服务端同意跨域。这里的细节我不明白。

function XHR(opts){  Polling.call(this, opts);....}/** * Inherits from Polling. */inherit(XHR, Polling);

XHR继承Polling,Polling又继承Transport。这里调用构造函数时都调用了父类的构造函数，里面就是初始化了一些参数。

这样createTransport()就走完了。接着是transport.open();

Transport.prototype.open = function () {  if ('closed' == this.readyState || '' == this.readyState) {    this.readyState = 'opening';    this.doOpen();  }  return this;};

Polling.prototype.doOpen = function(){  this.poll();};

Polling.prototype.poll = function(){  debug('polling');  this.polling = true;  this.doPoll();  this.emit('poll');};

XHR.prototype.doPoll = function(){  debug('xhr poll');  var req = this.request();  var self = this;  req.on('data', function(data){    self.onData(data);  });  req.on('error', function(err){    self.onError('xhr poll error', err);  });  this.pollXhr = req;};

doPoll()里对req进行了事件绑定，后面ajax有数据成功返回时会触发上面的data事件，执行onData().this.request()返回的是ajax的封装。

XHR.prototype.request = function(opts){  opts = opts || {};  opts.uri = this.uri();。。。  return new Request(opts);};

function Request(opts){  this.method = opts.method || 'GET';  this.uri = opts.uri; ....  this.create();}

Request.prototype.create = function(){  var opts = { agent: this.agent, xdomain: this.xd, xscheme: this.xs, enablesXDR: this.enablesXDR };  // SSL options for Node.js client  opts.pfx = this.pfx;.........  var xhr = this.xhr = new xmlhttpRequest(opts);  var self = this;  try {    debug('xhr open %s: %s', this.method, this.uri);    xhr.open(this.method, this.uri, this.async);    if (this.supportsBinary) {      // This has to be done after open because Firefox is stupid      // http://stackoverflow.com/questions/13216903/get-binary-data-with-xmlhttprequest-in-a-firefox-extension      xhr.responseType = 'arraybuffer';    }    if ('POST' == this.method) {      try {        if (this.isBinary) {          xhr.setRequestHeader('Content-type', 'application/octet-stream');        } else {          xhr.setRequestHeader('Content-type', 'text/plain;charset=UTF-8');        }      } catch (e) {}    }    // ie6 check    if ('withCredentials' in xhr) {      xhr.withCredentials = true;    }    if (this.hasXDR()) {      xhr.onload = function(){        self.onLoad();      };      xhr.onerror = function(){        self.onError(xhr.responseText);      };    } else {      xhr.onreadystatechange = function(){        if (4 != xhr.readyState) return;        if (200 == xhr.status || 1223 == xhr.status) {          self.onLoad();        } else {          // make sure the `error` event handler that's user-set          // does not throw in the same tick and gets caught here          setTimeout(function(){            self.onError(xhr.status);          }, 0);        }      };    }    debug('xhr data %s', this.data);    xhr.send(this.data);  } catch (e) {    // Need to defer since .create() is called directly fhrom the constructor    // and thus the 'error' event can only be only bound *after* this exception    // occurs.  Therefore, also, we cannot throw here at all.    setTimeout(function() {      self.onError(e);    }, 0);    return;  }};

create()发出ajax请求，这时请求方法是默认的get.前面的文章中说过，服务端会针对不同的请求方法执行不同的策略。这里在创建第一次请求（握手）。

如果响应成功，self.onLoad();，取出数据。onData()

Request.prototype.onLoad = function(){  var data;  try {    var contentType;    try {      contentType = this.xhr.getResponseHeader('Content-Type').split(';')[0];    } catch (e) {}    if (contentType === 'application/octet-stream') {      data = this.xhr.response;    } else {      if (!this.supportsBinary) {        data = this.xhr.responseText;      } else {        data = 'ok';      }    }  } catch (e) {    this.onError(e);  }  if (null != data) {    this.onData(data);  }};

Request.prototype.onSuccess = function(){  this.emit('success');  this.cleanup();};Request.prototype.onData = function(data){  this.emit('data', data);  this.onSuccess();};

this.emit('data', data);会触发前面doPoll（）里面的onData().这里的onData()不是Request里面的onData().

  req.on('data', function(data){    self.onData(data);  });

注意，这里的emit不是服务端的emit,在源码后面可以看到，它相当于jquery里面的fire(),用来触发自定义事件。

Polling.prototype.onData = function(data){  var self = this;  debug('polling got data %s', data);  var callback = function(packet, index, total) {    // if its the first message we consider the transport open    if ('opening' == self.readyState) {      self.onOpen();    }    // if its a close packet, we close the ongoing requests    if ('close' == packet.type) {      self.onClose();      return false;    }    // otherwise bypass onData and handle the message    self.onPacket(packet);  };  // decode payload  parser.decodePayload(data, this.socket.binaryType, callback);  // if an event did not trigger closing  if ('closed' != this.readyState) {    // if we got data we're not polling    this.polling = false;    this.emit('pollComplete');    if ('open' == this.readyState) {      this.poll();    } else {      debug('ignoring poll - transport state "%s"', this.readyState);    }  }};

parser.decodePayload（）和服务端里面的一样的作用，就是解析数据，可以简单的认为它把数据解析成关联数值，然后执行回调。

回调里面，先判断是不是握手响应，根据readystate是不是opening.如果是的话，onOpen()

Transport.prototype.onOpen = function () {  this.readyState = 'open';  this.writable = true;  this.emit('open');};

改变readystate,触发客户端的open事件。

这时readystate是open了，然后是onData()里面的poll().然后执行poll()->doPoll()->request()发出xhr（get方法）请求。

get方法用来做长连接的，当然广播返回的数据也是通过这个get方法。

post方法是向服务端传送心跳的，服务端在一定时间内（pingInterval）收到这个post请求，则认定这个客户端还在。具体的后面会说到。

回到回调函数里面，self.onPacket(packet);

Transport.prototype.onPacket = function (packet) {  this.emit('packet', packet);};

packet绑定和服务端一样，

Socket.prototype.setTransport = function(transport){  // set up transport  this.transport = transport;  // set up transport listeners  transport  .on('drain', function(){    self.onDrain();  })  .on('packet', function(packet){    self.onPacket(packet);  })  .on('error', function(e){    self.onError(e);  })  .on('close', function(){    self.onClose('transport close');  });};

触发socket.js里面的onPacket()

Socket.prototype.onPacket = function (packet) {  if ('opening' == this.readyState || 'open' == this.readyState)