fix: request body edge cases

Readable can emit 'end' after .destroy() as well
as legacy streams might not implement the whole
Readable contract properly. Try to handle these
edge cases better.

Fixes: nodejs#66

lib/client.js

@@ -18,6 +18,8 @@ const kLastBody = Symbol('lastBody')
 const kStream = Symbol('kStream')
 const kClosed = Symbol('kClosed')
 
+function nop () {}
+
 function connect (client) {
   var socket = null
   var url = client.url
@@ -74,7 +76,7 @@ function reconnect (client, err) {
   client.socket.removeAllListeners('end')
   client.socket.removeAllListeners('finish')
   client.socket.removeAllListeners('error')
-  client.socket.on('error', () => {})
+  client.socket.on('error', nop)
   client.socket = null
 
   // we reset the callbacks
@@ -160,19 +162,14 @@ class Client extends EventEmitter {
         this.socket.write(body)
         endRequest()
       } else if (body && typeof body.pipe === 'function') {
-        const cleanup = this[kStream].finished(this.socket, (err) => {
-          if (err) {
-            body.destroy(err)
-          }
-        })
-
         if (chunked) {
           this.socket.write('transfer-encoding: chunked\r\n', 'ascii')
         } else {
           this.socket.write('\r\n', 'ascii')
         }
 
-        // TODO we should pause the queue while we are piping
+        // TODO: Pause the queue while piping.
+
         const onData = (chunk) => {
           if (chunked) {
             this.socket.write('\r\n' + Buffer.byteLength(chunk).toString(16) + '\r\n')
@@ -181,39 +178,56 @@ class Client extends EventEmitter {
             body.pause()
           }
         }
+        const onEnd = () => {
+          this.socket.removeListener('drain', onDrain)
+
+          if (chunked) {
+            this.socket.cork()
+            this.socket.write('\r\n0\r\n', 'ascii')
+          }
+
+          endRequest()
+        }
         const onDrain = () => {
           body.resume()
         }
+        const onFinished = (err) => {
+          freeSocketFinished()
+          freeBodyFinished()
 
-        body.on('data', onData)
-        this.socket.on('drain', onDrain)
-
-        this.socket.uncork()
-        this[kStream].finished(body, (err) => {
-          cleanup()
-          if (err) {
-            // TODO we might want to wait before previous in-flight
-            // requests are finished before destroying
-            if (this.socket) {
-              finishedSocket(this, cb)
-              this.socket.destroy(err)
-            } else {
-              callback(err, null)
-            }
+          if (!err) {
             return
           }
 
-          assert(this.socket)
+          body
+            .removeListener('data', onData)
+            .removeListener('end', onEnd)
 
-          this.socket.removeListener('drain', onDrain)
+          if (typeof body.destroy === 'function') {
+            body.destroy(err)
+          }
 
-          if (chunked) {
-            this.socket.cork()
-            this.socket.write('\r\n0\r\n', 'ascii')
+          // TODO we might want to wait before previous in-flight
+          // requests are finished before destroying
+          if (this.socket) {
+            finishedSocket(this, cb)
+            this.socket.destroy(err)
+          } else {
+            callback(err, null)
           }
+        }
 
-          endRequest()
-        })
+        body
+          .on('data', onData)
+          .on('end', onEnd)
+          .on('error', nop)
+
+        this.socket
+          .on('drain', onDrain)
+          .uncork()
+
+        const freeSocketFinished = this[kStream].finished(this.socket, onFinished)
+        const freeBodyFinished = this[kStream].finished(body, onFinished)
       } else {
         assert(!body)
         endRequest()

test/client-errors.js

@@ -440,3 +440,65 @@ test('fail invalid body regexp', (t) => {
     })
   })
 })
+
+test('socket fail while writing request body', (t) => {
+  t.plan(1)
+
+  const server = createServer()
+  server.once('request', (req, res) => {
+    res.write('asd')
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    t.tearDown(client.close.bind(client))
+    const body = new Readable({ read () {} })
+    body.push('asd')
+
+    client.on('connect', () => {
+      client.socket.destroy('kaboom')
+    })
+
+    client.request({
+      path: '/',
+      method: 'POST',
+      body
+    }, (err) => {
+      t.ok(err)
+    })
+  })
+})
+
+test('socket fail while ending request body', (t) => {
+  t.plan(2)
+
+  const server = createServer()
+  server.once('request', (req, res) => {
+    res.end()
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    t.tearDown(client.close.bind(client))
+
+    client.request({
+      path: '/',
+      method: 'POST',
+      body: 'asd'
+    }, (err) => {
+      t.error(err)
+      const body = new Readable({ read () {} })
+      body.push(null)
+      client.request({
+        path: '/',
+        method: 'POST',
+        body
+      }, (err) => {
+        t.ok(err)
+      })
+      client.socket.destroy('kaboom')
+    })
+  })
+})