fix: Client.stream writableNeedDrain

Fixes: #441
Refs: nodejs/node#35348
Refs: nodejs/node#35341

README.md

@@ -473,7 +473,7 @@ The `handler` parameter is defined as follow:
   * `statusCode: Number`
   * `headers: Array|Null`
   * `socket: Duplex`
-* `onHeaders(statusCode, headers, resume): Void`, invoked when statusCode and headers have been received.
+* `onHeaders(statusCode, headers, resume): Boolean`, invoked when statusCode and headers have been received.
   May be invoked multiple times due to 1xx informational headers.
   * `statusCode: Number`
   * `headers: Array|Null`, an array of key-value pairs. Keys are not automatically lowercased.

lib/client-stream.js

@@ -109,6 +109,12 @@ class StreamHandler extends AsyncResource {
     })
 
     this.res = res
+
+    const needDrain = res.writableNeedDrain !== undefined
+      ? res.writableNeedDrain
+      : res._writableState && res._writableState.needDrain
+
+    return needDrain === true
   }
 
   onData (chunk) {

lib/core/client.js

@@ -535,7 +535,9 @@ class Parser extends HTTPParser {
     }
 
     try {
-      request.onHeaders(statusCode, headers, statusCode < 200 ? null : socket[kResume])
+      if (request.onHeaders(statusCode, headers, statusCode < 200 ? null : socket[kResume]) === false) {
+        socket[kPause]()
+      }
     } catch (err) {
       util.destroy(socket, err)
       return 1

lib/core/request.js

@@ -117,7 +117,7 @@ class Request {
 
     clearRequestTimeout(this)
 
-    this[kHandler].onHeaders(statusCode, headers, resume)
+    return this[kHandler].onHeaders(statusCode, headers, resume)
   }
 
   onBody (chunk, offset, length) {