fix: pause while piping

Fixes: #52
nodejs · May 6, 2020 · d948a3f · d948a3f
1 parent e6715eb
commit d948a3f
Show file tree

Hide file tree

Showing 2 changed files with 69 additions and 3 deletions.
diff --git a/lib/client.js b/lib/client.js
@@ -254,14 +254,14 @@ class Client extends EventEmitter {
         this.socket.write(body)
         endRequest()
       } else if (body && typeof body.pipe === 'function') {
+        this[kQueue].pause()
+
         if (chunked) {
           this.socket.write('transfer-encoding: chunked\r\n', 'ascii')
         } else {
           this.socket.write('\r\n', 'ascii')
         }
 
-        // TODO: Pause the queue while piping.
-
         let finished = false
 
         const socket = this.socket
@@ -309,6 +309,10 @@ class Client extends EventEmitter {
             if (chunked) {
               socket.cork()
               socket.write('\r\n0\r\n', 'ascii')
+              // Don't resume queue. Server should destroy connection once
+              // completed. Queue will automatically resume once reconnected.
+            } else {
+              this[kQueue].resume()
             }
 
             endRequest()

diff --git a/test/client.js b/test/client.js
@@ -4,7 +4,7 @@ const { test } = require('tap')
 const { Client } = require('..')
 const { createServer } = require('http')
 const { readFileSync, createReadStream } = require('fs')
-const { finished } = require('readable-stream')
+const { finished, Readable } = require('readable-stream')
 
 test('basic get', (t) => {
   t.plan(7)
@@ -655,3 +655,65 @@ test('close waits until socket is destroyed', (t) => {
     }
   })
 })
+
+test('pipelined chunked POST ', (t) => {
+  t.plan(4)
+
+  const server = createServer((req, res) => {
+    req.on('data', chunk => {
+    }).on('end', () => {
+      res.end()
+    })
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`, {
+      pipelining: 2
+    })
+
+    client.request({
+      path: '/',
+      method: 'GET'
+    }, (err, { body }) => {
+      body.resume()
+      t.error(err)
+    })
+
+    let a = 0
+    client.request({
+      path: '/',
+      method: 'POST',
+      body: new Readable({
+        read () {
+          this.push(a++ > 128 ? null : 'a')
+        }
+      })
+    }, (err, { body }) => {
+      body.resume()
+      t.error(err)
+    })
+
+    client.request({
+      path: '/',
+      method: 'GET'
+    }, (err, { body }) => {
+      body.resume()
+      t.error(err)
+    })
+
+    let b = 0
+    client.request({
+      path: '/',
+      method: 'POST',
+      body: new Readable({
+        read () {
+          this.push(b++ > 128 ? null : 'b')
+        }
+      })
+    }, (err, { body }) => {
+      body.resume()
+      t.error(err)
+    })
+  })
+})