Skip to content
31 changes: 17 additions & 14 deletions lib/client.js
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ const net = require('net')
const http = require('http')
const { pipeline } = require('stream')
const util = require('./core/util')
const { channels } = require('./core/diagnostics')
const timers = require('./timers')
const Request = require('./core/request')
const DispatcherBase = require('./dispatcher-base')
Expand Down Expand Up @@ -108,20 +109,20 @@ const FastBuffer = Buffer[Symbol.species]

const kClosedResolve = Symbol('kClosedResolve')

const channels = {}

try {
const diagnosticsChannel = require('diagnostics_channel')
channels.sendHeaders = diagnosticsChannel.channel('undici:client:sendHeaders')
channels.beforeConnect = diagnosticsChannel.channel('undici:client:beforeConnect')
channels.connectError = diagnosticsChannel.channel('undici:client:connectError')
channels.connected = diagnosticsChannel.channel('undici:client:connected')
} catch {
channels.sendHeaders = { hasSubscribers: false }
channels.beforeConnect = { hasSubscribers: false }
channels.connectError = { hasSubscribers: false }
channels.connected = { hasSubscribers: false }
}
// const channels = {}

// try {
// const diagnosticsChannel = require('diagnostics_channel')
// channels.sendHeaders = diagnosticsChannel.channel('undici:client:sendHeaders')
// channels.beforeConnect = diagnosticsChannel.channel('undici:client:beforeConnect')
// channels.connectError = diagnosticsChannel.channel('undici:client:connectError')
// channels.connected = diagnosticsChannel.channel('undici:client:connected')
// } catch {
// channels.sendHeaders = { hasSubscribers: false }
// channels.beforeConnect = { hasSubscribers: false }
// channels.connectError = { hasSubscribers: false }
// channels.connected = { hasSubscribers: false }
// }

/**
* @type {import('../types/client').default}
Expand Down Expand Up @@ -1191,6 +1192,7 @@ async function connect (client) {
hostname,
protocol,
port,
version: client[kHTTPConnVersion],
servername: client[kServerName],
localAddress: client[kLocalAddress]
},
Expand Down Expand Up @@ -1284,6 +1286,7 @@ async function connect (client) {
hostname,
protocol,
port,
version: client[kHTTPConnVersion],
servername: client[kServerName],
localAddress: client[kLocalAddress]
},
Expand Down
214 changes: 214 additions & 0 deletions lib/core/diagnostics.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,214 @@
let channels
try {
const diagnosticsChannel = require('diagnostics_channel')
let isClientSet = false
channels = {
// Client
beforeConnect: diagnosticsChannel.channel('undici:client:beforeConnect'),
connected: diagnosticsChannel.channel('undici:client:connected'),
connectError: diagnosticsChannel.channel('undici:client:connectError'),
sendHeaders: diagnosticsChannel.channel('undici:client:sendHeaders'),
// Request
create: diagnosticsChannel.channel('undici:request:create'),
bodySent: diagnosticsChannel.channel('undici:request:bodySent'),
headers: diagnosticsChannel.channel('undici:request:headers'),
trailers: diagnosticsChannel.channel('undici:request:trailers'),
error: diagnosticsChannel.channel('undici:request:error'),
// WebSocket
open: diagnosticsChannel.channel('undici:websocket:open'),
close: diagnosticsChannel.channel('undici:websocket:close'),
socketError: diagnosticsChannel.channel('undici:websocket:socket_error'),
ping: diagnosticsChannel.channel('undici:websocket:ping'),
pong: diagnosticsChannel.channel('undici:websocket:pong')
}

if (process.env.NODE_DEBUG.match(/(fetch|undici)/) != null) {
// Track all Client events
diagnosticsChannel.channel('undici:client:beforeConnect').subscribe(evt => {
const {
connectParams: { version, protocol, port, host }
} = evt
console.log(
`HTTP:undici ${process.pid}: connecting to ${host}${
port ? `:${port}` : ''
} using ${protocol}${version}`
)
})

diagnosticsChannel.channel('undici:client:connected').subscribe(evt => {
const {
connectParams: { version, protocol, port, host }
} = evt
console.log(
`HTTP:undici ${process.pid}: connected to ${host}${
port ? `:${port}` : ''
} using ${protocol}${version}`
)
})

diagnosticsChannel.channel('undici:client:connectError').subscribe(evt => {
const {
connectParams: { version, protocol, port, host },
error
} = evt
console.log(
`HTTP:undici ${process.pid}: connection to ${host}${
port ? `:${port}` : ''
} using ${protocol}${version} errored - ${error.message}`
)
})

diagnosticsChannel.channel('undici:client:sendHeaders').subscribe(evt => {
const {
request: { method, path, origin }
} = evt
console.log(
`HTTP:undici ${process.pid}: sending request to ${method} ${origin}/${path}`
)
})

// Track Request events
diagnosticsChannel.channel('undici:request:headers').subscribe(evt => {
const {
request: { method, path, origin },
response: { statusCode }
} = evt
console.log(
`HTTP:undici ${process.pid}: received response ${method} ${origin}/${path} - HTTP ${statusCode}`
)
})

diagnosticsChannel.channel('undici:request:trailers').subscribe(evt => {
const {
request: { method, path, origin }
} = evt
console.log(
`HTTP:undici ${process.pid}: trailers received from ${method} ${origin}/${path}`
)
})

diagnosticsChannel.channel('undici:request:error').subscribe(evt => {
const {
request: { method, path, origin },
error
} = evt
console.log(
`HTTP:undici ${process.pid}: request errored ${method} ${origin}/${path} - ${error.message}`
)
})

isClientSet = true
}

if (process.env.NODE_DEBUG.match(/websocket/) != null) {
if (!isClientSet) {
diagnosticsChannel
.channel('undici:client:beforeConnect')
.subscribe(evt => {
const {
connectParams: { version, protocol, port, host }
} = evt
console.log(
`HTTP:undici ${process.pid}: connecting to ${host}${
port ? `:${port}` : ''
} using ${protocol}${version}`
)
})

diagnosticsChannel.channel('undici:client:connected').subscribe(evt => {
const {
connectParams: { version, protocol, port, host }
} = evt
console.log(
`HTTP:undici ${process.pid}: connected to ${host}${
port ? `:${port}` : ''
} using ${protocol}${version}`
)
})

diagnosticsChannel
.channel('undici:client:connectError')
.subscribe(evt => {
const {
connectParams: { version, protocol, port, host },
error
} = evt
console.log(
`HTTP:undici ${process.pid}: connection to ${host}${
port ? `:${port}` : ''
} using ${protocol}${version} errored - ${error.message}`
)
})

diagnosticsChannel.channel('undici:client:sendHeaders').subscribe(evt => {
const {
request: { method, path, origin }
} = evt
console.log(
`HTTP:undici ${process.pid}: sending request to ${method} ${origin}/${path}`
)
})
}

// Track all Client events
diagnosticsChannel.channel('undici:websocket:open').subscribe(evt => {
const {
address: { address, port },
protocol,
extensions
} = evt
console.log(
`WebSocket:undici ${process.pid}: connection opened ${address}${
port ? `:${port}` : ''
} using ${protocol}-${extensions}`
)
})

diagnosticsChannel.channel('undici:websocket:close').subscribe(evt => {
const { websocket, code, reason } = evt
console.log(
`WebSocket:undici ${process.pid}: closed connection to ${websocket.url} - ${code} ${reason}`
)
})

diagnosticsChannel
.channel('undici:websocket:socket_error')
.subscribe(err => {
console.log(
`WebSocket:undici ${process.pid}: connection errored - ${err.message}`
)
})

diagnosticsChannel.channel('undici:websocket:ping').subscribe(evt => {
console.log(`WebSocket:undici ${process.pid}: ping received`)
})

diagnosticsChannel.channel('undici:websocket:pong').subscribe(evt => {
console.log(`WebSocket:undici ${process.pid}: pong received`)
})
}
} catch (error) {
channels = {
// Client
sendHeaders: { hasSubcribers: false },
beforeConnect: { hasSubcribers: false },
connectError: { hasSubcribers: false },
connected: { hasSubcribers: false },
// Request
create: { hasSubcribers: false },
bodySent: { hasSubcribers: false },
headers: { hasSubcribers: false },
trailers: { hasSubcribers: false },
error: { hasSubcribers: false },
// WebSocket
open: { hasSubcribers: false },
close: { hasSubcribers: false },
socketError: { hasSubcribers: false },
ping: { hasSubcribers: false },
pong: { hasSubcribers: false }
}
}

module.exports = {
channels
}
31 changes: 16 additions & 15 deletions lib/core/request.js
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ const {
const assert = require('assert')
const { kHTTP2BuildRequest, kHTTP2CopyHeaders, kHTTP1BuildRequest } = require('./symbols')
const util = require('./util')
const { channels } = require('./diagnostics.js')
const { headerNameLowerCasedRecord } = require('./constants')

// headerCharRegex have been lifted from
Expand All @@ -25,24 +26,24 @@ const invalidPathRegex = /[^\u0021-\u00ff]/

const kHandler = Symbol('handler')

const channels = {}
// const channels = {}

let extractBody

try {
const diagnosticsChannel = require('diagnostics_channel')
channels.create = diagnosticsChannel.channel('undici:request:create')
channels.bodySent = diagnosticsChannel.channel('undici:request:bodySent')
channels.headers = diagnosticsChannel.channel('undici:request:headers')
channels.trailers = diagnosticsChannel.channel('undici:request:trailers')
channels.error = diagnosticsChannel.channel('undici:request:error')
} catch {
channels.create = { hasSubscribers: false }
channels.bodySent = { hasSubscribers: false }
channels.headers = { hasSubscribers: false }
channels.trailers = { hasSubscribers: false }
channels.error = { hasSubscribers: false }
}
// try {
// const diagnosticsChannel = require('diagnostics_channel')
// channels.create = diagnosticsChannel.channel('undici:request:create')
// channels.bodySent = diagnosticsChannel.channel('undici:request:bodySent')
// channels.headers = diagnosticsChannel.channel('undici:request:headers')
// channels.trailers = diagnosticsChannel.channel('undici:request:trailers')
// channels.error = diagnosticsChannel.channel('undici:request:error')
// } catch {
// channels.create = { hasSubscribers: false }
// channels.bodySent = { hasSubscribers: false }
// channels.headers = { hasSubscribers: false }
// channels.trailers = { hasSubscribers: false }
// channels.error = { hasSubscribers: false }
// }

class Request {
constructor (origin, {
Expand Down
10 changes: 5 additions & 5 deletions lib/websocket/connection.js
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
'use strict'

const diagnosticsChannel = require('diagnostics_channel')
const { uid, states } = require('./constants')
const {
kReadyState,
Expand All @@ -9,17 +8,18 @@ const {
kReceivedClose
} = require('./symbols')
const { fireEvent, failWebsocketConnection } = require('./util')
const { channels } = require('../core/diagnostics')
const { CloseEvent } = require('./events')
const { makeRequest } = require('../fetch/request')
const { fetching } = require('../fetch/index')
const { Headers } = require('../fetch/headers')
const { getGlobalDispatcher } = require('../global')
const { kHeadersList } = require('../core/symbols')

const channels = {}
channels.open = diagnosticsChannel.channel('undici:websocket:open')
channels.close = diagnosticsChannel.channel('undici:websocket:close')
channels.socketError = diagnosticsChannel.channel('undici:websocket:socket_error')
// const channels = {}
// channels.open = diagnosticsChannel.channel('undici:websocket:open')
// channels.close = diagnosticsChannel.channel('undici:websocket:close')
// channels.socketError = diagnosticsChannel.channel('undici:websocket:socket_error')

/** @type {import('crypto')} */
let crypto
Expand Down
8 changes: 4 additions & 4 deletions lib/websocket/receiver.js
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
'use strict'

const { Writable } = require('stream')
const diagnosticsChannel = require('diagnostics_channel')
const { parserStates, opcodes, states, emptyBuffer } = require('./constants')
const { kReadyState, kSentClose, kResponse, kReceivedClose } = require('./symbols')
const { channels } = require('../core/diagnostics')
const { isValidStatusCode, failWebsocketConnection, websocketMessageReceived } = require('./util')
const { WebsocketFrameSend } = require('./frame')

Expand All @@ -12,9 +12,9 @@ const { WebsocketFrameSend } = require('./frame')
// Copyright (c) 2013 Arnout Kazemier and contributors
// Copyright (c) 2016 Luigi Pinca and contributors

const channels = {}
channels.ping = diagnosticsChannel.channel('undici:websocket:ping')
channels.pong = diagnosticsChannel.channel('undici:websocket:pong')
// const channels = {}
// channels.ping = diagnosticsChannel.channel('undici:websocket:ping')
// channels.pong = diagnosticsChannel.channel('undici:websocket:pong')

class ByteParser extends Writable {
#buffers = []
Expand Down