feat: Complete zCode CLI X with Telegram bot integration

- Add full Telegram bot functionality with Z.AI API integration
- Implement 4 tools: Bash, FileEdit, WebSearch, Git
- Add 3 agents: Code Reviewer, Architect, DevOps Engineer
- Add 6 skills for common coding tasks
- Add systemd service file for 24/7 operation
- Add nginx configuration for HTTPS webhook
- Add comprehensive documentation
- Implement WebSocket server for real-time updates
- Add logging system with Winston
- Add environment validation

🤖 zCode CLI X - Agentic coder with Z.AI + Telegram integration
This commit is contained in:
admin
2026-05-05 09:01:26 +00:00
Unverified
parent 4a7035dd92
commit 875c7f9b91
24688 changed files with 3224957 additions and 221 deletions

View File

@@ -0,0 +1,578 @@
'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, this.#cacheKey.headers)) {
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
* @param {import('../../types/header.d.ts').IncomingHttpHeaders} [reqHeaders]
*/
function canCacheResponse (cacheType, statusCode, resHeaders, cacheControlDirectives, reqHeaders) {
// 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 (reqHeaders?.authorization) {
if (
!cacheControlDirectives.public &&
!cacheControlDirectives['s-maxage'] &&
!cacheControlDirectives['must-revalidate']
) {
return false
}
if (typeof reqHeaders.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 (cacheControlDirectives.immutable && staleWhileRevalidate === -Infinity && staleIfError === -Infinity) {
immutable = now + 31536000000
}
// When no stale directives or immutable flag, add a revalidation buffer
// equal to the freshness lifetime so the entry survives past staleAt long
// enough to be revalidated instead of silently disappearing.
if (staleWhileRevalidate === -Infinity && staleIfError === -Infinity && immutable === -Infinity) {
const freshnessLifetime = staleAt - now
return staleAt + freshnessLifetime
}
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

View File

@@ -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

View File

@@ -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 () {}
}

View File

@@ -0,0 +1,460 @@
'use strict'
const { RequestAbortedError } = require('../core/errors')
/**
* @typedef {import('../../types/dispatcher.d.ts').default.DispatchHandler} DispatchHandler
*/
const DEFAULT_MAX_BUFFER_SIZE = 5 * 1024 * 1024
/**
* @typedef {Object} WaitingHandler
* @property {DispatchHandler} handler
* @property {import('../../types/dispatcher.d.ts').default.DispatchController} controller
* @property {Buffer[]} bufferedChunks
* @property {number} bufferedBytes
* @property {object | null} pendingTrailers
* @property {boolean} done
*/
/**
* Handler that forwards response events to multiple waiting handlers.
* Used for request deduplication.
*
* @implements {DispatchHandler}
*/
class DeduplicationHandler {
/**
* @type {DispatchHandler}
*/
#primaryHandler
/**
* @type {WaitingHandler[]}
*/
#waitingHandlers = []
/**
* @type {number}
*/
#maxBufferSize = DEFAULT_MAX_BUFFER_SIZE
/**
* @type {number}
*/
#statusCode = 0
/**
* @type {Record<string, string | string[]>}
*/
#headers = {}
/**
* @type {string}
*/
#statusMessage = ''
/**
* @type {boolean}
*/
#aborted = false
/**
* @type {boolean}
*/
#responseStarted = false
/**
* @type {boolean}
*/
#responseDataStarted = false
/**
* @type {boolean}
*/
#completed = 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
* @param {number} [maxBufferSize] Maximum paused buffer size per waiting handler
*/
constructor (primaryHandler, onComplete, maxBufferSize = DEFAULT_MAX_BUFFER_SIZE) {
this.#primaryHandler = primaryHandler
this.#onComplete = onComplete
this.#maxBufferSize = maxBufferSize
}
/**
* Add a waiting handler that will receive response events.
* Returns false if deduplication can no longer safely attach this handler.
*
* @param {DispatchHandler} handler
* @returns {boolean}
*/
addWaitingHandler (handler) {
if (this.#completed || this.#responseDataStarted) {
return false
}
const waitingHandler = this.#createWaitingHandler(handler)
const waitingController = waitingHandler.controller
try {
handler.onRequestStart?.(waitingController, null)
if (waitingController.aborted) {
waitingHandler.done = true
return true
}
if (this.#responseStarted) {
handler.onResponseStart?.(
waitingController,
this.#statusCode,
this.#headers,
this.#statusMessage
)
}
} catch {
// Ignore errors from waiting handlers
waitingHandler.done = true
return true
}
if (!waitingController.aborted) {
this.#waitingHandlers.push(waitingHandler)
}
return true
}
/**
* @param {import('../../types/dispatcher.d.ts').default.DispatchController} controller
* @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.#responseStarted = true
this.#statusCode = statusCode
this.#headers = headers
this.#statusMessage = statusMessage
this.#primaryHandler.onResponseStart?.(controller, statusCode, headers, statusMessage)
for (const waitingHandler of this.#waitingHandlers) {
const { handler, controller: waitingController } = waitingHandler
if (waitingHandler.done || waitingController.aborted) {
waitingHandler.done = true
continue
}
try {
handler.onResponseStart?.(
waitingController,
statusCode,
headers,
statusMessage
)
} catch {
// Ignore errors from waiting handlers
}
if (waitingController.aborted) {
waitingHandler.done = true
}
}
this.#pruneDoneWaitingHandlers()
}
/**
* @param {import('../../types/dispatcher.d.ts').default.DispatchController} controller
* @param {Buffer} chunk
*/
onResponseData (controller, chunk) {
if (this.#aborted || this.#completed) {
return
}
this.#responseDataStarted = true
this.#primaryHandler.onResponseData?.(controller, chunk)
for (const waitingHandler of this.#waitingHandlers) {
const { handler, controller: waitingController } = waitingHandler
if (waitingHandler.done || waitingController.aborted) {
waitingHandler.done = true
continue
}
if (waitingController.paused) {
this.#bufferWaitingChunk(waitingHandler, chunk)
continue
}
try {
handler.onResponseData?.(waitingController, chunk)
} catch {
// Ignore errors from waiting handlers
}
if (waitingController.aborted) {
waitingHandler.done = true
waitingHandler.bufferedChunks = []
waitingHandler.bufferedBytes = 0
}
}
this.#pruneDoneWaitingHandlers()
}
/**
* @param {import('../../types/dispatcher.d.ts').default.DispatchController} controller
* @param {object} trailers
*/
onResponseEnd (controller, trailers) {
if (this.#aborted || this.#completed) {
return
}
this.#completed = true
this.#primaryHandler.onResponseEnd?.(controller, trailers)
for (const waitingHandler of this.#waitingHandlers) {
if (waitingHandler.done || waitingHandler.controller.aborted) {
waitingHandler.done = true
continue
}
this.#flushWaitingHandler(waitingHandler)
if (waitingHandler.done || waitingHandler.controller.aborted) {
waitingHandler.done = true
continue
}
if (waitingHandler.controller.paused && waitingHandler.bufferedChunks.length > 0) {
waitingHandler.pendingTrailers = trailers
continue
}
try {
waitingHandler.handler.onResponseEnd?.(waitingHandler.controller, trailers)
} catch {
// Ignore errors from waiting handlers
}
waitingHandler.done = true
}
this.#pruneDoneWaitingHandlers()
this.#onComplete?.()
}
/**
* @param {import('../../types/dispatcher.d.ts').default.DispatchController} controller
* @param {Error} err
*/
onResponseError (controller, err) {
if (this.#completed) {
return
}
this.#aborted = true
this.#completed = true
this.#primaryHandler.onResponseError?.(controller, err)
for (const waitingHandler of this.#waitingHandlers) {
this.#errorWaitingHandler(waitingHandler, err)
}
this.#waitingHandlers = []
this.#onComplete?.()
}
/**
* @param {DispatchHandler} handler
* @returns {WaitingHandler}
*/
#createWaitingHandler (handler) {
/** @type {WaitingHandler} */
const waitingHandler = {
handler,
controller: null,
bufferedChunks: [],
bufferedBytes: 0,
pendingTrailers: null,
done: false
}
const state = {
aborted: false,
paused: false,
reason: null
}
waitingHandler.controller = {
resume: () => {
if (state.aborted) {
return
}
state.paused = false
this.#flushWaitingHandler(waitingHandler)
if (
this.#completed &&
waitingHandler.pendingTrailers &&
waitingHandler.bufferedChunks.length === 0 &&
!state.paused &&
!state.aborted
) {
try {
waitingHandler.handler.onResponseEnd?.(waitingHandler.controller, waitingHandler.pendingTrailers)
} catch {
// Ignore errors from waiting handlers
}
waitingHandler.pendingTrailers = null
waitingHandler.done = true
}
this.#pruneDoneWaitingHandlers()
},
pause: () => {
if (!state.aborted) {
state.paused = true
}
},
get paused () { return state.paused },
get aborted () { return state.aborted },
get reason () { return state.reason },
abort: (reason) => {
state.aborted = true
state.reason = reason ?? null
waitingHandler.done = true
waitingHandler.pendingTrailers = null
waitingHandler.bufferedChunks = []
waitingHandler.bufferedBytes = 0
}
}
return waitingHandler
}
/**
* @param {WaitingHandler} waitingHandler
* @param {Buffer} chunk
*/
#bufferWaitingChunk (waitingHandler, chunk) {
if (waitingHandler.done || waitingHandler.controller.aborted) {
waitingHandler.done = true
waitingHandler.bufferedChunks = []
waitingHandler.bufferedBytes = 0
return
}
const bufferedChunk = Buffer.from(chunk)
waitingHandler.bufferedChunks.push(bufferedChunk)
waitingHandler.bufferedBytes += bufferedChunk.length
if (waitingHandler.bufferedBytes > this.#maxBufferSize) {
const err = new RequestAbortedError(`Deduplicated waiting handler exceeded maxBufferSize (${this.#maxBufferSize} bytes) while paused`)
this.#errorWaitingHandler(waitingHandler, err)
}
}
/**
* @param {WaitingHandler} waitingHandler
*/
#flushWaitingHandler (waitingHandler) {
const { handler, controller } = waitingHandler
while (
!waitingHandler.done &&
!controller.aborted &&
!controller.paused &&
waitingHandler.bufferedChunks.length > 0
) {
const bufferedChunk = waitingHandler.bufferedChunks.shift()
waitingHandler.bufferedBytes -= bufferedChunk.length
try {
handler.onResponseData?.(controller, bufferedChunk)
} catch {
// Ignore errors from waiting handlers
}
if (controller.aborted) {
waitingHandler.done = true
waitingHandler.pendingTrailers = null
waitingHandler.bufferedChunks = []
waitingHandler.bufferedBytes = 0
break
}
}
}
/**
* @param {WaitingHandler} waitingHandler
* @param {Error} err
*/
#errorWaitingHandler (waitingHandler, err) {
if (waitingHandler.done) {
return
}
waitingHandler.done = true
waitingHandler.pendingTrailers = null
waitingHandler.bufferedChunks = []
waitingHandler.bufferedBytes = 0
try {
waitingHandler.controller.abort(err)
waitingHandler.handler.onResponseError?.(waitingHandler.controller, err)
} catch {
// Ignore errors from waiting handlers
}
}
#pruneDoneWaitingHandlers () {
this.#waitingHandlers = this.#waitingHandlers.filter(waitingHandler => waitingHandler.done === false)
}
}
module.exports = DeduplicationHandler

View File

@@ -0,0 +1,238 @@
'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 = util.hasSafeIterator(headers) ? 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

View File

@@ -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

View File

@@ -0,0 +1,100 @@
'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)
}
onResponseStarted () {
return this.#handler.onResponseStarted?.()
}
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)
}
}

View File

@@ -0,0 +1,105 @@
'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)
}
onResponseStarted () {
return this.#handler.onResponseStarted?.()
}
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, 'latin1'), toRawHeaderValue(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, 'latin1'), toRawHeaderValue(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, 'latin1'), toRawHeaderValue(val))
}
this.#handler.onComplete?.(rawTrailers)
}
onResponseError (controller, err) {
if (!this.#handler.onError) {
throw new InvalidArgumentError('invalid onError method')
}
this.#handler.onError?.(err)
}
}
function toRawHeaderValue (value) {
return Array.isArray(value)
? value.map((item) => Buffer.from(item, 'latin1'))
: Buffer.from(value, 'latin1')
}