docs: function StatefulListener(ptcl, readable, writable, opts) {
    MessageListener.call(this, ptcl, opts);
    this._adapter = undefined;
    this._writable = writable.on('finish', onFinish);
    this._readable = readable.on('data', onHandshake).on('end', onEnd);
  
    this.once('eot', function () {
      // Clean up any references to the listener on the underlying streams.
      this._writable.removeListener('finish', onFinish);
      if (this._endWritable) {
        this._writable.end();
      }
      this._readable
        .removeListener('data', onHandshake)
        .removeListener('data', onRequest)
        .removeListener('end', onEnd);
    });
  
    var self = this;
  
    function onHandshake(obj) {
      var id = obj.id;
      if (!self._matchesPrefix(id)) {
        return;
      }
      var buf = Buffer.concat(obj.payload);
      var err;
      try {
        var parts = readHead(HANDSHAKE_REQUEST_TYPE, buf);
        var hreq = parts.head;
        self._adapter = self._getAdapter(hreq);
      } catch (cause) {
        err = wrapError('invalid handshake request', cause);
      }
      if (err) {
        // Either the client's protocol was unknown or it isn't compatible.
        done(encodeError(err));
      } else {
        self._readable
          .removeListener('data', onHandshake)
          .on('data', onRequest);
        self._receive(parts.tail, self._adapter, done);
      }
  
      function done(resBuf) {
        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});
      }
    }
  
    function onRequest(obj) {
      // These requests are not prefixed with handshakes.
      var id = obj.id;
      if (!self._matchesPrefix(id)) {
        return;
      }
      var reqBuf = Buffer.concat(obj.payload);
      self._receive(reqBuf, self._adapter, function (resBuf, isOneWay) {
        if (!isOneWay) {
          self._writable.write({id: id, payload: [resBuf]});
        }
      });
    }
  
    function onEnd() { self.destroy(); }
  
    function onFinish() { self.destroy(true); }
  }