_onReadable() {
    // Read all the data until one of two conditions is met
    // 1. there is nothing left to read on the socket
    // 2. reading is paused because the consumer is slow
    while (!this._readingPaused) {
      // First step is reading the 32-bit integer from the socket
      // and if there is not a value, we simply abort processing
      let lenBuf = this._socket.read(4);
      if (!lenBuf) return;

      // Now that we have a length buffer we can convert it
      // into a number by reading the UInt32BE value
      // from the buffer.
      let len = lenBuf.readUInt32BE();

      // ensure that we don't exceed the max size of 256KiB
      if (len > 2 ** 18) {
        this.socket.destroy(new Error('Max length exceeded'));
        return;
      }

      // With the length, we can then consume the rest of the body.
      let body = this._socket.read(len);

      // If we did not have enough data on the wire to read the body
      // we will wait for the body to arrive and push the length
      // back into the socket's read buffer with unshift.
      if (!body) {
        this._socket.unshift(lenBuf);
        return;
      }

      // Try to parse the data and if it fails destroy the socket.
      let json;
      try {
        json = JSON.parse(body);
      } catch (ex) {
        this.socket.destroy(ex);
        return;
      }

      // Push the data into the read buffer and capture whether
      // we are hitting the back pressure limits
      let pushOk = this.push(json);

      // When the push fails, we need to pause the ability to read
      // messages because the consumer is getting backed up.
      if (!pushOk) this._readingPaused = true;
    }
  }
/**
    Performs data read events which are triggered under two conditions:
    1. underlying `readable` events emitted when there is new data
       available on the socket
    2. the consumer requested additional data
    @private
   */
Performs data read events which are triggered under two conditions:
1. underlying `readable` events emitted when there is new data
available on the socket
2. the consumer requested additional data
Performs data read events which are triggered under two conditions : 1 . underlying ` readable ` events emitted when there is new data available on the socket 2 . the consumer requested additional data
_onReadable() {
    while (!this._readingPaused) {
      let lenBuf = this._socket.read(4);
      if (!lenBuf) return;
      let len = lenBuf.readUInt32BE();
      if (len > 2 ** 18) {
        this.socket.destroy(new Error('Max length exceeded'));
        return;
      }
      let body = this._socket.read(len);
      if (!body) {
        this._socket.unshift(lenBuf);
        return;
      }
      let json;
      try {
        json = JSON.parse(body);
      } catch (ex) {
        this.socket.destroy(ex);
        return;
      }
      let pushOk = this.push(json);
      if (!pushOk) this._readingPaused = true;
    }
  }
_onReadable ( ) { while ( ! this . _readingPaused ) { let lenBuf = this . _socket . read ( 4 ) ; if ( ! lenBuf ) return ; let len = lenBuf . readUInt32BE ( ) ; if ( len > 2 ** 18 ) { this . socket . destroy ( new Error ( 'Max length exceeded' ) ) ; return ; } let body = this . _socket . read ( len ) ; if ( ! body ) { this . _socket . unshift ( lenBuf ) ; return ; } let json ; try { json = JSON . parse ( body ) ; } catch ( ex ) { this . socket . destroy ( ex ) ; return ; } let pushOk = this . push ( json ) ; if ( ! pushOk ) this . _readingPaused = true ; } }
Performs data read events which are triggered under two conditions:
1. underlying `readable` events emitted when there is new data
available on the socket
2. the consumer requested additional data
Performs data read events which are triggered under two conditions : 1 . underlying ` readable ` events emitted when there is new data available on the socket 2 . the consumer requested additional data
// Read all the data until one of two conditions is met // 1. there is nothing left to read on the socket // 2. reading is paused because the consumer is slow // First step is reading the 32-bit integer from the socket // and if there is not a value, we simply abort processing // Now that we have a length buffer we can convert it // into a number by reading the UInt32BE value // from the buffer. // ensure that we don't exceed the max size of 256KiB // With the length, we can then consume the rest of the body. // If we did not have enough data on the wire to read the body // we will wait for the body to arrive and push the length // back into the socket's read buffer with unshift. // Try to parse the data and if it fails destroy the socket. // Push the data into the read buffer and capture whether // we are hitting the back pressure limits // When the push fails, we need to pause the ability to read // messages because the consumer is getting backed up.