fix: closed client should be able to reconnect

lib/client.js

@@ -50,31 +50,27 @@ function connect (client) {
   })
 
   client[kStream].finished(socket, (err) => {
-    reconnect(client, err || new Error('other side closed'))
+    reconnect(client, err)
   })
 }
 
-function destroyMaybe (client) {
-  if (
-    client.closed &&
-    !client[kLastBody] &&
-    client[kQueue].length() === 0 &&
-    client[kInflight].length === 0
-  ) {
-    client.destroy()
-  }
-}
-
 function reconnect (client, err) {
-  err = err || new Error('premature close')
+  err = err || new Error('other side closed')
 
   if (client[kLastBody]) {
     client[kLastBody].destroy(err)
     client[kLastBody] = null
   }
 
-  if (client.closed) {
-    // TODO what do we do with the error?
+  if (client.destroyed) {
+    // reset callbacks
+    const inflight = client[kInflight].splice(0)
+    for (const { callback } of inflight) {
+      callback(err, null)
+    }
+    // flush queue
+    // TODO: Forward err?
+    client[kQueue].resume()
     return
   }
 
@@ -269,7 +265,9 @@ class Client extends EventEmitter {
           headers: parseHeaders(headers),
           body: this[kLastBody]
         })
-        destroyMaybe(this)
+        if (this.closed) {
+          destroyMaybe(this)
+        }
         return skipBody
       }
 
@@ -283,7 +281,9 @@ class Client extends EventEmitter {
         if (body !== null) {
           body.push(null)
         }
-        destroyMaybe(this)
+        if (this.closed) {
+          destroyMaybe(this)
+        }
       }
     }
     this[kResetParser]()
@@ -425,6 +425,16 @@ function parseHeaders (headers) {
 
 module.exports = Client
 
+function destroyMaybe (client) {
+  if (
+    !client[kLastBody] &&
+    client[kQueue].length() === 0 &&
+    client[kInflight].length === 0
+  ) {
+    client.destroy()
+  }
+}
+
 function destroySocket (client, err, cb) {
   // This code is basically the same as...
   // stream.finished(socket, er => cb(err || er))

test/client-errors.js

@@ -265,7 +265,7 @@ test('invalid URL throws', (t) => {
 })
 
 test('POST which fails should error response', (t) => {
-  t.plan(2)
+  t.plan(4)
 
   const server = createServer()
   server.once('request', (req, res) => {

test/client.js

@@ -586,3 +586,33 @@ test('close should call callback once finished', (t) => {
     }
   })
 })
+
+test('close should still reconnect', (t) => {
+  t.plan(6)
+
+  const server = createServer((req, res) => {
+    res.end(req.url)
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`, {
+      pipelining: 1
+    })
+
+    t.ok(makeRequest())
+    t.ok(!makeRequest())
+
+    client.close((err) => {
+      t.strictEqual(err, null)
+      t.strictEqual(client.closed, true)
+    })
+    client.socket.destroy()
+
+    function makeRequest () {
+      return client.request({ path: '/', method: 'GET' }, (err, data) => {
+        t.error(err)
+      })
+    }
+  })
+})

test/close-and-destroy.js

@@ -47,3 +47,34 @@ test('close waits for queued requests to finish', (t) => {
     })
   }
 })
+
+test('destroy invoked all pending callbacks', (t) => {
+  t.plan(4)
+
+  const server = createServer()
+
+  server.on('request', (req, res) => {
+    res.write('hello')
+  })
+  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, data) => {
+      t.error(err)
+      data.body.on('error', (err) => {
+        t.ok(err)
+      })
+      client.destroy()
+    })
+    client.request({ path: '/', method: 'GET' }, (err) => {
+      t.ok(err)
+    })
+    client.request({ path: '/', method: 'GET' }, (err) => {
+      t.ok(err)
+    })
+  })
+})