fix: CONNECT calls onUpgrade multiple times (#348)

Author: ronag

Diff for: lib/core/client.js

@@ -454,7 +454,6 @@ class Parser extends HTTPParser {
         client[kQueue][client[kRunningIdx]++] = null
         client.emit('disconnect', new InformationalError('upgrade'))
 
-        this.unconsume()
         setImmediate(() => this.close())
 
         resume(client)
@@ -499,6 +498,7 @@ class Parser extends HTTPParser {
     this.shouldKeepAlive = shouldKeepAlive
 
     if (upgrade || request.method === 'CONNECT') {
+      this.unconsume()
       this.upgrade = true
       return 2
     }

Diff for: test/client-dispatch.js

@@ -269,3 +269,62 @@ test('dispatch onConnect error', (t) => {
     })
   })
 })
+
+test('connect call onUpgrade once', (t) => {
+  t.plan(2)
+
+  const server = http.createServer((c) => {
+    t.fail()
+  })
+  server.on('connect', (req, socket, firstBodyChunk) => {
+    socket.write('HTTP/1.1 200 Connection established\r\n\r\n')
+
+    let data = firstBodyChunk.toString()
+    socket.on('data', (buf) => {
+      data += buf.toString()
+    })
+
+    socket.on('end', () => {
+      socket.end(data)
+    })
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, async () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    t.tearDown(client.close.bind(client))
+
+    let recvData = ''
+    let count = 0
+    client.dispatch({
+      method: 'CONNECT',
+      path: '/'
+    }, {
+      onConnect () {
+      },
+      onUpgrade (statusCode, headers, socket) {
+        t.strictEqual(count++, 0)
+
+        socket.on('data', (d) => {
+          recvData += d
+        })
+
+        socket.on('end', () => {
+          t.strictEqual(recvData.toString(), 'Body')
+        })
+
+        socket.write('Body')
+        socket.end()
+      },
+      onData (buf) {
+        t.fail()
+      },
+      onComplete (trailers) {
+        t.fail()
+      },
+      onError () {
+        t.fail()
+      }
+    })
+  })
+})

Diff for: test/client-upgrade.js

@@ -346,6 +346,9 @@ test('basic upgrade error', (t) => {
       c.write('\r\n')
       c.write('Body')
     })
+    c.on('error', () => {
+
+    })
   })
   t.tearDown(server.close.bind(server))