lib/protocols.js
function StatelessListener(ptcl, readableFactory, opts) {
  MessageListener.call(this, ptcl, opts);
  var self = this;
  var readable;

  process.nextTick(function () {
    // Delay listening to allow handlers to be attached even if the factory is
    // purely synchronous.
    readable = readableFactory.call(this, function (err, writable) {
      if (err) {
        self.emit('error', err);
        // Since stateless listeners are only used once, it is safe to destroy.
        onFinish();
        return;
      }
      self._writable = writable.on('finish', onFinish);
      self.emit('_writable');
    }).on('data', onRequest)
      .on('end', onEnd);
  });

  function onRequest(obj) {
    var id = obj.id;
    var buf = Buffer.concat(obj.payload);
    var err = null;
    try {
      var parts = readHead(HANDSHAKE_REQUEST_TYPE, buf);
      var hreq = parts.head;
      var adapter = self._getAdapter(hreq);
    } catch (cause) {
      err = wrapError('invalid handshake request', cause);
    }

    if (err) {
      done(encodeError(err));
    } else {
      self._receive(parts.tail, adapter, done);
    }

    function done(resBuf) {
      if (!self._writable) {
        self.once('_writable', function () { done(resBuf); });
        return;
      }
      var hres = self._createHandshakeResponse(err, hreq);
      self.emit('handshake', hreq, hres);
      var payload = [
        HANDSHAKE_RESPONSE_TYPE.toBuffer(hres),
        resBuf
      ];
      self._writable.write({id: id, payload: payload});
      if (self._endWritable) {
        self._writable.end();
      }
    }
  }

  function onEnd() { self.destroy(); }

  function onFinish() {
    if (readable) {
      readable
        .removeListener('data', onRequest)
        .removeListener('end', onEnd);
    }
    self.destroy(true);
  }
}
function StatelessListener(ptcl, readableFactory, opts) {
  MessageListener.call(this, ptcl, opts);
  var self = this;
  var readable;

  process.nextTick(function () {
    // Delay listening to allow handlers to be attached even if the factory is
    // purely synchronous.
    readable = readableFactory.call(this, function (err, writable) {
      if (err) {
        self.emit('error', err);
        // Since stateless listeners are only used once, it is safe to destroy.
        onFinish();
        return;
      }
      self._writable = writable.on('finish', onFinish);
      self.emit('_writable');
    }).on('data', onRequest)
      .on('end', onEnd);
  });

  function onRequest(obj) {
    var id = obj.id;
    var buf = Buffer.concat(obj.payload);
    var err = null;
    try {
      var parts = readHead(HANDSHAKE_REQUEST_TYPE, buf);
      var hreq = parts.head;
      var adapter = self._getAdapter(hreq);
    } catch (cause) {
      err = wrapError('invalid handshake request', cause);
    }

    if (err) {
      done(encodeError(err));
    } else {
      self._receive(parts.tail, adapter, done);
    }

    function done(resBuf) {
      if (!self._writable) {
        self.once('_writable', function () { done(resBuf); });
        return;
      }
      var hres = self._createHandshakeResponse(err, hreq);
      self.emit('handshake', hreq, hres);
      var payload = [
        HANDSHAKE_RESPONSE_TYPE.toBuffer(hres),
        resBuf
      ];
      self._writable.write({id: id, payload: payload});
      if (self._endWritable) {
        self._writable.end();
      }
    }
  }

  function onEnd() { self.destroy(); }

  function onFinish() {
    if (readable) {
      readable
        .removeListener('data', onRequest)
        .removeListener('end', onEnd);
    }
    self.destroy(true);
  }
}
function StatelessListener ( ptcl , readableFactory , opts ) { MessageListener . call ( this , ptcl , opts ) ; var self = this ; var readable ; process . nextTick ( function ( ) { // Delay listening to allow handlers to be attached even if the factory is // purely synchronous. readable = readableFactory . call ( this , function ( err , writable ) { if ( err ) { self . emit ( 'error' , err ) ; // Since stateless listeners are only used once, it is safe to destroy. onFinish ( ) ; return ; } self . _writable = writable . on ( 'finish' , onFinish ) ; self . emit ( '_writable' ) ; } ) . on ( 'data' , onRequest ) . on ( 'end' , onEnd ) ; } ) ; function onRequest ( obj ) { var id = obj . id ; var buf = Buffer . concat ( obj . payload ) ; var err = null ; try { var parts = readHead ( HANDSHAKE_REQUEST_TYPE , buf ) ; var hreq = parts . head ; var adapter = self . _getAdapter ( hreq ) ; } catch ( cause ) { err = wrapError ( 'invalid handshake request' , cause ) ; } if ( err ) { done ( encodeError ( err ) ) ; } else { self . _receive ( parts . tail , adapter , done ) ; } function done ( resBuf ) { if ( ! self . _writable ) { self . once ( '_writable' , function ( ) { done ( resBuf ) ; } ) ; return ; } var hres = self . _createHandshakeResponse ( err , hreq ) ; self . emit ( 'handshake' , hreq , hres ) ; var payload = [ HANDSHAKE_RESPONSE_TYPE . toBuffer ( hres ) , resBuf ] ; self . _writable . write ( { id : id , payload : payload } ) ; if ( self . _endWritable ) { self . _writable . end ( ) ; } } } function onEnd ( ) { self . destroy ( ) ; } function onFinish ( ) { if ( readable ) { readable . removeListener ( 'data' , onRequest ) . removeListener ( 'end' , onEnd ) ; } self . destroy ( true ) ; } }
MessageListener for stateless transport.

This listener expect a handshake to precede each message.
MessageListener for stateless transport .