feat: allow overriding stream implementation

Fixes: nodejs#17

lib/client.js

@@ -7,8 +7,6 @@ const net = require('net')
 const tls = require('tls')
 const Q = require('fastq')
 const { HTTPParser } = require('http-parser-js')
-const { Readable } = require('readable-stream')
-const eos = require('end-of-stream')
 const retimer = require('retimer')
 const { EventEmitter } = require('events')
 const Request = require('./request')
@@ -22,6 +20,7 @@ const kCallbacks = Symbol('callbacks')
 const kRequests = Symbol('requests')
 const kTimer = Symbol('kTimer')
 const kTLSOpts = Symbol('TLS Options')
+const kStream = Symbol('kStream')
 
 function connect (client) {
   var socket = null
@@ -45,7 +44,7 @@ function connect (client) {
     client[kQueue].resume()
   })
 
-  eos(socket, (err) => {
+  client[kStream].finished(socket, (err) => {
     reconnect(client, err || new Error('other side closed'))
   })
 }
@@ -113,6 +112,8 @@ class Client extends EventEmitter {
 
     this.url = url
 
+    this[kStream] = opts.stream || require('readable-stream')
+
     // state machine, might need more states
     this.closed = false
     this.destroyed = false
@@ -186,7 +187,7 @@ class Client extends EventEmitter {
         }
         this.socket.write(body)
       } else if (body && typeof body.pipe === 'function') {
-        const cleanup = eos(this.socket, err => {
+        const cleanup = this[kStream].finished(this.socket, err => {
           if (err) {
             body.destroy(err)
           }
@@ -215,7 +216,7 @@ class Client extends EventEmitter {
         this.socket.on('drain', onDrain)
 
         this.socket.uncork()
-        eos(body, (err) => {
+        this[kStream].finished(body, (err) => {
           cleanup()
           if (err || !this.socket) {
             // TODO we might want to wait before previous in-flight
@@ -263,7 +264,7 @@ class Client extends EventEmitter {
       const skipBody = request.method === 'HEAD'
 
       if (!skipBody) {
-        this._lastBody = new Readable({ read: this[kRead].bind(this) })
+        this._lastBody = new this[kStream].Readable({ read: this[kRead].bind(this) })
         this._lastBody.push = request.wrapSimple(this._lastBody, this._lastBody.push)
       }
       cb(null, {
@@ -368,8 +369,8 @@ class Client extends EventEmitter {
     }
 
     if (cb) {
-      eos(this.socket, (err) => {
-        if (err && err.message !== 'premature close') {
+      this[kStream].finished(this.socket, (err) => {
+        if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
           cb(err, null)
         } else {
           cb(null, null)

package.json

@@ -25,10 +25,9 @@
     "tap": "^14.0.0"
   },
   "dependencies": {
-    "end-of-stream": "^1.4.1",
     "fastq": "^1.6.0",
-    "readable-stream": "^3.0.0",
     "http-parser-js": "^0.5.2",
+    "readable-stream": "^3.0.0",
     "retimer": "^2.0.0"
   }
 }

test/pool.js

@@ -6,7 +6,7 @@ const { Pool } = require('..')
 const { createServer } = require('http')
 const { EventEmitter } = require('events')
 const { promisify } = require('util')
-const eos = require('end-of-stream')
+const eos = require('readable-stream').finished
 
 test('basic get', (t) => {
   t.plan(6)