fix: Client.stream writableNeedDrain (#442)

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

@@ -374,19 +374,53 @@ class Parser extends HTTPParser {
  this.headers = null
  this.shouldKeepAlive = false
  this.request = null
+ this.paused = false
+
+ // Parser can't be paused from within a callback.
+ // Use a buffer in JS land in order to stop further progress while paused.
+ this.resuming = false
+ this.queue = []
 
  this._resume = () => {
- // TODO: Resume parser.
+ if (!this.paused || this.resuming) {
+ return
+ }
+
+ this.paused = false
+
+ this.resuming = true
+ while (this.queue.length) {
+ const [fn, ...args] = this.queue.shift()
+
+ Reflect.apply(fn, this, args)
+
+ if (this.paused) {
+ this.resuming = false
+ return
+ }
+ }
+ this.resuming = false
+
  socketResume(socket)
  }
 
  this._pause = () => {
- // TODO: Pause parser.
+ if (this.paused) {
+ return
+ }
+
+ this.paused = true
+
  socketPause(socket)
  }
  }
 
  [HTTPParser.kOnHeaders] (rawHeaders) {
+ if (this.paused) {
+ this.queue.push([this[HTTPParser.kOnHeaders], rawHeaders])
+ return
+ }
+
  if (this.headers) {
  Array.prototype.push.apply(this.headers, rawHeaders)
  } else {
@@ -395,6 +429,11 @@ class Parser extends HTTPParser {
  }
 
  [HTTPParser.kOnExecute] (ret) {
+ if (this.paused) {
+ this.queue.push([this[HTTPParser.kOnExecute], ret])
+ return
+ }
+
  const { upgrade, socket } = this
 
  if (!Number.isFinite(ret)) {
@@ -466,6 +505,12 @@ class Parser extends HTTPParser {
 
  [HTTPParser.kOnHeadersComplete] (versionMajor, versionMinor, rawHeaders, method,
  url, statusCode, statusMessage, upgrade, shouldKeepAlive) {
+ if (this.paused) {
+ this.queue.push([this[HTTPParser.kOnHeadersComplete], versionMajor, versionMinor, rawHeaders, method,
+ url, statusCode, statusMessage, upgrade, shouldKeepAlive])
+ return
+ }
+
  const { client, socket } = this
 
  const request = client[kQueue][client[kRunningIdx]]
@@ -561,6 +606,11 @@ class Parser extends HTTPParser {
  }
 
  [HTTPParser.kOnBody] (chunk, offset, length) {
+ if (this.paused) {
+ this.queue.push([this[HTTPParser.kOnBody], chunk, offset, length])
+ return
+ }
+
  const { socket, statusCode, request } = this
 
  if (socket.destroyed) {
@@ -579,13 +629,19 @@ class Parser extends HTTPParser {
  }
 
  [HTTPParser.kOnMessageComplete] () {
+ if (this.paused) {
+ this.queue.push([this[HTTPParser.kOnMessageComplete]])
+ return
+ }
+
  const { client, socket, statusCode, headers, upgrade, request, trailers } = this
 
  if (socket.destroyed) {
  return
  }
 
  assert(statusCode >= 100)
+ assert(this.resuming || (socket._handle && socket._handle.reading))
 
  if (upgrade) {
  // TODO: When, how and why does this happen?
@@ -643,7 +699,6 @@ class Parser extends HTTPParser {
  // have been queued since then.
  util.destroy(socket, new InformationalError('reset'))
  } else {
- socketResume(socket)
  resume(client)
  }
  }

lib/core/request.js

@@ -162,6 +162,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,46 @@ 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, () => {
+ const client = new Client(`https://localhost:${server.address().port}`)
+ t.tearDown(() => {
+ 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
+ })
+
+ setImmediate(() => {
+ dst.write = (...args) => {
+ orgWrite.call(dst, ...args)
+ }
+ dst.resume()
+ })
+
+ p.then(() => {
+ t.pass()
+ })
+ })
+})