fix: Client.stream writableNeedDrain

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

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/request.js

@@ -156,6 +156,7 @@ class Request {
 
   onBody (chunk, offset, length) {
     assert(!this.aborted)
+    assert(!this[kPaused])
 
     if (this[kTimeout] && this[kTimeout].refresh) {
       this[kTimeout].refresh()

test/client-stream.js

@@ -657,3 +657,48 @@ test('stream body destroyed on invalid callback', (t) => {
     }
   })
 })
+
+test('stream needDrain', (t) => {
+  t.plan(1)
+
+  const server = createServer((req, res) => {
+    res.end(Buffer.alloc(4096))
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, async () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    t.tearDown(() => {
+      console.error(3)
+      client.destroy()
+    })
+
+    const dst = new PassThrough()
+    dst.pause()
+
+    while (dst.write(Buffer.alloc(4096))) {
+
+    }
+
+    const orgWrite = dst.write
+    dst.write = () => t.fail()
+    const p = client.stream({
+      path: '/',
+      method: 'GET'
+    }, () => {
+      return dst
+    })
+
+    setTimeout(() => {
+      dst.write = (...args) => {
+        console.error("ASD")
+        orgWrite.call(dst, ...args)
+      }
+      dst.resume()
+    }, 1e3)
+
+    await p
+
+    t.pass()
+  })
+})