fixup

README.md

@@ -431,16 +431,18 @@ Options:
 
 The `handler` parameter is defined as follow:
 
+* `onConnect(abort)`, invoked before request is dispatched on socket.
+ * `abort(): Void`, abort request.
 * `onUpgrade(statusCode, headers, socket): Void`, invoked when request is upgraded either due to a `Upgrade` header or `CONNECT` method.
  * `statusCode: Number`
  * `headers: Array|Null`
  * `socket: Duplex`
-* `onHeaders(statusCode, headers, resume): Void`, invoked when statusCode and headers have been received. 
+* `onHeaders(statusCode, headers, resume): Void`, 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.
  * `resume(): Void`, resume `onData` after returning `false`.
-* `onData(chunk): Null|Boolean`, invoked when response payload data is received.
+* `onData(chunk): Boolean`, invoked when response payload data is received.
  * `chunk: Buffer`
 * `onComplete(trailers): Void`, invoked when response payload and trailers have been received and the request has completed.
  * `trailers: Array|Null`

benchmarks/index.js

@@ -149,7 +149,11 @@ class NoopRequest {
  this.deferred = deferred
  }
 
- onHeaders () {
+ onConnect (abort) {
+
+ }
+
+ onHeaders (statusCode, headers, resume) {
 
  }
 
@@ -167,6 +171,9 @@ class SimpleRequest {
  this.dst = dst
  }
 
+ onConnect (abort) {
+ }
+
  onHeaders (statusCode, headers, resume) {
  this.dst.on('drain', resume)
  }

lib/client-connect.js

@@ -18,6 +18,9 @@ class ConnectHandler extends AsyncResource {
  this.callback = callback
  }
 
+ onConnect (abort) {
+ }
+
  onUpgrade (statusCode, headers, socket) {
  const { callback, opaque } = this
 

lib/client-pipeline.js

@@ -45,12 +45,15 @@ class PipelineRequest extends Readable {
 
 class PipelineResponse extends Readable {
  constructor (resume) {
- super({ autoDestroy: true, read: resume })
+ super({ autoDestroy: true })
+ this[kResume] = resume
  }
 
- _destroy (err, callback) {
- this._read()
+ _read () {
+ this[kResume]()
+ }
 
+ _destroy (err, callback) {
  if (!err && !this._readableState.endEmitted) {
  err = new RequestAbortedError()
  }
@@ -77,6 +80,7 @@ class PipelineHandler extends AsyncResource {
 
  this.opaque = opts.opaque || null
  this.handler = handler
+ this.abort = null
 
  this.req = new PipelineRequest()
 
@@ -100,12 +104,16 @@ class PipelineHandler extends AsyncResource {
  }
  },
  destroy: (err, callback) => {
- const { body, req, res, ret } = this
+ const { body, req, res, ret, abort } = this
 
  if (!err && !ret._readableState.endEmitted) {
  err = new RequestAbortedError()
  }
 
+ if (abort && err) {
+ abort()
+ }
+
  util.destroy(body, err)
  util.destroy(req, err)
  util.destroy(res, err)
@@ -122,6 +130,15 @@ class PipelineHandler extends AsyncResource {
  this.res = null
  }
 
+ onConnect (abort) {
+ const { ret } = this
+ if (ret.destroyed) {
+ abort()
+ } else {
+ this.abort = abort
+ }
+ }
+
  onHeaders (statusCode, headers, resume) {
  const { opaque, handler } = this
 
@@ -180,29 +197,17 @@ class PipelineHandler extends AsyncResource {
 
  onData (chunk) {
  const { res } = this
-
- if (res._readableState.destroyed) {
- return
- }
-
  return res.push(chunk)
  }
 
  onComplete (trailers) {
  const { res } = this
-
- if (res._readableState.destroyed) {
- return
- }
-
  res.push(null)
  }
 
  onError (err) {
  const { ret } = this
-
  this.handler = null
-
  util.destroy(ret, err)
  }
 }

lib/client-request.js

@@ -8,18 +8,23 @@ const {
 const util = require('./util')
 const { AsyncResource } = require('async_hooks')
 
+const kAbort = Symbol('abort')
+
 class RequestResponse extends Readable {
- constructor (resume) {
+ constructor (resume, abort) {
  super({ autoDestroy: true, read: resume })
+ this[kAbort] = abort
  }
 
  _destroy (err, callback) {
- this._read()
-
  if (!err && !this._readableState.endEmitted) {
  err = new RequestAbortedError()
  }
 
+ if (err) {
+ this[kAbort]()
+ }
+
  callback(err)
  }
 }
@@ -39,16 +44,21 @@ class RequestHandler extends AsyncResource {
  this.opaque = opts.opaque || null
  this.callback = callback
  this.res = null
+ this.abort = null
+ }
+
+ onConnect (abort) {
+ this.abort = abort
  }
 
  onHeaders (statusCode, headers, resume) {
- const { callback, opaque } = this
+ const { callback, opaque, abort } = this
 
  if (statusCode < 200) {
  return
  }
 
- const body = new RequestResponse(resume)
+ const body = new RequestResponse(resume, abort)
 
  this.callback = null
  this.res = body
@@ -63,21 +73,11 @@ class RequestHandler extends AsyncResource {
 
  onData (chunk) {
  const { res } = this
-
- if (res._readableState.destroyed) {
- return
- }
-
  return res.push(chunk)
  }
 
  onComplete (trailers) {
  const { res } = this
-
- if (res._readableState.destroyed) {
- return
- }
-
  res.push(null)
  }
 

lib/client-stream.js

@@ -28,9 +28,14 @@ class StreamHandler extends AsyncResource {
  this.factory = factory
  this.callback = callback
  this.res = null
+ this.abort = null
  this.trailers = null
  }
 
+ onConnect (abort) {
+ this.abort = abort
+ }
+
  onHeaders (statusCode, headers, resume) {
  const { factory, opaque } = this
 
@@ -57,13 +62,17 @@ class StreamHandler extends AsyncResource {
  res.on('drain', resume)
  // TODO: Avoid finished. It registers an unecessary amount of listeners.
  finished(res, { readable: false }, (err) => {
- const { callback, res, opaque, trailers } = this
+ const { callback, res, opaque, trailers, abort } = this
 
  this.res = null
  if (err || !res.readable) {
  util.destroy(res, err)
  }
 
+ if (err) {
+ abort()
+ }
+
  this.callback = null
  this.runInAsyncScope(callback, null, err || null, { opaque, trailers })
  })
@@ -73,23 +82,12 @@ class StreamHandler extends AsyncResource {
 
  onData (chunk) {
  const { res } = this
-
- if (util.isDestroyed(res)) {
- return
- }
-
  return res.write(chunk)
  }
 
  onComplete (trailers) {
  const { res } = this
-
- if (util.isDestroyed(res)) {
- return
- }
-
  this.trailers = trailers ? util.parseHeaders(trailers) : {}
-
  res.end()
  }
 
@@ -101,7 +99,7 @@ class StreamHandler extends AsyncResource {
  if (res) {
  this.res = null
  util.destroy(res, err)
- } else {
+ } else if (callback) {
  this.callback = null
  this.runInAsyncScope(callback, null, err, { opaque })
  }

lib/client-upgrade.js

@@ -18,6 +18,9 @@ class UpgradeHandler extends AsyncResource {
  this.callback = callback
  }
 
+ onConnect (abort) {
+ }
+
  onUpgrade (statusCode, headers, socket) {
  const { callback, opaque } = this
 

lib/client.js

@@ -166,6 +166,9 @@ class Client extends EventEmitter {
  this[kResuming] = 0 // 0, idle, 1, scheduled, 2 resuming
  this[kDrained] = false
  this[kMaxAbortedPayload] = maxAbortedPayload || 1048576
+ this[kResume] = () => {
+ resume(this)
+ }
 
  // kQueue is built up of 3 sections separated by
  // the kRunningIdx and kPendingIdx indices.
@@ -932,11 +935,10 @@ function _resume (client) {
  return
  }
 
- try {
- write(client, request)
+ if (write(client, request)) {
  client[kPendingIdx]++
- } catch (err) {
- request.onError(err)
+ } else {
+ client[kQueue].splice(client[kPendingIdx], 1)
  }
  }
 }
@@ -965,7 +967,11 @@ function write (client, request) {
  }
 
  if (request.contentLength !== null && request.contentLength !== contentLength) {
- throw new ContentLengthMismatchError()
+ request.onError(new ContentLengthMismatchError())
+ }
+
+ if (!request.onConnect(client[kSocket])) {
+ return false
  }
 
  // TODO: Expect: 100-continue
@@ -1114,6 +1120,8 @@ function write (client, request) {
 
  client[kWriting] = true
  }
+
+ return true
 }
 
 module.exports = Client