docs: function StatefulEmitter(ptcl, readable, writable, opts) {
    MessageEmitter.call(this, ptcl, opts);
    this._readable = readable;
    this._writable = writable;
    this._connected = !!(opts && opts.noPing);
    this._readable.on('end', onEnd);
    this._writable.on('finish', onFinish);
  
    this.once('eot', function () {
      debug('emitter eot');
      // Remove references to this emitter to avoid potential memory leaks.
      this._writable.removeListener('finish', onFinish);
      if (this._endWritable) {
        this._writable.end();
      }
      this._readable
        .removeListener('data', onPing)
        .removeListener('data', onMessage)
        .removeListener('end', onEnd);
    });
  
    var self = this;
    var hreq; // For handshake events.
    if (this._connected) {
      this._readable.on('data', onMessage);
    } else {
      this._readable.on('data', onPing);
      process.nextTick(ping);
    }
  
    function ping(retry) {
      if (self._destroyed) {
        return;
      }
      hreq = self._createHandshakeRequest(self._adapter, !retry);
      var payload = [
        HANDSHAKE_REQUEST_TYPE.toBuffer(hreq),
        new Buffer([0, 0]) // No header, no data (empty message name).
      ];
      self._writable.write({id: self._prefix, payload: payload});
    }
  
    function onPing(obj) {
      if (!self._matchesPrefix(obj.id)) {
        debug('discarding unscoped response %s (still connecting)', obj.id);
        return;
      }
      var buf = Buffer.concat(obj.payload);
      try {
        var hres = readHead(HANDSHAKE_RESPONSE_TYPE, buf).head;
        if (hres.serverHash) {
          self._adapter = self._getAdapter(hres);
        }
      } catch (err) {
        self.destroy(true); // Not a recoverable error.
        self.emit('error', wrapError('handshake error (this can happen if the ' +
          'underlying transport is shared by multiple emitters or listeners ' +
          'using the same scope)', err));
        return;
      }
      self.emit('handshake', hreq, hres);
      if (hres.match === 'NONE') {
        ping(true);
      } else {
        debug('successfully connected');
        self._readable.removeListener('data', onPing).on('data', onMessage);
        self._connected = true;
        self.emit('_connected');
        hreq = null; // Release reference.
      }
    }
  
    // Callback used after a connection has been established.
    function onMessage(obj) {
      var id = obj.id;
      if (!self._matchesPrefix(id)) {
        debug('discarding unscoped request %s', id);
        return;
      }
      var cb = self._registry.get(id);
      if (cb) {
        process.nextTick(function () {
          debug('received request %s', id);
          // Ensure that the initial callback gets called asynchronously, even
          // for completely synchronous transports (otherwise the number of
          // pending requests will sometimes be inconsistent between stateful and
          // stateless transports).
          cb(null, Buffer.concat(obj.payload), self._adapter);
        });
      }
    }
  
    function onEnd() { self.destroy(true); }
    function onFinish() { self.destroy(); }
  }