lib/protocols.js
function emit(retry) {
    if (self._interrupted) {
      // The request's callback will already have been called.
      return;
    }

    var hreq = self._createHandshakeRequest(adapter, !retry);

    var writable = self._writableFactory.call(self, function (err, readable) {
      if (err) {
        cb(err);
        return;
      }
      readable.on('data', function (obj) {
        debug('received response %s', obj.id);
        // We don't check that the prefix matches since the ID likely hasn't
        // been propagated to the response (see default stateless codec).
        var buf = Buffer.concat(obj.payload);
        try {
          var parts = readHead(HANDSHAKE_RESPONSE_TYPE, buf);
          var hres = parts.head;
          if (hres.serverHash) {
            adapter = self._getAdapter(hres);
          }
        } catch (err) {
          cb(err);
          return;
        }
        self.emit('handshake', hreq, hres);
        if (hres.match === 'NONE') {
          process.nextTick(function() { emit(true); });
          return;
        }
        // Change the default adapter.
        self._adapter = adapter;
        cb(null, parts.tail, adapter);
      });
    });

    writable.write({
      id: id,
      payload: [HANDSHAKE_REQUEST_TYPE.toBuffer(hreq), reqBuf]
    });
    if (self._endWritable) {
      writable.end();
    }
  }
function emit(retry) {
    if (self._interrupted) {
      // The request's callback will already have been called.
      return;
    }

    var hreq = self._createHandshakeRequest(adapter, !retry);

    var writable = self._writableFactory.call(self, function (err, readable) {
      if (err) {
        cb(err);
        return;
      }
      readable.on('data', function (obj) {
        debug('received response %s', obj.id);
        // We don't check that the prefix matches since the ID likely hasn't
        // been propagated to the response (see default stateless codec).
        var buf = Buffer.concat(obj.payload);
        try {
          var parts = readHead(HANDSHAKE_RESPONSE_TYPE, buf);
          var hres = parts.head;
          if (hres.serverHash) {
            adapter = self._getAdapter(hres);
          }
        } catch (err) {
          cb(err);
          return;
        }
        self.emit('handshake', hreq, hres);
        if (hres.match === 'NONE') {
          process.nextTick(function() { emit(true); });
          return;
        }
        // Change the default adapter.
        self._adapter = adapter;
        cb(null, parts.tail, adapter);
      });
    });

    writable.write({
      id: id,
      payload: [HANDSHAKE_REQUEST_TYPE.toBuffer(hreq), reqBuf]
    });
    if (self._endWritable) {
      writable.end();
    }
  }
function emit ( retry ) { if ( self . _interrupted ) { // The request's callback will already have been called. return ; } var hreq = self . _createHandshakeRequest ( adapter , ! retry ) ; var writable = self . _writableFactory . call ( self , function ( err , readable ) { if ( err ) { cb ( err ) ; return ; } readable . on ( 'data' , function ( obj ) { debug ( 'received response %s' , obj . id ) ; // We don't check that the prefix matches since the ID likely hasn't // been propagated to the response (see default stateless codec). var buf = Buffer . concat ( obj . payload ) ; try { var parts = readHead ( HANDSHAKE_RESPONSE_TYPE , buf ) ; var hres = parts . head ; if ( hres . serverHash ) { adapter = self . _getAdapter ( hres ) ; } } catch ( err ) { cb ( err ) ; return ; } self . emit ( 'handshake' , hreq , hres ) ; if ( hres . match === 'NONE' ) { process . nextTick ( function ( ) { emit ( true ) ; } ) ; return ; } // Change the default adapter. self . _adapter = adapter ; cb ( null , parts . tail , adapter ) ; } ) ; } ) ; writable . write ( { id : id , payload : [ HANDSHAKE_REQUEST_TYPE . toBuffer ( hreq ) , reqBuf ] } ) ; if ( self . _endWritable ) { writable . end ( ) ; } }
Each writable is only used once, no risk of buffering.
Each writable is only used once no risk of buffering .