fix: don't resume while piping request

lib/client.js

@@ -1,7 +1,5 @@
 'use strict'
 
-/* eslint no-prototype-builtins: "off" */
-
 const { URL } = require('url')
 const net = require('net')
 const tls = require('tls')
@@ -12,6 +10,7 @@ const assert = require('assert')
 const stream = require('stream')
 
 const kUrl = Symbol('url')
+const kWriting = Symbol('writing')
 const kQueue = Symbol('queue')
 const kTimeout = Symbol('timeout')
 const kTLSOpts = Symbol('TLS Options')
@@ -36,7 +35,6 @@ class Parser extends HTTPParser {
     this.socket = socket
     this.body = null
     this.read = 0
-    this.callback = null
   }
 
   [HTTPParser.kOnHeaders] () {
@@ -150,12 +148,9 @@ class Parser extends HTTPParser {
         process.nextTick(callback, err, null)
       }
     }
-    client[kQueue].splice(0, client[kInflight])
-    client[kInflight] = client[kComplete] = 0
-
-    while (retry.length) {
-      client[kQueue].unshift(retry.pop())
-    }
+    client[kQueue].splice(0, client[kInflight], ...retry)
+    client[kInflight] = 0
+    client[kComplete] = 0
 
     resume(client)
   }
@@ -219,7 +214,7 @@ function _connect (client) {
     client._socket = null
     client._parser = null
 
-    if (client[kQueue].length > 0) {
+    if (client.pending > 0) {
       connect(client)
     }
 
@@ -243,6 +238,8 @@ function connect (client) {
 const endRequest = (client) => {
   client._socket.write('\r\n', 'ascii')
   client._socket.uncork()
+
+  client[kWriting] = false
   resume(client)
 }
 
@@ -279,6 +276,10 @@ function resume (client) {
     return
   }
 
+  if (client[kWriting]) {
+    return
+  }
+
   const {
     host,
     method,
@@ -309,33 +310,34 @@ function resume (client) {
   }
 
   client[kInflight]++
+  client[kWriting] = true
+
+  const socket = client._socket
 
-  client._socket.cork()
-  client._socket.write(`${method} ${path} HTTP/1.1\r\nConnection: keep-alive\r\n`, 'ascii')
+  socket.cork()
+  socket.write(`${method} ${path} HTTP/1.1\r\nConnection: keep-alive\r\n`, 'ascii')
   if (!host) {
-    client._socket.write('Host: ' + client[kUrl].hostname + '\r\n', 'ascii')
+    socket.write('Host: ' + client[kUrl].hostname + '\r\n', 'ascii')
   }
-  client._socket.write(rawHeaders, 'ascii')
+  socket.write(rawHeaders, 'ascii')
 
   if (typeof body === 'string' || body instanceof Uint8Array) {
     if (chunked) {
-      client._socket.write(`content-length: ${Buffer.byteLength(body)}\r\n\r\n`, 'ascii')
+      socket.write(`content-length: ${Buffer.byteLength(body)}\r\n\r\n`, 'ascii')
     } else {
-      client._socket.write('\r\n')
+      socket.write('\r\n')
     }
-    client._socket.write(body)
+    socket.write(body)
     endRequest(client)
   } else if (body && typeof body.pipe === 'function') {
     if (chunked) {
-      client._socket.write('transfer-encoding: chunked\r\n', 'ascii')
+      socket.write('transfer-encoding: chunked\r\n', 'ascii')
     } else {
-      client._socket.write('\r\n', 'ascii')
+      socket.write('\r\n', 'ascii')
     }
 
     let finished = false
 
-    const socket = client._socket
-
     const onData = (chunk) => {
       if (chunked) {
         socket.write('\r\n' + Buffer.byteLength(chunk).toString(16) + '\r\n')
@@ -370,6 +372,9 @@ function resume (client) {
         if (!socket.destroyed) {
           socket.destroy(err)
         }
+
+        client[kWriting] = false
+        resume(client)
       } else {
         if (chunked) {
           socket.cork()
@@ -396,6 +401,7 @@ function resume (client) {
     endRequest(client)
   }
 }
+
 class Client extends EventEmitter {
   constructor (url, opts = {}) {
     super()
@@ -429,6 +435,7 @@ class Client extends EventEmitter {
     this[kRetryDelay] = 0
     this[kRetryTimeout] = null
     this[kOnDestroyed] = []
+    this[kWriting] = false
     this[kQueue] = []
     this[kInflight] = 0
     this[kComplete] = 0

test/client-pipelining.js

@@ -372,3 +372,54 @@ test('pipelining non-idempotent', (t) => {
     })
   })
 })
+
+test('pipelining non-idempotent w body', (t) => {
+  t.plan(4)
+
+  const server = createServer()
+  server.on('request', (req, res) => {
+    setImmediate(() => {
+      res.end('asd')
+    })
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`, {
+      pipelining: 2
+    })
+    t.tearDown(client.close.bind(client))
+
+    let ended = false
+    client.request({
+      path: '/',
+      method: 'POST',
+      body: new Readable({
+        read () {
+          this.push('asd')
+          setImmediate(() => {
+            this.push(null)
+            ended = true
+          })
+        }
+      })
+    }, (err, data) => {
+      t.error(err)
+      data.body
+        .resume()
+        .on('end', () => {
+          t.pass()
+        })
+    })
+
+    client.request({
+      path: '/',
+      method: 'GET',
+      idempotent: false
+    }, (err, data) => {
+      t.error(err)
+      t.strictEqual(ended, true)
+      data.body.resume()
+    })
+  })
+})