lib/services.js
function StatelessServerChannel(server, readableFactory, opts) {
  ServerChannel.call(this, server, opts);

  this._writable = undefined;
  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(self, function (err, writable) {
      process.nextTick(function () {
        // We delay once more here in case this call is synchronous, to allow
        // the readable to always be populated first.
        if (err) {
          onFinish(err);
          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;
    try {
      var parts = readHead(HANDSHAKE_REQUEST_TYPE, buf);
      var hreq = parts.head;
      var adapter = self._getAdapter(hreq);
    } catch (cause) {
      err = toRpcError('INVALID_HANDSHAKE_REQUEST', cause);
    }

    var hres = self._createHandshakeResponse(err, hreq);
    self.emit('handshake', hreq, hres);
    if (err) {
      done(self._encodeSystemError(err));
    } else {
      self._receive(parts.tail, adapter, done);
    }

    function done(resBuf) {
      if (!self.destroyed) {
        if (!self._writable) {
          self.once('_writable', function () { done(resBuf); });
          return;
        }
        self._writable.write({
          id: id,
          payload: [HANDSHAKE_RESPONSE_TYPE.toBuffer(hres), resBuf]
        });
      }
      if (self._writable && self._endWritable) {
        self._writable.end();
      }
    }
  }

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

  function onFinish(err) {
    readable
      .removeListener('data', onRequest)
      .removeListener('end', onEnd);
    self.destroy(err || true);
  }
}
function StatelessServerChannel(server, readableFactory, opts) {
  ServerChannel.call(this, server, opts);

  this._writable = undefined;
  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(self, function (err, writable) {
      process.nextTick(function () {
        // We delay once more here in case this call is synchronous, to allow
        // the readable to always be populated first.
        if (err) {
          onFinish(err);
          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;
    try {
      var parts = readHead(HANDSHAKE_REQUEST_TYPE, buf);
      var hreq = parts.head;
      var adapter = self._getAdapter(hreq);
    } catch (cause) {
      err = toRpcError('INVALID_HANDSHAKE_REQUEST', cause);
    }

    var hres = self._createHandshakeResponse(err, hreq);
    self.emit('handshake', hreq, hres);
    if (err) {
      done(self._encodeSystemError(err));
    } else {
      self._receive(parts.tail, adapter, done);
    }

    function done(resBuf) {
      if (!self.destroyed) {
        if (!self._writable) {
          self.once('_writable', function () { done(resBuf); });
          return;
        }
        self._writable.write({
          id: id,
          payload: [HANDSHAKE_RESPONSE_TYPE.toBuffer(hres), resBuf]
        });
      }
      if (self._writable && self._endWritable) {
        self._writable.end();
      }
    }
  }

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

  function onFinish(err) {
    readable
      .removeListener('data', onRequest)
      .removeListener('end', onEnd);
    self.destroy(err || true);
  }
}
function StatelessServerChannel ( server , readableFactory , opts ) { ServerChannel . call ( this , server , opts ) ; this . _writable = undefined ; 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 ( self , function ( err , writable ) { process . nextTick ( function ( ) { // We delay once more here in case this call is synchronous, to allow // the readable to always be populated first. if ( err ) { onFinish ( err ) ; 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 ; try { var parts = readHead ( HANDSHAKE_REQUEST_TYPE , buf ) ; var hreq = parts . head ; var adapter = self . _getAdapter ( hreq ) ; } catch ( cause ) { err = toRpcError ( 'INVALID_HANDSHAKE_REQUEST' , cause ) ; } var hres = self . _createHandshakeResponse ( err , hreq ) ; self . emit ( 'handshake' , hreq , hres ) ; if ( err ) { done ( self . _encodeSystemError ( err ) ) ; } else { self . _receive ( parts . tail , adapter , done ) ; } function done ( resBuf ) { if ( ! self . destroyed ) { if ( ! self . _writable ) { self . once ( '_writable' , function ( ) { done ( resBuf ) ; } ) ; return ; } self . _writable . write ( { id : id , payload : [ HANDSHAKE_RESPONSE_TYPE . toBuffer ( hres ) , resBuf ] } ) ; } if ( self . _writable && self . _endWritable ) { self . _writable . end ( ) ; } } } function onEnd ( ) { self . destroy ( ) ; } function onFinish ( err ) { readable . removeListener ( 'data' , onRequest ) . removeListener ( 'end' , onEnd ) ; self . destroy ( err || true ) ; } }
Server channel for stateless transport.

This channel expect a handshake to precede each message.
Server channel for stateless transport .