5 files changed, 550 insertions, 0 deletions
diff --git a/test/diagnostics-channel/connect-error.js b/test/diagnostics-channel/connect-error.js
new file mode 100644
index 0000000..f7e842d
--- /dev/null
+++ b/test/diagnostics-channel/connect-error.js
@@ -0,0 +1,61 @@
+'use strict'
+
+const t = require('tap')
+
+let diagnosticsChannel
+
+try {
+  diagnosticsChannel = require('diagnostics_channel')
+} catch {
+  t.skip('missing diagnostics_channel')
+  process.exit(0)
+}
+
+const { Client } = require('../..')
+
+t.plan(16)
+
+const connectError = new Error('custom error')
+
+let _connector
+diagnosticsChannel.channel('undici:client:beforeConnect').subscribe(({ connectParams, connector }) => {
+  _connector = connector
+
+  t.equal(typeof _connector, 'function')
+  t.equal(Object.keys(connectParams).length, 6)
+
+  const { host, hostname, protocol, port, servername } = connectParams
+
+  t.equal(host, 'localhost:1234')
+  t.equal(hostname, 'localhost')
+  t.equal(port, '1234')
+  t.equal(protocol, 'http:')
+  t.equal(servername, null)
+})
+
+diagnosticsChannel.channel('undici:client:connectError').subscribe(({ error, connectParams, connector }) => {
+  t.equal(Object.keys(connectParams).length, 6)
+  t.equal(_connector, connector)
+
+  const { host, hostname, protocol, port, servername } = connectParams
+
+  t.equal(error, connectError)
+  t.equal(host, 'localhost:1234')
+  t.equal(hostname, 'localhost')
+  t.equal(port, '1234')
+  t.equal(protocol, 'http:')
+  t.equal(servername, null)
+})
+
+const client = new Client('http://localhost:1234', {
+  connect: (_, cb) => { cb(connectError, null) }
+})
+
+t.teardown(client.close.bind(client))
+
+client.request({
+  path: '/',
+  method: 'GET'
+}, (err, data) => {
+  t.equal(err, connectError)
+})
diff --git a/test/diagnostics-channel/error.js b/test/diagnostics-channel/error.js
new file mode 100644
index 0000000..1f350b1
--- /dev/null
+++ b/test/diagnostics-channel/error.js
@@ -0,0 +1,52 @@
+'use strict'
+
+const t = require('tap')
+
+let diagnosticsChannel
+
+try {
+  diagnosticsChannel = require('diagnostics_channel')
+} catch {
+  t.skip('missing diagnostics_channel')
+  process.exit(0)
+}
+
+const { Client } = require('../..')
+const { createServer } = require('http')
+
+t.plan(3)
+
+const server = createServer((req, res) => {
+  res.destroy()
+})
+t.teardown(server.close.bind(server))
+
+const reqHeaders = {
+  foo: undefined,
+  bar: 'bar'
+}
+
+let _req
+diagnosticsChannel.channel('undici:request:create').subscribe(({ request }) => {
+  _req = request
+})
+
+diagnosticsChannel.channel('undici:request:error').subscribe(({ request, error }) => {
+  t.equal(_req, request)
+  t.equal(error.code, 'UND_ERR_SOCKET')
+})
+
+server.listen(0, () => {
+  const client = new Client(`http://localhost:${server.address().port}`, {
+    keepAliveTimeout: 300e3
+  })
+  t.teardown(client.close.bind(client))
+
+  client.request({
+    path: '/',
+    method: 'GET',
+    headers: reqHeaders
+  }, (err, data) => {
+    t.equal(err.code, 'UND_ERR_SOCKET')
+  })
+})
diff --git a/test/diagnostics-channel/get.js b/test/diagnostics-channel/get.js
new file mode 100644
index 0000000..9d868c3
--- /dev/null
+++ b/test/diagnostics-channel/get.js
@@ -0,0 +1,141 @@
+'use strict'
+
+const t = require('tap')
+
+let diagnosticsChannel
+
+try {
+  diagnosticsChannel = require('diagnostics_channel')
+} catch {
+  t.skip('missing diagnostics_channel')
+  process.exit(0)
+}
+
+const { Client } = require('../..')
+const { createServer } = require('http')
+
+t.plan(32)
+
+const server = createServer((req, res) => {
+  res.setHeader('Content-Type', 'text/plain')
+  res.setHeader('trailer', 'foo')
+  res.write('hello')
+  res.addTrailers({
+    foo: 'oof'
+  })
+  res.end()
+})
+t.teardown(server.close.bind(server))
+
+const reqHeaders = {
+  foo: undefined,
+  bar: 'bar'
+}
+
+let _req
+diagnosticsChannel.channel('undici:request:create').subscribe(({ request }) => {
+  _req = request
+  t.equal(request.origin, `http://localhost:${server.address().port}`)
+  t.equal(request.completed, false)
+  t.equal(request.method, 'GET')
+  t.equal(request.path, '/')
+  t.equal(request.headers, 'bar: bar\r\n')
+  request.addHeader('hello', 'world')
+  t.equal(request.headers, 'bar: bar\r\nhello: world\r\n')
+})
+
+let _connector
+diagnosticsChannel.channel('undici:client:beforeConnect').subscribe(({ connectParams, connector }) => {
+  _connector = connector
+
+  t.equal(typeof _connector, 'function')
+  t.equal(Object.keys(connectParams).length, 6)
+
+  const { host, hostname, protocol, port, servername } = connectParams
+
+  t.equal(host, `localhost:${server.address().port}`)
+  t.equal(hostname, 'localhost')
+  t.equal(port, String(server.address().port))
+  t.equal(protocol, 'http:')
+  t.equal(servername, null)
+})
+
+let _socket
+diagnosticsChannel.channel('undici:client:connected').subscribe(({ connectParams, socket, connector }) => {
+  _socket = socket
+
+  t.equal(_connector, connector)
+  t.equal(Object.keys(connectParams).length, 6)
+
+  const { host, hostname, protocol, port, servername } = connectParams
+
+  t.equal(host, `localhost:${server.address().port}`)
+  t.equal(hostname, 'localhost')
+  t.equal(port, String(server.address().port))
+  t.equal(protocol, 'http:')
+  t.equal(servername, null)
+})
+
+diagnosticsChannel.channel('undici:client:sendHeaders').subscribe(({ request, headers, socket }) => {
+  t.equal(_req, request)
+  t.equal(_socket, socket)
+
+  const expectedHeaders = [
+    'GET / HTTP/1.1',
+    `host: localhost:${server.address().port}`,
+    'connection: keep-alive',
+    'bar: bar',
+    'hello: world'
+  ]
+
+  t.equal(headers, expectedHeaders.join('\r\n') + '\r\n')
+})
+
+diagnosticsChannel.channel('undici:request:headers').subscribe(({ request, response }) => {
+  t.equal(_req, request)
+  t.equal(response.statusCode, 200)
+  const expectedHeaders = [
+    Buffer.from('Content-Type'),
+    Buffer.from('text/plain'),
+    Buffer.from('trailer'),
+    Buffer.from('foo'),
+    Buffer.from('Date'),
+    response.headers[5], // This is a date
+    Buffer.from('Connection'),
+    Buffer.from('keep-alive'),
+    Buffer.from('Keep-Alive'),
+    Buffer.from('timeout=5'),
+    Buffer.from('Transfer-Encoding'),
+    Buffer.from('chunked')
+  ]
+  t.same(response.headers, expectedHeaders)
+  t.equal(response.statusText, 'OK')
+})
+
+let endEmitted = false
+diagnosticsChannel.channel('undici:request:trailers').subscribe(({ request, trailers }) => {
+  t.equal(request.completed, true)
+  t.equal(_req, request)
+  // This event is emitted after the last chunk has been added to the body stream,
+  // not when it was consumed by the application
+  t.equal(endEmitted, false)
+  t.same(trailers, [Buffer.from('foo'), Buffer.from('oof')])
+})
+
+server.listen(0, () => {
+  const client = new Client(`http://localhost:${server.address().port}`, {
+    keepAliveTimeout: 300e3
+  })
+  t.teardown(client.close.bind(client))
+
+  client.request({
+    path: '/',
+    method: 'GET',
+    headers: reqHeaders
+  }, (err, data) => {
+    t.error(err)
+    data.body.on('end', function () {
+      endEmitted = true
+    })
+  })
+})
diff --git a/test/diagnostics-channel/post-stream.js b/test/diagnostics-channel/post-stream.js
new file mode 100644
index 0000000..236b9bb
--- /dev/null
+++ b/test/diagnostics-channel/post-stream.js
@@ -0,0 +1,149 @@
+'use strict'
+
+const t = require('tap')
+const { Readable } = require('stream')
+
+let diagnosticsChannel
+
+try {
+  diagnosticsChannel = require('diagnostics_channel')
+} catch {
+  t.skip('missing diagnostics_channel')
+  process.exit(0)
+}
+
+const { Client } = require('../..')
+const { createServer } = require('http')
+
+t.plan(33)
+
+const server = createServer((req, res) => {
+  req.resume()
+  res.setHeader('Content-Type', 'text/plain')
+  res.setHeader('trailer', 'foo')
+  res.write('hello')
+  res.addTrailers({
+    foo: 'oof'
+  })
+  res.end()
+})
+t.teardown(server.close.bind(server))
+
+const reqHeaders = {
+  foo: undefined,
+  bar: 'bar'
+}
+const body = Readable.from(['hello', ' ', 'world'])
+
+let _req
+diagnosticsChannel.channel('undici:request:create').subscribe(({ request }) => {
+  _req = request
+  t.equal(request.completed, false)
+  t.equal(request.method, 'POST')
+  t.equal(request.path, '/')
+  t.equal(request.headers, 'bar: bar\r\n')
+  request.addHeader('hello', 'world')
+  t.equal(request.headers, 'bar: bar\r\nhello: world\r\n')
+  t.same(request.body, body)
+})
+
+let _connector
+diagnosticsChannel.channel('undici:client:beforeConnect').subscribe(({ connectParams, connector }) => {
+  _connector = connector
+
+  t.equal(typeof _connector, 'function')
+  t.equal(Object.keys(connectParams).length, 6)
+
+  const { host, hostname, protocol, port, servername } = connectParams
+
+  t.equal(host, `localhost:${server.address().port}`)
+  t.equal(hostname, 'localhost')
+  t.equal(port, String(server.address().port))
+  t.equal(protocol, 'http:')
+  t.equal(servername, null)
+})
+
+let _socket
+diagnosticsChannel.channel('undici:client:connected').subscribe(({ connectParams, socket, connector }) => {
+  _socket = socket
+
+  t.equal(Object.keys(connectParams).length, 6)
+  t.equal(_connector, connector)
+
+  const { host, hostname, protocol, port, servername } = connectParams
+
+  t.equal(host, `localhost:${server.address().port}`)
+  t.equal(hostname, 'localhost')
+  t.equal(port, String(server.address().port))
+  t.equal(protocol, 'http:')
+  t.equal(servername, null)
+})
+
+diagnosticsChannel.channel('undici:client:sendHeaders').subscribe(({ request, headers, socket }) => {
+  t.equal(_req, request)
+  t.equal(_socket, socket)
+
+  const expectedHeaders = [
+    'POST / HTTP/1.1',
+    `host: localhost:${server.address().port}`,
+    'connection: keep-alive',
+    'bar: bar',
+    'hello: world'
+  ]
+
+  t.equal(headers, expectedHeaders.join('\r\n') + '\r\n')
+})
+
+diagnosticsChannel.channel('undici:request:headers').subscribe(({ request, response }) => {
+  t.equal(_req, request)
+  t.equal(response.statusCode, 200)
+  const expectedHeaders = [
+    Buffer.from('Content-Type'),
+    Buffer.from('text/plain'),
+    Buffer.from('trailer'),
+    Buffer.from('foo'),
+    Buffer.from('Date'),
+    response.headers[5], // This is a date
+    Buffer.from('Connection'),
+    Buffer.from('keep-alive'),
+    Buffer.from('Keep-Alive'),
+    Buffer.from('timeout=5'),
+    Buffer.from('Transfer-Encoding'),
+    Buffer.from('chunked')
+  ]
+  t.same(response.headers, expectedHeaders)
+  t.equal(response.statusText, 'OK')
+})
+
+diagnosticsChannel.channel('undici:request:bodySent').subscribe(({ request }) => {
+  t.equal(_req, request)
+})
+
+let endEmitted = false
+diagnosticsChannel.channel('undici:request:trailers').subscribe(({ request, trailers }) => {
+  t.equal(request.completed, true)
+  t.equal(_req, request)
+  // This event is emitted after the last chunk has been added to the body stream,
+  // not when it was consumed by the application
+  t.equal(endEmitted, false)
+  t.same(trailers, [Buffer.from('foo'), Buffer.from('oof')])
+})
+
+server.listen(0, () => {
+  const client = new Client(`http://localhost:${server.address().port}`, {
+    keepAliveTimeout: 300e3
+  })
+  t.teardown(client.close.bind(client))
+
+  client.request({
+    path: '/',
+    method: 'POST',
+    headers: reqHeaders,
+    body
+  }, (err, data) => {
+    t.error(err)
+    data.body.on('end', function () {
+      endEmitted = true
+    })
+  })
+})
diff --git a/test/diagnostics-channel/post.js b/test/diagnostics-channel/post.js
new file mode 100644
index 0000000..fc02eb5
--- /dev/null
+++ b/test/diagnostics-channel/post.js
@@ -0,0 +1,147 @@
+'use strict'
+
+const t = require('tap')
+
+let diagnosticsChannel
+
+try {
+  diagnosticsChannel = require('diagnostics_channel')
+} catch {
+  t.skip('missing diagnostics_channel')
+  process.exit(0)
+}
+
+const { Client } = require('../..')
+const { createServer } = require('http')
+
+t.plan(33)
+
+const server = createServer((req, res) => {
+  req.resume()
+  res.setHeader('Content-Type', 'text/plain')
+  res.setHeader('trailer', 'foo')
+  res.write('hello')
+  res.addTrailers({
+    foo: 'oof'
+  })
+  res.end()
+})
+t.teardown(server.close.bind(server))
+
+const reqHeaders = {
+  foo: undefined,
+  bar: 'bar'
+}
+
+let _req
+diagnosticsChannel.channel('undici:request:create').subscribe(({ request }) => {
+  _req = request
+  t.equal(request.completed, false)
+  t.equal(request.method, 'POST')
+  t.equal(request.path, '/')
+  t.equal(request.headers, 'bar: bar\r\n')
+  request.addHeader('hello', 'world')
+  t.equal(request.headers, 'bar: bar\r\nhello: world\r\n')
+  t.same(request.body, Buffer.from('hello world'))
+})
+
+let _connector
+diagnosticsChannel.channel('undici:client:beforeConnect').subscribe(({ connectParams, connector }) => {
+  _connector = connector
+
+  t.equal(typeof _connector, 'function')
+  t.equal(Object.keys(connectParams).length, 6)
+
+  const { host, hostname, protocol, port, servername } = connectParams
+
+  t.equal(host, `localhost:${server.address().port}`)
+  t.equal(hostname, 'localhost')
+  t.equal(port, String(server.address().port))
+  t.equal(protocol, 'http:')
+  t.equal(servername, null)
+})
+
+let _socket
+diagnosticsChannel.channel('undici:client:connected').subscribe(({ connectParams, socket, connector }) => {
+  _socket = socket
+
+  t.equal(Object.keys(connectParams).length, 6)
+  t.equal(_connector, connector)
+
+  const { host, hostname, protocol, port, servername } = connectParams
+
+  t.equal(host, `localhost:${server.address().port}`)
+  t.equal(hostname, 'localhost')
+  t.equal(port, String(server.address().port))
+  t.equal(protocol, 'http:')
+  t.equal(servername, null)
+})
+
+diagnosticsChannel.channel('undici:client:sendHeaders').subscribe(({ request, headers, socket }) => {
+  t.equal(_req, request)
+  t.equal(_socket, socket)
+
+  const expectedHeaders = [
+    'POST / HTTP/1.1',
+    `host: localhost:${server.address().port}`,
+    'connection: keep-alive',
+    'bar: bar',
+    'hello: world'
+  ]
+
+  t.equal(headers, expectedHeaders.join('\r\n') + '\r\n')
+})
+
+diagnosticsChannel.channel('undici:request:headers').subscribe(({ request, response }) => {
+  t.equal(_req, request)
+  t.equal(response.statusCode, 200)
+  const expectedHeaders = [
+    Buffer.from('Content-Type'),
+    Buffer.from('text/plain'),
+    Buffer.from('trailer'),
+    Buffer.from('foo'),
+    Buffer.from('Date'),
+    response.headers[5], // This is a date
+    Buffer.from('Connection'),
+    Buffer.from('keep-alive'),
+    Buffer.from('Keep-Alive'),
+    Buffer.from('timeout=5'),
+    Buffer.from('Transfer-Encoding'),
+    Buffer.from('chunked')
+  ]
+  t.same(response.headers, expectedHeaders)
+  t.equal(response.statusText, 'OK')
+})
+
+diagnosticsChannel.channel('undici:request:bodySent').subscribe(({ request }) => {
+  t.equal(_req, request)
+})
+
+let endEmitted = false
+diagnosticsChannel.channel('undici:request:trailers').subscribe(({ request, trailers }) => {
+  t.equal(request.completed, true)
+  t.equal(_req, request)
+  // This event is emitted after the last chunk has been added to the body stream,
+  // not when it was consumed by the application
+  t.equal(endEmitted, false)
+  t.same(trailers, [Buffer.from('foo'), Buffer.from('oof')])
+})
+
+server.listen(0, () => {
+  const client = new Client(`http://localhost:${server.address().port}`, {
+    keepAliveTimeout: 300e3
+  })
+  t.teardown(client.close.bind(client))
+
+  client.request({
+    path: '/',
+    method: 'POST',
+    headers: reqHeaders,
+    body: 'hello world'
+  }, (err, data) => {
+    t.error(err)
+    data.body.on('end', function () {
+      endEmitted = true
+    })
+  })
+})