fix: avoid access internal socket state (#920)

Refs: nodejs/node#39495

lib/client.js

@@ -523,8 +523,18 @@ class Parser {
     }
 
     this.paused = false
-    this.socket.resume()
-    this.execute(EMPTY_BUF) // Flush parser.
+    this.execute(this.socket.read() || EMPTY_BUF) // Flush parser.
+    this.readMore()
+  }
+
+  readMore () {
+    while (!this.paused && this.ptr) {
+      const chunk = this.socket.read()
+      if (chunk === null) {
+        break
+      }
+      this.execute(chunk)
+    }
   }
 
   execute (data) {
@@ -569,7 +579,6 @@ class Parser {
         this.onUpgrade(data.slice(offset))
       } else if (ret === constants.ERROR.PAUSED) {
         this.paused = true
-        socket.pause()
         socket.unshift(data.slice(offset))
       } else if (ret !== constants.ERROR.OK) {
         const ptr = llhttp.exports.llhttp_get_error_reason(this.ptr)
@@ -709,7 +718,6 @@ class Parser {
 
     assert(!socket.destroyed)
     assert(socket === client[kSocket])
-    assert(!socket.isPaused())
     assert(!this.paused)
     assert(request.upgrade || request.method === 'CONNECT')
 
@@ -720,16 +728,6 @@ class Parser {
     this.headers = []
     this.headersSize = 0
 
-    // _readableState.flowing might be `true` if the socket has been
-    // explicitly `resume()`:d even if we never registered a 'data'
-    // listener.
-
-    // We need to stop unshift from emitting 'data'. However, we cannot
-    // call pause()  as that will stop socket from automatically resuming
-    // when 'data' listener is registered.
-
-    // Reset socket state to non flowing:
-    socket._readableState.flowing = null
     socket.unshift(head)
 
     socket[kParser].destroy()
@@ -739,7 +737,7 @@ class Parser {
     socket[kError] = null
     socket
       .removeListener('error', onSocketError)
-      .removeListener('data', onSocketData)
+      .removeListener('readable', onSocketReadable)
       .removeListener('end', onSocketEnd)
       .removeListener('close', onSocketClose)
 
@@ -995,9 +993,9 @@ function onParserTimeout (parser) {
   }
 }
 
-function onSocketData (data) {
+function onSocketReadable (data) {
   const { [kParser]: parser } = this
-  parser.execute(data)
+  parser.readMore()
 }
 
 function onSocketError (err) {
@@ -1138,7 +1136,7 @@ function connect (client) {
       socket[kClient] = client
       socket
         .on('error', onSocketError)
-        .on('data', onSocketData)
+        .on('readable', onSocketReadable)
         .on('end', onSocketEnd)
         .on('close', onSocketClose)
 

test/client-upgrade.js

@@ -173,7 +173,7 @@ test('upgrade invalid opts', (t) => {
 })
 
 test('basic upgrade2', (t) => {
-  t.plan(4)
+  t.plan(3)
 
   const server = http.createServer()
   server.on('upgrade', (req, c, head) => {
@@ -200,8 +200,6 @@ test('basic upgrade2', (t) => {
 
       const { headers, socket } = data
 
-      t.equal(socket._readableState.flowing, null)
-
       let recvData = ''
       data.socket.on('data', (d) => {
         recvData += d