feat: onInfo & onTrailers options

Fixes: #179
Fixes: #196

README.md

@@ -106,6 +106,9 @@ Options:
   If `false` the request won't be sent until all preceeding
   requests in the pipeline has completed.
   Default: `true` if `method` is `HEAD` or `GET`.
+* `onInfo(data)`, function invoked for informational
+  1xx responses. `100 Continue` is ignored and will not invoke `onInfo`.
+* `onTrailers(data)`, function invoked for trailers.
 
 Headers are represented by an object like this:
 
@@ -130,6 +133,17 @@ The `data` parameter in `callback` is defined as follow:
   either fully consume or destroy the body unless there is an error, or no further requests
   will be processed.
 
+The `data` parameter in `onInfo` is defined as follow:
+
+* `statusCode: Number`
+* `opaque: Any`
+* `headers: Object`, an object where all keys have been lowercased.
+
+The `data` parameter in `onTrailers` is defined as follow:
+
+* `opaque: Any`
+* `trailers: Object`, an object where all keys have been lowercased.
+
 Returns a promise if no callback is provided.
 
 Example:
@@ -243,6 +257,7 @@ The `data` parameter in `factory` is defined as follow:
 The `data` parameter in `callback` is defined as follow:
 
 * `opaque: Any`
+* `trailers: Object`, an object where all keys have been lowercased.
 
 Returns a promise if no callback is provided.
 

lib/client-pipeline.js

@@ -21,8 +21,28 @@ class PipelineRequest extends Request {
   constructor (client, opts, callback) {
     super(opts, client)
 
+    if (opts.onInfo && typeof opts.onInfo !== 'function') {
+      throw new InvalidArgumentError('invalid opts.onInfo')
+    }
+
+    if (opts.onTrailers && typeof opts.onTrailers !== 'function') {
+      throw new InvalidArgumentError('invalid opts.onTrailers')
+    }
+
     this.callback = callback
     this.aborted = false
+    this.onInfo = opts.onInfo
+    this.onTrailers = opts.onTrailers
+  }
+
+  _onInfo (statusCode, headers) {
+    if (this.onInfo) {
+      try {
+        this.onInfo({ statusCode, headers, opaque: this.opaque })
+      } catch (err) {
+        this.onError(err)
+      }
+    }
   }
 
   _onHeaders (statusCode, headers, resume) {
@@ -42,12 +62,18 @@ class PipelineRequest extends Request {
     return this.res(null, chunk)
   }
 
-  _onComplete () {
-    // TODO: Trailers?
-
+  _onComplete (trailers) {
     const res = this.res
     this.res = null
     res(null, null)
+
+    if (trailers && this.onTrailers) {
+      try {
+        this.onTrailers({ trailers, opaque: this.opaque })
+      } catch (err) {
+        this.onError(err)
+      }
+    }
   }
 
   _onError (err) {

lib/client-request.js

@@ -47,10 +47,30 @@ class RequestRequest extends Request {
       throw new InvalidArgumentError('invalid method')
     }
 
+    if (opts.onInfo && typeof opts.onInfo !== 'function') {
+      throw new InvalidArgumentError('invalid opts.onInfo')
+    }
+
+    if (opts.onTrailers && typeof opts.onTrailers !== 'function') {
+      throw new InvalidArgumentError('invalid opts.onTrailers')
+    }
+
     super(opts, client)
 
     this.callback = callback
     this.res = null
+    this.onInfo = opts.onInfo
+    this.onTrailers = opts.onTrailers
+  }
+
+  _onInfo (statusCode, headers) {
+    if (this.onInfo) {
+      try {
+        this.onInfo({ statusCode, headers })
+      } catch (err) {
+        this.onError(err)
+      }
+    }
   }
 
   _onHeaders (statusCode, headers, resume) {
@@ -67,8 +87,18 @@ class RequestRequest extends Request {
     return this.res.push(chunk)
   }
 
-  _onComplete () {
-    this.res.push(null)
+  _onComplete (trailers) {
+    const { res, opaque } = this
+
+    res.push(null)
+
+    if (trailers && this.onTrailers) {
+      try {
+        this.onTrailers({ trailers, opaque })
+      } catch (err) {
+        this.onError(err)
+      }
+    }
   }
 
   _onError (err) {

lib/client-stream.js

@@ -23,11 +23,34 @@ class StreamRequest extends Request {
       throw new InvalidArgumentError('invalid method')
     }
 
+    if (opts.onInfo && typeof opts.onInfo !== 'function') {
+      throw new InvalidArgumentError('invalid opts.onInfo')
+    }
+
+    if (opts.onTrailers && typeof opts.onTrailers !== 'function') {
+      throw new InvalidArgumentError('invalid opts.onTrailers')
+    }
+
     super(opts, client)
 
     this.factory = factory
     this.callback = callback
     this.res = null
+    this.trailers = null
+    this.onInfo = opts.onInfo
+    this.onTrailers = opts.onTrailers
+  }
+
+  _onInfo (statusCode, headers) {
+    const { opaque } = this
+
+    if (this.onInfo) {
+      try {
+        this.onInfo({ statusCode, headers, opaque })
+      } catch (err) {
+        this.onError(err)
+      }
+    }
   }
 
   _onHeaders (statusCode, headers, resume) {
@@ -68,15 +91,15 @@ class StreamRequest extends Request {
         return
       }
 
-      const { callback, res, opaque } = this
+      const { callback, res, opaque, trailers } = this
 
       this.res = null
       if (!res.readable) {
         util.destroy(res)
       }
 
       this.callback = null
-      callback(null, { opaque })
+      callback(null, { opaque, trailers })
     })
 
     if (typeof res.destroy === 'function') {
@@ -92,10 +115,24 @@ class StreamRequest extends Request {
     return res.write(chunk)
   }
 
-  _onComplete () {
-    const { res } = this
+  _onComplete (trailers) {
+    const { res, opaque } = this
+
+    if (!res) {
+      return
+    }
+
+    this.trailers = trailers || {}
 
     res.end()
+
+    if (trailers && this.onTrailers) {
+      try {
+        this.onTrailers({ trailers, opaque })
+      } catch (err) {
+        this.onError(err)
+      }
+    }
   }
 
   _onError (err) {

lib/client.js

@@ -903,6 +903,8 @@ function write (client, request) {
     socket.write(`content-length: ${contentLength}\r\n`, 'ascii')
   }
 
+  // TODO: Expect: 100-continue
+
   if (method === 'HEAD') {
     // https://github.com/mcollina/undici/issues/258
 

test/client-pipeline.js

@@ -893,3 +893,64 @@ test('pipeline body without destroy', (t) => {
       .resume()
   })
 })
+
+test('trailers', (t) => {
+  t.plan(2)
+
+  const server = createServer((req, res) => {
+    res.writeHead(200, { Trailer: 'Content-MD5' })
+    res.addTrailers({ 'Content-MD5': 'test' })
+    res.end()
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    t.tearDown(client.close.bind(client))
+
+    client.pipeline({
+      path: '/',
+      method: 'GET',
+      onTrailers: ({ trailers }) => {
+        t.strictDeepEqual({ 'content-md5': 'test' }, trailers)
+      }
+    }, ({ body }) => body)
+      .end()
+      .resume()
+      .on('end', () => {
+        t.pass()
+      })
+  })
+})
+
+test('info', (t) => {
+  t.plan(2)
+
+  const server = createServer((req, res) => {
+    res.writeProcessing()
+    req.pipe(res)
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    t.tearDown(client.close.bind(client))
+
+    let recv = ''
+
+    client.pipeline({
+      path: '/',
+      method: 'POST',
+      onInfo: ({ statusCode }) => {
+        t.strictEqual(statusCode, 102)
+      }
+    }, ({ body }) => body)
+      .end('hello')
+      .on('data', chunk => {
+        recv += chunk
+      })
+      .on('end', () => {
+        t.strictEqual(recv, 'hello')
+      })
+  })
+})

test/client-request.js

@@ -38,3 +38,64 @@ test('request abort before headers', (t) => {
     })
   })
 })
+
+test('trailers', (t) => {
+  t.plan(3)
+
+  const server = createServer((req, res) => {
+    res.writeHead(200, { Trailer: 'Content-MD5' })
+    res.addTrailers({ 'Content-MD5': 'test' })
+    res.end()
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    t.tearDown(client.close.bind(client))
+
+    client.request({
+      path: '/',
+      method: 'GET',
+      onTrailers: ({ trailers }) => {
+        t.strictDeepEqual({ 'content-md5': 'test' }, trailers)
+      }
+    }, (err, data) => {
+      t.error(err)
+      data.body.on('end', () => {
+        t.pass()
+      }).resume()
+    })
+  })
+})
+
+test('info', (t) => {
+  t.plan(3)
+
+  const server = createServer((req, res) => {
+    res.writeProcessing()
+    req.pipe(res)
+  })
+  t.tearDown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    t.tearDown(client.close.bind(client))
+
+    client.request({
+      path: '/',
+      method: 'POST',
+      body: 'hello',
+      onInfo: ({ statusCode, headers }) => {
+        t.strictEqual(statusCode, 102)
+      }
+    }, (err, data) => {
+      t.error(err)
+      let recv = ''
+      data.body.on('end', () => {
+        t.strictEqual(recv, 'hello')
+      }).on('data', buf => {
+        recv += buf
+      })
+    })
+  })
+})