diff options
Diffstat (limited to 'vanilla/node_modules/undici/lib/handler')
8 files changed, 1790 insertions, 0 deletions
diff --git a/vanilla/node_modules/undici/lib/handler/cache-handler.js b/vanilla/node_modules/undici/lib/handler/cache-handler.js new file mode 100644 index 0000000..93a70e8 --- /dev/null +++ b/vanilla/node_modules/undici/lib/handler/cache-handler.js @@ -0,0 +1,561 @@ +'use strict' + +const util = require('../core/util') +const { + parseCacheControlHeader, + parseVaryHeader, + isEtagUsable +} = require('../util/cache') +const { parseHttpDate } = require('../util/date.js') + +function noop () {} + +// Status codes that we can use some heuristics on to cache +const HEURISTICALLY_CACHEABLE_STATUS_CODES = [ + 200, 203, 204, 206, 300, 301, 308, 404, 405, 410, 414, 501 +] + +// Status codes which semantic is not handled by the cache +// https://datatracker.ietf.org/doc/html/rfc9111#section-3 +// This list should not grow beyond 206 unless the RFC is updated +// by a newer one including more. Please introduce another list if +// implementing caching of responses with the 'must-understand' directive. +const NOT_UNDERSTOOD_STATUS_CODES = [ + 206 +] + +const MAX_RESPONSE_AGE = 2147483647000 + +/** + * @typedef {import('../../types/dispatcher.d.ts').default.DispatchHandler} DispatchHandler + * + * @implements {DispatchHandler} + */ +class CacheHandler { + /** + * @type {import('../../types/cache-interceptor.d.ts').default.CacheKey} + */ + #cacheKey + + /** + * @type {import('../../types/cache-interceptor.d.ts').default.CacheHandlerOptions['type']} + */ + #cacheType + + /** + * @type {number | undefined} + */ + #cacheByDefault + + /** + * @type {import('../../types/cache-interceptor.d.ts').default.CacheStore} + */ + #store + + /** + * @type {import('../../types/dispatcher.d.ts').default.DispatchHandler} + */ + #handler + + /** + * @type {import('node:stream').Writable | undefined} + */ + #writeStream + + /** + * @param {import('../../types/cache-interceptor.d.ts').default.CacheHandlerOptions} opts + * @param {import('../../types/cache-interceptor.d.ts').default.CacheKey} cacheKey + * @param {import('../../types/dispatcher.d.ts').default.DispatchHandler} handler + */ + constructor ({ store, type, cacheByDefault }, cacheKey, handler) { + this.#store = store + this.#cacheType = type + this.#cacheByDefault = cacheByDefault + this.#cacheKey = cacheKey + this.#handler = handler + } + + onRequestStart (controller, context) { + this.#writeStream?.destroy() + this.#writeStream = undefined + this.#handler.onRequestStart?.(controller, context) + } + + onRequestUpgrade (controller, statusCode, headers, socket) { + this.#handler.onRequestUpgrade?.(controller, statusCode, headers, socket) + } + + /** + * @param {import('../../types/dispatcher.d.ts').default.DispatchController} controller + * @param {number} statusCode + * @param {import('../../types/header.d.ts').IncomingHttpHeaders} resHeaders + * @param {string} statusMessage + */ + onResponseStart ( + controller, + statusCode, + resHeaders, + statusMessage + ) { + const downstreamOnHeaders = () => + this.#handler.onResponseStart?.( + controller, + statusCode, + resHeaders, + statusMessage + ) + const handler = this + + if ( + !util.safeHTTPMethods.includes(this.#cacheKey.method) && + statusCode >= 200 && + statusCode <= 399 + ) { + // Successful response to an unsafe method, delete it from cache + // https://www.rfc-editor.org/rfc/rfc9111.html#name-invalidating-stored-response + try { + this.#store.delete(this.#cacheKey)?.catch?.(noop) + } catch { + // Fail silently + } + return downstreamOnHeaders() + } + + const cacheControlHeader = resHeaders['cache-control'] + const heuristicallyCacheable = resHeaders['last-modified'] && HEURISTICALLY_CACHEABLE_STATUS_CODES.includes(statusCode) + if ( + !cacheControlHeader && + !resHeaders['expires'] && + !heuristicallyCacheable && + !this.#cacheByDefault + ) { + // Don't have anything to tell us this response is cachable and we're not + // caching by default + return downstreamOnHeaders() + } + + const cacheControlDirectives = cacheControlHeader ? parseCacheControlHeader(cacheControlHeader) : {} + if (!canCacheResponse(this.#cacheType, statusCode, resHeaders, cacheControlDirectives)) { + return downstreamOnHeaders() + } + + const now = Date.now() + const resAge = resHeaders.age ? getAge(resHeaders.age) : undefined + if (resAge && resAge >= MAX_RESPONSE_AGE) { + // Response considered stale + return downstreamOnHeaders() + } + + const resDate = typeof resHeaders.date === 'string' + ? parseHttpDate(resHeaders.date) + : undefined + + const staleAt = + determineStaleAt(this.#cacheType, now, resAge, resHeaders, resDate, cacheControlDirectives) ?? + this.#cacheByDefault + if (staleAt === undefined || (resAge && resAge > staleAt)) { + return downstreamOnHeaders() + } + + const baseTime = resDate ? resDate.getTime() : now + const absoluteStaleAt = staleAt + baseTime + if (now >= absoluteStaleAt) { + // Response is already stale + return downstreamOnHeaders() + } + + let varyDirectives + if (this.#cacheKey.headers && resHeaders.vary) { + varyDirectives = parseVaryHeader(resHeaders.vary, this.#cacheKey.headers) + if (!varyDirectives) { + // Parse error + return downstreamOnHeaders() + } + } + + const deleteAt = determineDeleteAt(baseTime, cacheControlDirectives, absoluteStaleAt) + const strippedHeaders = stripNecessaryHeaders(resHeaders, cacheControlDirectives) + + /** + * @type {import('../../types/cache-interceptor.d.ts').default.CacheValue} + */ + const value = { + statusCode, + statusMessage, + headers: strippedHeaders, + vary: varyDirectives, + cacheControlDirectives, + cachedAt: resAge ? now - resAge : now, + staleAt: absoluteStaleAt, + deleteAt + } + + // Not modified, re-use the cached value + // https://www.rfc-editor.org/rfc/rfc9111.html#name-handling-304-not-modified + if (statusCode === 304) { + const handle304 = (cachedValue) => { + if (!cachedValue) { + // Do not create a new cache entry, as a 304 won't have a body - so cannot be cached. + return downstreamOnHeaders() + } + + // Re-use the cached value: statuscode, statusmessage, headers and body + value.statusCode = cachedValue.statusCode + value.statusMessage = cachedValue.statusMessage + value.etag = cachedValue.etag + value.headers = { ...cachedValue.headers, ...strippedHeaders } + + downstreamOnHeaders() + + this.#writeStream = this.#store.createWriteStream(this.#cacheKey, value) + + if (!this.#writeStream || !cachedValue?.body) { + return + } + + if (typeof cachedValue.body.values === 'function') { + const bodyIterator = cachedValue.body.values() + + const streamCachedBody = () => { + for (const chunk of bodyIterator) { + const full = this.#writeStream.write(chunk) === false + this.#handler.onResponseData?.(controller, chunk) + // when stream is full stop writing until we get a 'drain' event + if (full) { + break + } + } + } + + this.#writeStream + .on('error', function () { + handler.#writeStream = undefined + handler.#store.delete(handler.#cacheKey) + }) + .on('drain', () => { + streamCachedBody() + }) + .on('close', function () { + if (handler.#writeStream === this) { + handler.#writeStream = undefined + } + }) + + streamCachedBody() + } else if (typeof cachedValue.body.on === 'function') { + // Readable stream body (e.g. from async/remote cache stores) + cachedValue.body + .on('data', (chunk) => { + this.#writeStream.write(chunk) + this.#handler.onResponseData?.(controller, chunk) + }) + .on('end', () => { + this.#writeStream.end() + }) + .on('error', () => { + this.#writeStream = undefined + this.#store.delete(this.#cacheKey) + }) + + this.#writeStream + .on('error', function () { + handler.#writeStream = undefined + handler.#store.delete(handler.#cacheKey) + }) + .on('close', function () { + if (handler.#writeStream === this) { + handler.#writeStream = undefined + } + }) + } + } + + /** + * @type {import('../../types/cache-interceptor.d.ts').default.CacheValue} + */ + const result = this.#store.get(this.#cacheKey) + if (result && typeof result.then === 'function') { + result.then(handle304) + } else { + handle304(result) + } + } else { + if (typeof resHeaders.etag === 'string' && isEtagUsable(resHeaders.etag)) { + value.etag = resHeaders.etag + } + + this.#writeStream = this.#store.createWriteStream(this.#cacheKey, value) + + if (!this.#writeStream) { + return downstreamOnHeaders() + } + + this.#writeStream + .on('drain', () => controller.resume()) + .on('error', function () { + // TODO (fix): Make error somehow observable? + handler.#writeStream = undefined + + // Delete the value in case the cache store is holding onto state from + // the call to createWriteStream + handler.#store.delete(handler.#cacheKey) + }) + .on('close', function () { + if (handler.#writeStream === this) { + handler.#writeStream = undefined + } + + // TODO (fix): Should we resume even if was paused downstream? + controller.resume() + }) + + downstreamOnHeaders() + } + } + + onResponseData (controller, chunk) { + if (this.#writeStream?.write(chunk) === false) { + controller.pause() + } + + this.#handler.onResponseData?.(controller, chunk) + } + + onResponseEnd (controller, trailers) { + this.#writeStream?.end() + this.#handler.onResponseEnd?.(controller, trailers) + } + + onResponseError (controller, err) { + this.#writeStream?.destroy(err) + this.#writeStream = undefined + this.#handler.onResponseError?.(controller, err) + } +} + +/** + * @see https://www.rfc-editor.org/rfc/rfc9111.html#name-storing-responses-to-authen + * + * @param {import('../../types/cache-interceptor.d.ts').default.CacheOptions['type']} cacheType + * @param {number} statusCode + * @param {import('../../types/header.d.ts').IncomingHttpHeaders} resHeaders + * @param {import('../../types/cache-interceptor.d.ts').default.CacheControlDirectives} cacheControlDirectives + */ +function canCacheResponse (cacheType, statusCode, resHeaders, cacheControlDirectives) { + // Status code must be final and understood. + if (statusCode < 200 || NOT_UNDERSTOOD_STATUS_CODES.includes(statusCode)) { + return false + } + // Responses with neither status codes that are heuristically cacheable, nor "explicit enough" caching + // directives, are not cacheable. "Explicit enough": see https://www.rfc-editor.org/rfc/rfc9111.html#section-3 + if (!HEURISTICALLY_CACHEABLE_STATUS_CODES.includes(statusCode) && !resHeaders['expires'] && + !cacheControlDirectives.public && + cacheControlDirectives['max-age'] === undefined && + // RFC 9111: a private response directive, if the cache is not shared + !(cacheControlDirectives.private && cacheType === 'private') && + !(cacheControlDirectives['s-maxage'] !== undefined && cacheType === 'shared') + ) { + return false + } + + if (cacheControlDirectives['no-store']) { + return false + } + + if (cacheType === 'shared' && cacheControlDirectives.private === true) { + return false + } + + // https://www.rfc-editor.org/rfc/rfc9111.html#section-4.1-5 + if (resHeaders.vary?.includes('*')) { + return false + } + + // https://www.rfc-editor.org/rfc/rfc9111.html#name-storing-responses-to-authen + if (resHeaders.authorization) { + if (!cacheControlDirectives.public || typeof resHeaders.authorization !== 'string') { + return false + } + + if ( + Array.isArray(cacheControlDirectives['no-cache']) && + cacheControlDirectives['no-cache'].includes('authorization') + ) { + return false + } + + if ( + Array.isArray(cacheControlDirectives['private']) && + cacheControlDirectives['private'].includes('authorization') + ) { + return false + } + } + + return true +} + +/** + * @param {string | string[]} ageHeader + * @returns {number | undefined} + */ +function getAge (ageHeader) { + const age = parseInt(Array.isArray(ageHeader) ? ageHeader[0] : ageHeader) + + return isNaN(age) ? undefined : age * 1000 +} + +/** + * @param {import('../../types/cache-interceptor.d.ts').default.CacheOptions['type']} cacheType + * @param {number} now + * @param {number | undefined} age + * @param {import('../../types/header.d.ts').IncomingHttpHeaders} resHeaders + * @param {Date | undefined} responseDate + * @param {import('../../types/cache-interceptor.d.ts').default.CacheControlDirectives} cacheControlDirectives + * + * @returns {number | undefined} time that the value is stale at in seconds or undefined if it shouldn't be cached + */ +function determineStaleAt (cacheType, now, age, resHeaders, responseDate, cacheControlDirectives) { + if (cacheType === 'shared') { + // Prioritize s-maxage since we're a shared cache + // s-maxage > max-age > Expire + // https://www.rfc-editor.org/rfc/rfc9111.html#section-5.2.2.10-3 + const sMaxAge = cacheControlDirectives['s-maxage'] + if (sMaxAge !== undefined) { + return sMaxAge > 0 ? sMaxAge * 1000 : undefined + } + } + + const maxAge = cacheControlDirectives['max-age'] + if (maxAge !== undefined) { + return maxAge > 0 ? maxAge * 1000 : undefined + } + + if (typeof resHeaders.expires === 'string') { + // https://www.rfc-editor.org/rfc/rfc9111.html#section-5.3 + const expiresDate = parseHttpDate(resHeaders.expires) + if (expiresDate) { + if (now >= expiresDate.getTime()) { + return undefined + } + + if (responseDate) { + if (responseDate >= expiresDate) { + return undefined + } + + if (age !== undefined && age > (expiresDate - responseDate)) { + return undefined + } + } + + return expiresDate.getTime() - now + } + } + + if (typeof resHeaders['last-modified'] === 'string') { + // https://www.rfc-editor.org/rfc/rfc9111.html#name-calculating-heuristic-fresh + const lastModified = new Date(resHeaders['last-modified']) + if (isValidDate(lastModified)) { + if (lastModified.getTime() >= now) { + return undefined + } + + const responseAge = now - lastModified.getTime() + + return responseAge * 0.1 + } + } + + if (cacheControlDirectives.immutable) { + // https://www.rfc-editor.org/rfc/rfc8246.html#section-2.2 + return 31536000 + } + + return undefined +} + +/** + * @param {number} now + * @param {import('../../types/cache-interceptor.d.ts').default.CacheControlDirectives} cacheControlDirectives + * @param {number} staleAt + */ +function determineDeleteAt (now, cacheControlDirectives, staleAt) { + let staleWhileRevalidate = -Infinity + let staleIfError = -Infinity + let immutable = -Infinity + + if (cacheControlDirectives['stale-while-revalidate']) { + staleWhileRevalidate = staleAt + (cacheControlDirectives['stale-while-revalidate'] * 1000) + } + + if (cacheControlDirectives['stale-if-error']) { + staleIfError = staleAt + (cacheControlDirectives['stale-if-error'] * 1000) + } + + if (staleWhileRevalidate === -Infinity && staleIfError === -Infinity) { + immutable = now + 31536000000 + } + + return Math.max(staleAt, staleWhileRevalidate, staleIfError, immutable) +} + +/** + * Strips headers required to be removed in cached responses + * @param {import('../../types/header.d.ts').IncomingHttpHeaders} resHeaders + * @param {import('../../types/cache-interceptor.d.ts').default.CacheControlDirectives} cacheControlDirectives + * @returns {Record<string, string | string []>} + */ +function stripNecessaryHeaders (resHeaders, cacheControlDirectives) { + const headersToRemove = [ + 'connection', + 'proxy-authenticate', + 'proxy-authentication-info', + 'proxy-authorization', + 'proxy-connection', + 'te', + 'transfer-encoding', + 'upgrade', + // We'll add age back when serving it + 'age' + ] + + if (resHeaders['connection']) { + if (Array.isArray(resHeaders['connection'])) { + // connection: a + // connection: b + headersToRemove.push(...resHeaders['connection'].map(header => header.trim())) + } else { + // connection: a, b + headersToRemove.push(...resHeaders['connection'].split(',').map(header => header.trim())) + } + } + + if (Array.isArray(cacheControlDirectives['no-cache'])) { + headersToRemove.push(...cacheControlDirectives['no-cache']) + } + + if (Array.isArray(cacheControlDirectives['private'])) { + headersToRemove.push(...cacheControlDirectives['private']) + } + + let strippedHeaders + for (const headerName of headersToRemove) { + if (resHeaders[headerName]) { + strippedHeaders ??= { ...resHeaders } + delete strippedHeaders[headerName] + } + } + + return strippedHeaders ?? resHeaders +} + +/** + * @param {Date} date + * @returns {boolean} + */ +function isValidDate (date) { + return date instanceof Date && Number.isFinite(date.valueOf()) +} + +module.exports = CacheHandler diff --git a/vanilla/node_modules/undici/lib/handler/cache-revalidation-handler.js b/vanilla/node_modules/undici/lib/handler/cache-revalidation-handler.js new file mode 100644 index 0000000..393d16d --- /dev/null +++ b/vanilla/node_modules/undici/lib/handler/cache-revalidation-handler.js @@ -0,0 +1,124 @@ +'use strict' + +const assert = require('node:assert') + +/** + * This takes care of revalidation requests we send to the origin. If we get + * a response indicating that what we have is cached (via a HTTP 304), we can + * continue using the cached value. Otherwise, we'll receive the new response + * here, which we then just pass on to the next handler (most likely a + * CacheHandler). Note that this assumes the proper headers were already + * included in the request to tell the origin that we want to revalidate the + * response (i.e. if-modified-since or if-none-match). + * + * @see https://www.rfc-editor.org/rfc/rfc9111.html#name-validation + * + * @implements {import('../../types/dispatcher.d.ts').default.DispatchHandler} + */ +class CacheRevalidationHandler { + #successful = false + + /** + * @type {((boolean, any) => void) | null} + */ + #callback + + /** + * @type {(import('../../types/dispatcher.d.ts').default.DispatchHandler)} + */ + #handler + + #context + + /** + * @type {boolean} + */ + #allowErrorStatusCodes + + /** + * @param {(boolean) => void} callback Function to call if the cached value is valid + * @param {import('../../types/dispatcher.d.ts').default.DispatchHandlers} handler + * @param {boolean} allowErrorStatusCodes + */ + constructor (callback, handler, allowErrorStatusCodes) { + if (typeof callback !== 'function') { + throw new TypeError('callback must be a function') + } + + this.#callback = callback + this.#handler = handler + this.#allowErrorStatusCodes = allowErrorStatusCodes + } + + onRequestStart (_, context) { + this.#successful = false + this.#context = context + } + + onRequestUpgrade (controller, statusCode, headers, socket) { + this.#handler.onRequestUpgrade?.(controller, statusCode, headers, socket) + } + + onResponseStart ( + controller, + statusCode, + headers, + statusMessage + ) { + assert(this.#callback != null) + + // https://www.rfc-editor.org/rfc/rfc9111.html#name-handling-a-validation-respo + // https://datatracker.ietf.org/doc/html/rfc5861#section-4 + this.#successful = statusCode === 304 || + (this.#allowErrorStatusCodes && statusCode >= 500 && statusCode <= 504) + this.#callback(this.#successful, this.#context) + this.#callback = null + + if (this.#successful) { + return true + } + + this.#handler.onRequestStart?.(controller, this.#context) + this.#handler.onResponseStart?.( + controller, + statusCode, + headers, + statusMessage + ) + } + + onResponseData (controller, chunk) { + if (this.#successful) { + return + } + + return this.#handler.onResponseData?.(controller, chunk) + } + + onResponseEnd (controller, trailers) { + if (this.#successful) { + return + } + + this.#handler.onResponseEnd?.(controller, trailers) + } + + onResponseError (controller, err) { + if (this.#successful) { + return + } + + if (this.#callback) { + this.#callback(false) + this.#callback = null + } + + if (typeof this.#handler.onResponseError === 'function') { + this.#handler.onResponseError(controller, err) + } else { + throw err + } + } +} + +module.exports = CacheRevalidationHandler diff --git a/vanilla/node_modules/undici/lib/handler/decorator-handler.js b/vanilla/node_modules/undici/lib/handler/decorator-handler.js new file mode 100644 index 0000000..50fbb0c --- /dev/null +++ b/vanilla/node_modules/undici/lib/handler/decorator-handler.js @@ -0,0 +1,67 @@ +'use strict' + +const assert = require('node:assert') +const WrapHandler = require('./wrap-handler') + +/** + * @deprecated + */ +module.exports = class DecoratorHandler { + #handler + #onCompleteCalled = false + #onErrorCalled = false + #onResponseStartCalled = false + + constructor (handler) { + if (typeof handler !== 'object' || handler === null) { + throw new TypeError('handler must be an object') + } + this.#handler = WrapHandler.wrap(handler) + } + + onRequestStart (...args) { + this.#handler.onRequestStart?.(...args) + } + + onRequestUpgrade (...args) { + assert(!this.#onCompleteCalled) + assert(!this.#onErrorCalled) + + return this.#handler.onRequestUpgrade?.(...args) + } + + onResponseStart (...args) { + assert(!this.#onCompleteCalled) + assert(!this.#onErrorCalled) + assert(!this.#onResponseStartCalled) + + this.#onResponseStartCalled = true + + return this.#handler.onResponseStart?.(...args) + } + + onResponseData (...args) { + assert(!this.#onCompleteCalled) + assert(!this.#onErrorCalled) + + return this.#handler.onResponseData?.(...args) + } + + onResponseEnd (...args) { + assert(!this.#onCompleteCalled) + assert(!this.#onErrorCalled) + + this.#onCompleteCalled = true + return this.#handler.onResponseEnd?.(...args) + } + + onResponseError (...args) { + this.#onErrorCalled = true + return this.#handler.onResponseError?.(...args) + } + + /** + * @deprecated + */ + onBodySent () {} +} diff --git a/vanilla/node_modules/undici/lib/handler/deduplication-handler.js b/vanilla/node_modules/undici/lib/handler/deduplication-handler.js new file mode 100644 index 0000000..33a2c4f --- /dev/null +++ b/vanilla/node_modules/undici/lib/handler/deduplication-handler.js @@ -0,0 +1,216 @@ +'use strict' + +/** + * @typedef {import('../../types/dispatcher.d.ts').default.DispatchHandler} DispatchHandler + */ + +/** + * Handler that buffers response data and notifies multiple waiting handlers. + * Used for request deduplication. + * + * @implements {DispatchHandler} + */ +class DeduplicationHandler { + /** + * @type {DispatchHandler} + */ + #primaryHandler + + /** + * @type {DispatchHandler[]} + */ + #waitingHandlers = [] + + /** + * @type {Buffer[]} + */ + #chunks = [] + + /** + * @type {number} + */ + #statusCode = 0 + + /** + * @type {Record<string, string | string[]>} + */ + #headers = {} + + /** + * @type {string} + */ + #statusMessage = '' + + /** + * @type {boolean} + */ + #aborted = false + + /** + * @type {import('../../types/dispatcher.d.ts').default.DispatchController | null} + */ + #controller = null + + /** + * @type {(() => void) | null} + */ + #onComplete = null + + /** + * @param {DispatchHandler} primaryHandler The primary handler + * @param {() => void} onComplete Callback when request completes + */ + constructor (primaryHandler, onComplete) { + this.#primaryHandler = primaryHandler + this.#onComplete = onComplete + } + + /** + * Add a waiting handler that will receive the buffered response + * @param {DispatchHandler} handler + */ + addWaitingHandler (handler) { + this.#waitingHandlers.push(handler) + } + + /** + * @param {() => void} abort + * @param {any} context + */ + onRequestStart (controller, context) { + this.#controller = controller + this.#primaryHandler.onRequestStart?.(controller, context) + } + + /** + * @param {import('../../types/dispatcher.d.ts').default.DispatchController} controller + * @param {number} statusCode + * @param {import('../../types/header.d.ts').IncomingHttpHeaders} headers + * @param {Socket} socket + */ + onRequestUpgrade (controller, statusCode, headers, socket) { + this.#primaryHandler.onRequestUpgrade?.(controller, statusCode, headers, socket) + } + + /** + * @param {import('../../types/dispatcher.d.ts').default.DispatchController} controller + * @param {number} statusCode + * @param {Record<string, string | string[]>} headers + * @param {string} statusMessage + */ + onResponseStart (controller, statusCode, headers, statusMessage) { + this.#statusCode = statusCode + this.#headers = headers + this.#statusMessage = statusMessage + this.#primaryHandler.onResponseStart?.(controller, statusCode, headers, statusMessage) + } + + /** + * @param {import('../../types/dispatcher.d.ts').default.DispatchController} controller + * @param {Buffer} chunk + */ + onResponseData (controller, chunk) { + // Buffer the chunk for waiting handlers + this.#chunks.push(Buffer.from(chunk)) + this.#primaryHandler.onResponseData?.(controller, chunk) + } + + /** + * @param {import('../../types/dispatcher.d.ts').default.DispatchController} controller + * @param {object} trailers + */ + onResponseEnd (controller, trailers) { + this.#primaryHandler.onResponseEnd?.(controller, trailers) + this.#notifyWaitingHandlers() + this.#onComplete?.() + } + + /** + * @param {import('../../types/dispatcher.d.ts').default.DispatchController} controller + * @param {Error} err + */ + onResponseError (controller, err) { + this.#aborted = true + this.#primaryHandler.onResponseError?.(controller, err) + this.#notifyWaitingHandlersError(err) + this.#onComplete?.() + } + + /** + * Notify all waiting handlers with the buffered response + */ + #notifyWaitingHandlers () { + const body = Buffer.concat(this.#chunks) + + for (const handler of this.#waitingHandlers) { + // Create a simple controller for each waiting handler + const waitingController = { + resume () {}, + pause () {}, + get paused () { return false }, + get aborted () { return false }, + get reason () { return null }, + abort () {} + } + + try { + handler.onRequestStart?.(waitingController, null) + + if (waitingController.aborted) { + continue + } + + handler.onResponseStart?.( + waitingController, + this.#statusCode, + this.#headers, + this.#statusMessage + ) + + if (waitingController.aborted) { + continue + } + + if (body.length > 0) { + handler.onResponseData?.(waitingController, body) + } + + handler.onResponseEnd?.(waitingController, {}) + } catch { + // Ignore errors from waiting handlers + } + } + + this.#waitingHandlers = [] + this.#chunks = [] + } + + /** + * Notify all waiting handlers of an error + * @param {Error} err + */ + #notifyWaitingHandlersError (err) { + for (const handler of this.#waitingHandlers) { + const waitingController = { + resume () {}, + pause () {}, + get paused () { return false }, + get aborted () { return true }, + get reason () { return err }, + abort () {} + } + + try { + handler.onRequestStart?.(waitingController, null) + handler.onResponseError?.(waitingController, err) + } catch { + // Ignore errors from waiting handlers + } + } + + this.#waitingHandlers = [] + this.#chunks = [] + } +} + +module.exports = DeduplicationHandler diff --git a/vanilla/node_modules/undici/lib/handler/redirect-handler.js b/vanilla/node_modules/undici/lib/handler/redirect-handler.js new file mode 100644 index 0000000..dd0f471 --- /dev/null +++ b/vanilla/node_modules/undici/lib/handler/redirect-handler.js @@ -0,0 +1,237 @@ +'use strict' + +const util = require('../core/util') +const { kBodyUsed } = require('../core/symbols') +const assert = require('node:assert') +const { InvalidArgumentError } = require('../core/errors') +const EE = require('node:events') + +const redirectableStatusCodes = [300, 301, 302, 303, 307, 308] + +const kBody = Symbol('body') + +const noop = () => {} + +class BodyAsyncIterable { + constructor (body) { + this[kBody] = body + this[kBodyUsed] = false + } + + async * [Symbol.asyncIterator] () { + assert(!this[kBodyUsed], 'disturbed') + this[kBodyUsed] = true + yield * this[kBody] + } +} + +class RedirectHandler { + static buildDispatch (dispatcher, maxRedirections) { + if (maxRedirections != null && (!Number.isInteger(maxRedirections) || maxRedirections < 0)) { + throw new InvalidArgumentError('maxRedirections must be a positive number') + } + + const dispatch = dispatcher.dispatch.bind(dispatcher) + return (opts, originalHandler) => dispatch(opts, new RedirectHandler(dispatch, maxRedirections, opts, originalHandler)) + } + + constructor (dispatch, maxRedirections, opts, handler) { + if (maxRedirections != null && (!Number.isInteger(maxRedirections) || maxRedirections < 0)) { + throw new InvalidArgumentError('maxRedirections must be a positive number') + } + + this.dispatch = dispatch + this.location = null + const { maxRedirections: _, ...cleanOpts } = opts + this.opts = cleanOpts // opts must be a copy, exclude maxRedirections + this.maxRedirections = maxRedirections + this.handler = handler + this.history = [] + + if (util.isStream(this.opts.body)) { + // TODO (fix): Provide some way for the user to cache the file to e.g. /tmp + // so that it can be dispatched again? + // TODO (fix): Do we need 100-expect support to provide a way to do this properly? + if (util.bodyLength(this.opts.body) === 0) { + this.opts.body + .on('data', function () { + assert(false) + }) + } + + if (typeof this.opts.body.readableDidRead !== 'boolean') { + this.opts.body[kBodyUsed] = false + EE.prototype.on.call(this.opts.body, 'data', function () { + this[kBodyUsed] = true + }) + } + } else if (this.opts.body && typeof this.opts.body.pipeTo === 'function') { + // TODO (fix): We can't access ReadableStream internal state + // to determine whether or not it has been disturbed. This is just + // a workaround. + this.opts.body = new BodyAsyncIterable(this.opts.body) + } else if ( + this.opts.body && + typeof this.opts.body !== 'string' && + !ArrayBuffer.isView(this.opts.body) && + util.isIterable(this.opts.body) && + !util.isFormDataLike(this.opts.body) + ) { + // TODO: Should we allow re-using iterable if !this.opts.idempotent + // or through some other flag? + this.opts.body = new BodyAsyncIterable(this.opts.body) + } + } + + onRequestStart (controller, context) { + this.handler.onRequestStart?.(controller, { ...context, history: this.history }) + } + + onRequestUpgrade (controller, statusCode, headers, socket) { + this.handler.onRequestUpgrade?.(controller, statusCode, headers, socket) + } + + onResponseStart (controller, statusCode, headers, statusMessage) { + if (this.opts.throwOnMaxRedirect && this.history.length >= this.maxRedirections) { + throw new Error('max redirects') + } + + // https://tools.ietf.org/html/rfc7231#section-6.4.2 + // https://fetch.spec.whatwg.org/#http-redirect-fetch + // In case of HTTP 301 or 302 with POST, change the method to GET + if ((statusCode === 301 || statusCode === 302) && this.opts.method === 'POST') { + this.opts.method = 'GET' + if (util.isStream(this.opts.body)) { + util.destroy(this.opts.body.on('error', noop)) + } + this.opts.body = null + } + + // https://tools.ietf.org/html/rfc7231#section-6.4.4 + // In case of HTTP 303, always replace method to be either HEAD or GET + if (statusCode === 303 && this.opts.method !== 'HEAD') { + this.opts.method = 'GET' + if (util.isStream(this.opts.body)) { + util.destroy(this.opts.body.on('error', noop)) + } + this.opts.body = null + } + + this.location = this.history.length >= this.maxRedirections || util.isDisturbed(this.opts.body) || redirectableStatusCodes.indexOf(statusCode) === -1 + ? null + : headers.location + + if (this.opts.origin) { + this.history.push(new URL(this.opts.path, this.opts.origin)) + } + + if (!this.location) { + this.handler.onResponseStart?.(controller, statusCode, headers, statusMessage) + return + } + + const { origin, pathname, search } = util.parseURL(new URL(this.location, this.opts.origin && new URL(this.opts.path, this.opts.origin))) + const path = search ? `${pathname}${search}` : pathname + + // Check for redirect loops by seeing if we've already visited this URL in our history + // This catches the case where Client/Pool try to handle cross-origin redirects but fail + // and keep redirecting to the same URL in an infinite loop + const redirectUrlString = `${origin}${path}` + for (const historyUrl of this.history) { + if (historyUrl.toString() === redirectUrlString) { + throw new InvalidArgumentError(`Redirect loop detected. Cannot redirect to ${origin}. This typically happens when using a Client or Pool with cross-origin redirects. Use an Agent for cross-origin redirects.`) + } + } + + // Remove headers referring to the original URL. + // By default it is Host only, unless it's a 303 (see below), which removes also all Content-* headers. + // https://tools.ietf.org/html/rfc7231#section-6.4 + this.opts.headers = cleanRequestHeaders(this.opts.headers, statusCode === 303, this.opts.origin !== origin) + this.opts.path = path + this.opts.origin = origin + this.opts.query = null + } + + onResponseData (controller, chunk) { + if (this.location) { + /* + https://tools.ietf.org/html/rfc7231#section-6.4 + + TLDR: undici always ignores 3xx response bodies. + + Redirection is used to serve the requested resource from another URL, so it assumes that + no body is generated (and thus can be ignored). Even though generating a body is not prohibited. + + For status 301, 302, 303, 307 and 308 (the latter from RFC 7238), the specs mention that the body usually + (which means it's optional and not mandated) contain just an hyperlink to the value of + the Location response header, so the body can be ignored safely. + + For status 300, which is "Multiple Choices", the spec mentions both generating a Location + response header AND a response body with the other possible location to follow. + Since the spec explicitly chooses not to specify a format for such body and leave it to + servers and browsers implementors, we ignore the body as there is no specified way to eventually parse it. + */ + } else { + this.handler.onResponseData?.(controller, chunk) + } + } + + onResponseEnd (controller, trailers) { + if (this.location) { + /* + https://tools.ietf.org/html/rfc7231#section-6.4 + + TLDR: undici always ignores 3xx response trailers as they are not expected in case of redirections + and neither are useful if present. + + See comment on onData method above for more detailed information. + */ + this.dispatch(this.opts, this) + } else { + this.handler.onResponseEnd(controller, trailers) + } + } + + onResponseError (controller, error) { + this.handler.onResponseError?.(controller, error) + } +} + +// https://tools.ietf.org/html/rfc7231#section-6.4.4 +function shouldRemoveHeader (header, removeContent, unknownOrigin) { + if (header.length === 4) { + return util.headerNameToString(header) === 'host' + } + if (removeContent && util.headerNameToString(header).startsWith('content-')) { + return true + } + if (unknownOrigin && (header.length === 13 || header.length === 6 || header.length === 19)) { + const name = util.headerNameToString(header) + return name === 'authorization' || name === 'cookie' || name === 'proxy-authorization' + } + return false +} + +// https://tools.ietf.org/html/rfc7231#section-6.4 +function cleanRequestHeaders (headers, removeContent, unknownOrigin) { + const ret = [] + if (Array.isArray(headers)) { + for (let i = 0; i < headers.length; i += 2) { + if (!shouldRemoveHeader(headers[i], removeContent, unknownOrigin)) { + ret.push(headers[i], headers[i + 1]) + } + } + } else if (headers && typeof headers === 'object') { + const entries = typeof headers[Symbol.iterator] === 'function' ? headers : Object.entries(headers) + for (const [key, value] of entries) { + if (!shouldRemoveHeader(key, removeContent, unknownOrigin)) { + ret.push(key, value) + } + } + } else { + assert(headers == null, 'headers must be an object or an array') + } + return ret +} + +module.exports = RedirectHandler diff --git a/vanilla/node_modules/undici/lib/handler/retry-handler.js b/vanilla/node_modules/undici/lib/handler/retry-handler.js new file mode 100644 index 0000000..1cbc789 --- /dev/null +++ b/vanilla/node_modules/undici/lib/handler/retry-handler.js @@ -0,0 +1,394 @@ +'use strict' +const assert = require('node:assert') + +const { kRetryHandlerDefaultRetry } = require('../core/symbols') +const { RequestRetryError } = require('../core/errors') +const WrapHandler = require('./wrap-handler') +const { + isDisturbed, + parseRangeHeader, + wrapRequestBody +} = require('../core/util') + +function calculateRetryAfterHeader (retryAfter) { + const retryTime = new Date(retryAfter).getTime() + return isNaN(retryTime) ? 0 : retryTime - Date.now() +} + +class RetryHandler { + constructor (opts, { dispatch, handler }) { + const { retryOptions, ...dispatchOpts } = opts + const { + // Retry scoped + retry: retryFn, + maxRetries, + maxTimeout, + minTimeout, + timeoutFactor, + // Response scoped + methods, + errorCodes, + retryAfter, + statusCodes, + throwOnError + } = retryOptions ?? {} + + this.error = null + this.dispatch = dispatch + this.handler = WrapHandler.wrap(handler) + this.opts = { ...dispatchOpts, body: wrapRequestBody(opts.body) } + this.retryOpts = { + throwOnError: throwOnError ?? true, + retry: retryFn ?? RetryHandler[kRetryHandlerDefaultRetry], + retryAfter: retryAfter ?? true, + maxTimeout: maxTimeout ?? 30 * 1000, // 30s, + minTimeout: minTimeout ?? 500, // .5s + timeoutFactor: timeoutFactor ?? 2, + maxRetries: maxRetries ?? 5, + // What errors we should retry + methods: methods ?? ['GET', 'HEAD', 'OPTIONS', 'PUT', 'DELETE', 'TRACE'], + // Indicates which errors to retry + statusCodes: statusCodes ?? [500, 502, 503, 504, 429], + // List of errors to retry + errorCodes: errorCodes ?? [ + 'ECONNRESET', + 'ECONNREFUSED', + 'ENOTFOUND', + 'ENETDOWN', + 'ENETUNREACH', + 'EHOSTDOWN', + 'EHOSTUNREACH', + 'EPIPE', + 'UND_ERR_SOCKET' + ] + } + + this.retryCount = 0 + this.retryCountCheckpoint = 0 + this.headersSent = false + this.start = 0 + this.end = null + this.etag = null + } + + onResponseStartWithRetry (controller, statusCode, headers, statusMessage, err) { + if (this.retryOpts.throwOnError) { + // Preserve old behavior for status codes that are not eligible for retry + if (this.retryOpts.statusCodes.includes(statusCode) === false) { + this.headersSent = true + this.handler.onResponseStart?.(controller, statusCode, headers, statusMessage) + } else { + this.error = err + } + + return + } + + if (isDisturbed(this.opts.body)) { + this.headersSent = true + this.handler.onResponseStart?.(controller, statusCode, headers, statusMessage) + return + } + + function shouldRetry (passedErr) { + if (passedErr) { + this.headersSent = true + this.handler.onResponseStart?.(controller, statusCode, headers, statusMessage) + controller.resume() + return + } + + this.error = err + controller.resume() + } + + controller.pause() + this.retryOpts.retry( + err, + { + state: { counter: this.retryCount }, + opts: { retryOptions: this.retryOpts, ...this.opts } + }, + shouldRetry.bind(this) + ) + } + + onRequestStart (controller, context) { + if (!this.headersSent) { + this.handler.onRequestStart?.(controller, context) + } + } + + onRequestUpgrade (controller, statusCode, headers, socket) { + this.handler.onRequestUpgrade?.(controller, statusCode, headers, socket) + } + + static [kRetryHandlerDefaultRetry] (err, { state, opts }, cb) { + const { statusCode, code, headers } = err + const { method, retryOptions } = opts + const { + maxRetries, + minTimeout, + maxTimeout, + timeoutFactor, + statusCodes, + errorCodes, + methods + } = retryOptions + const { counter } = state + + // Any code that is not a Undici's originated and allowed to retry + if (code && code !== 'UND_ERR_REQ_RETRY' && !errorCodes.includes(code)) { + cb(err) + return + } + + // If a set of method are provided and the current method is not in the list + if (Array.isArray(methods) && !methods.includes(method)) { + cb(err) + return + } + + // If a set of status code are provided and the current status code is not in the list + if ( + statusCode != null && + Array.isArray(statusCodes) && + !statusCodes.includes(statusCode) + ) { + cb(err) + return + } + + // If we reached the max number of retries + if (counter > maxRetries) { + cb(err) + return + } + + let retryAfterHeader = headers?.['retry-after'] + if (retryAfterHeader) { + retryAfterHeader = Number(retryAfterHeader) + retryAfterHeader = Number.isNaN(retryAfterHeader) + ? calculateRetryAfterHeader(headers['retry-after']) + : retryAfterHeader * 1e3 // Retry-After is in seconds + } + + const retryTimeout = + retryAfterHeader > 0 + ? Math.min(retryAfterHeader, maxTimeout) + : Math.min(minTimeout * timeoutFactor ** (counter - 1), maxTimeout) + + setTimeout(() => cb(null), retryTimeout) + } + + onResponseStart (controller, statusCode, headers, statusMessage) { + this.error = null + this.retryCount += 1 + + if (statusCode >= 300) { + const err = new RequestRetryError('Request failed', statusCode, { + headers, + data: { + count: this.retryCount + } + }) + + this.onResponseStartWithRetry(controller, statusCode, headers, statusMessage, err) + return + } + + // Checkpoint for resume from where we left it + if (this.headersSent) { + // Only Partial Content 206 supposed to provide Content-Range, + // any other status code that partially consumed the payload + // should not be retried because it would result in downstream + // wrongly concatenate multiple responses. + if (statusCode !== 206 && (this.start > 0 || statusCode !== 200)) { + throw new RequestRetryError('server does not support the range header and the payload was partially consumed', statusCode, { + headers, + data: { count: this.retryCount } + }) + } + + const contentRange = parseRangeHeader(headers['content-range']) + // If no content range + if (!contentRange) { + // We always throw here as we want to indicate that we entred unexpected path + throw new RequestRetryError('Content-Range mismatch', statusCode, { + headers, + data: { count: this.retryCount } + }) + } + + // Let's start with a weak etag check + if (this.etag != null && this.etag !== headers.etag) { + // We always throw here as we want to indicate that we entred unexpected path + throw new RequestRetryError('ETag mismatch', statusCode, { + headers, + data: { count: this.retryCount } + }) + } + + const { start, size, end = size ? size - 1 : null } = contentRange + + assert(this.start === start, 'content-range mismatch') + assert(this.end == null || this.end === end, 'content-range mismatch') + + return + } + + if (this.end == null) { + if (statusCode === 206) { + // First time we receive 206 + const range = parseRangeHeader(headers['content-range']) + + if (range == null) { + this.headersSent = true + this.handler.onResponseStart?.( + controller, + statusCode, + headers, + statusMessage + ) + return + } + + const { start, size, end = size ? size - 1 : null } = range + assert( + start != null && Number.isFinite(start), + 'content-range mismatch' + ) + assert(end != null && Number.isFinite(end), 'invalid content-length') + + this.start = start + this.end = end + } + + // We make our best to checkpoint the body for further range headers + if (this.end == null) { + const contentLength = headers['content-length'] + this.end = contentLength != null ? Number(contentLength) - 1 : null + } + + assert(Number.isFinite(this.start)) + assert( + this.end == null || Number.isFinite(this.end), + 'invalid content-length' + ) + + this.resume = true + this.etag = headers.etag != null ? headers.etag : null + + // Weak etags are not useful for comparison nor cache + // for instance not safe to assume if the response is byte-per-byte + // equal + if ( + this.etag != null && + this.etag[0] === 'W' && + this.etag[1] === '/' + ) { + this.etag = null + } + + this.headersSent = true + this.handler.onResponseStart?.( + controller, + statusCode, + headers, + statusMessage + ) + } else { + throw new RequestRetryError('Request failed', statusCode, { + headers, + data: { count: this.retryCount } + }) + } + } + + onResponseData (controller, chunk) { + if (this.error) { + return + } + + this.start += chunk.length + + this.handler.onResponseData?.(controller, chunk) + } + + onResponseEnd (controller, trailers) { + if (this.error && this.retryOpts.throwOnError) { + throw this.error + } + + if (!this.error) { + this.retryCount = 0 + return this.handler.onResponseEnd?.(controller, trailers) + } + + this.retry(controller) + } + + retry (controller) { + if (this.start !== 0) { + const headers = { range: `bytes=${this.start}-${this.end ?? ''}` } + + // Weak etag check - weak etags will make comparison algorithms never match + if (this.etag != null) { + headers['if-match'] = this.etag + } + + this.opts = { + ...this.opts, + headers: { + ...this.opts.headers, + ...headers + } + } + } + + try { + this.retryCountCheckpoint = this.retryCount + this.dispatch(this.opts, this) + } catch (err) { + this.handler.onResponseError?.(controller, err) + } + } + + onResponseError (controller, err) { + if (controller?.aborted || isDisturbed(this.opts.body)) { + this.handler.onResponseError?.(controller, err) + return + } + + function shouldRetry (returnedErr) { + if (!returnedErr) { + this.retry(controller) + return + } + + this.handler?.onResponseError?.(controller, returnedErr) + } + + // We reconcile in case of a mix between network errors + // and server error response + if (this.retryCount - this.retryCountCheckpoint > 0) { + // We count the difference between the last checkpoint and the current retry count + this.retryCount = + this.retryCountCheckpoint + + (this.retryCount - this.retryCountCheckpoint) + } else { + this.retryCount += 1 + } + + this.retryOpts.retry( + err, + { + state: { counter: this.retryCount }, + opts: { retryOptions: this.retryOpts, ...this.opts } + }, + shouldRetry.bind(this) + ) + } +} + +module.exports = RetryHandler diff --git a/vanilla/node_modules/undici/lib/handler/unwrap-handler.js b/vanilla/node_modules/undici/lib/handler/unwrap-handler.js new file mode 100644 index 0000000..865593a --- /dev/null +++ b/vanilla/node_modules/undici/lib/handler/unwrap-handler.js @@ -0,0 +1,96 @@ +'use strict' + +const { parseHeaders } = require('../core/util') +const { InvalidArgumentError } = require('../core/errors') + +const kResume = Symbol('resume') + +class UnwrapController { + #paused = false + #reason = null + #aborted = false + #abort + + [kResume] = null + + constructor (abort) { + this.#abort = abort + } + + pause () { + this.#paused = true + } + + resume () { + if (this.#paused) { + this.#paused = false + this[kResume]?.() + } + } + + abort (reason) { + if (!this.#aborted) { + this.#aborted = true + this.#reason = reason + this.#abort(reason) + } + } + + get aborted () { + return this.#aborted + } + + get reason () { + return this.#reason + } + + get paused () { + return this.#paused + } +} + +module.exports = class UnwrapHandler { + #handler + #controller + + constructor (handler) { + this.#handler = handler + } + + static unwrap (handler) { + // TODO (fix): More checks... + return !handler.onRequestStart ? handler : new UnwrapHandler(handler) + } + + onConnect (abort, context) { + this.#controller = new UnwrapController(abort) + this.#handler.onRequestStart?.(this.#controller, context) + } + + onUpgrade (statusCode, rawHeaders, socket) { + this.#handler.onRequestUpgrade?.(this.#controller, statusCode, parseHeaders(rawHeaders), socket) + } + + onHeaders (statusCode, rawHeaders, resume, statusMessage) { + this.#controller[kResume] = resume + this.#handler.onResponseStart?.(this.#controller, statusCode, parseHeaders(rawHeaders), statusMessage) + return !this.#controller.paused + } + + onData (data) { + this.#handler.onResponseData?.(this.#controller, data) + return !this.#controller.paused + } + + onComplete (rawTrailers) { + this.#handler.onResponseEnd?.(this.#controller, parseHeaders(rawTrailers)) + } + + onError (err) { + if (!this.#handler.onResponseError) { + throw new InvalidArgumentError('invalid onError method') + } + + this.#handler.onResponseError?.(this.#controller, err) + } +} diff --git a/vanilla/node_modules/undici/lib/handler/wrap-handler.js b/vanilla/node_modules/undici/lib/handler/wrap-handler.js new file mode 100644 index 0000000..47caa5f --- /dev/null +++ b/vanilla/node_modules/undici/lib/handler/wrap-handler.js @@ -0,0 +1,95 @@ +'use strict' + +const { InvalidArgumentError } = require('../core/errors') + +module.exports = class WrapHandler { + #handler + + constructor (handler) { + this.#handler = handler + } + + static wrap (handler) { + // TODO (fix): More checks... + return handler.onRequestStart ? handler : new WrapHandler(handler) + } + + // Unwrap Interface + + onConnect (abort, context) { + return this.#handler.onConnect?.(abort, context) + } + + onHeaders (statusCode, rawHeaders, resume, statusMessage) { + return this.#handler.onHeaders?.(statusCode, rawHeaders, resume, statusMessage) + } + + onUpgrade (statusCode, rawHeaders, socket) { + return this.#handler.onUpgrade?.(statusCode, rawHeaders, socket) + } + + onData (data) { + return this.#handler.onData?.(data) + } + + onComplete (trailers) { + return this.#handler.onComplete?.(trailers) + } + + onError (err) { + if (!this.#handler.onError) { + throw err + } + + return this.#handler.onError?.(err) + } + + // Wrap Interface + + onRequestStart (controller, context) { + this.#handler.onConnect?.((reason) => controller.abort(reason), context) + } + + onRequestUpgrade (controller, statusCode, headers, socket) { + const rawHeaders = [] + for (const [key, val] of Object.entries(headers)) { + rawHeaders.push(Buffer.from(key), Array.isArray(val) ? val.map(v => Buffer.from(v)) : Buffer.from(val)) + } + + this.#handler.onUpgrade?.(statusCode, rawHeaders, socket) + } + + onResponseStart (controller, statusCode, headers, statusMessage) { + const rawHeaders = [] + for (const [key, val] of Object.entries(headers)) { + rawHeaders.push(Buffer.from(key), Array.isArray(val) ? val.map(v => Buffer.from(v)) : Buffer.from(val)) + } + + if (this.#handler.onHeaders?.(statusCode, rawHeaders, () => controller.resume(), statusMessage) === false) { + controller.pause() + } + } + + onResponseData (controller, data) { + if (this.#handler.onData?.(data) === false) { + controller.pause() + } + } + + onResponseEnd (controller, trailers) { + const rawTrailers = [] + for (const [key, val] of Object.entries(trailers)) { + rawTrailers.push(Buffer.from(key), Array.isArray(val) ? val.map(v => Buffer.from(v)) : Buffer.from(val)) + } + + this.#handler.onComplete?.(rawTrailers) + } + + onResponseError (controller, err) { + if (!this.#handler.onError) { + throw new InvalidArgumentError('invalid onError method') + } + + this.#handler.onError?.(err) + } +} |