diff --git a/core/packages/teeny-request/README.md b/core/packages/teeny-request/README.md index fd646e9edc23..d5c76a0e196a 100644 --- a/core/packages/teeny-request/README.md +++ b/core/packages/teeny-request/README.md @@ -4,7 +4,7 @@ # teeny-request -Like `request`, but much smaller - and with less options. Uses `node-fetch` under the hood. +Like `request`, but much smaller - and with less options. Uses `undici` under the hood. Pop it in where you would use `request`. Improves load and parse time of modules. ```js diff --git a/core/packages/teeny-request/package.json b/core/packages/teeny-request/package.json index ed5ffe3a5bf2..6a9b67f0d298 100644 --- a/core/packages/teeny-request/package.json +++ b/core/packages/teeny-request/package.json @@ -30,7 +30,7 @@ }, "keywords": [ "request", - "node-fetch", + "undici", "fetch" ], "author": "fhinkel", @@ -40,15 +40,13 @@ }, "homepage": "https://github.com/googleapis/google-cloud-node/tree/main/core/packages/teeny-request", "dependencies": { - "http-proxy-agent": "^7.0.0", - "https-proxy-agent": "^7.0.1", - "node-fetch": "^3.3.2", - "stream-events": "^1.0.5" + "stream-events": "^1.0.5", + "undici": "^8.10.1" }, "devDependencies": { "@babel/plugin-proposal-private-methods": "^7.18.6", "@types/mocha": "^10.0.10", - "@types/node-fetch": "^2.6.12", + "@types/node": "^24.0.0", "@types/sinon": "^17.0.3", "c8": "^10.1.3", "codecov": "^3.8.3", @@ -57,7 +55,6 @@ "jsdoc-fresh": "^6.0.0", "jsdoc-region-tag": "^5.0.0", "mocha": "^11.1.0", - "nock": "^14.0.1", "sinon": "^19.0.2", "typescript": "^5.7.3" }, diff --git a/core/packages/teeny-request/src/agents.ts b/core/packages/teeny-request/src/agents.ts index bf27f8437ed2..401bfe3337be 100644 --- a/core/packages/teeny-request/src/agents.ts +++ b/core/packages/teeny-request/src/agents.ts @@ -15,13 +15,31 @@ * limitations under the License. */ -import {Agent as HTTPAgent} from 'http'; -import {Agent as HTTPSAgent} from 'https'; +import { + Agent, + Dispatcher, + ProxyAgent, + getGlobalDispatcher, + interceptors, +} from 'undici'; import {Options} from './'; -export const pool = new Map(); +export const pool = new Map(); -export type HttpAnyAgent = HTTPAgent | HTTPSAgent; +// undici only follows redirects through an interceptor; node-fetch +// followed up to 20, so preserve that +const redirect = interceptors.redirect({maxRedirections: 20}); + +const composed = new WeakMap(); + +function withRedirects(dispatcher: Dispatcher): Dispatcher { + let dispatcherWithRedirects = composed.get(dispatcher); + if (!dispatcherWithRedirects) { + dispatcherWithRedirects = dispatcher.compose(redirect); + composed.set(dispatcher, dispatcherWithRedirects); + } + return dispatcherWithRedirects; +} /** * Determines if a proxy should be considered based on the environment. @@ -55,18 +73,16 @@ function shouldUseProxyForURI(uri: string): boolean { } /** - * Returns a custom request Agent if one is found, otherwise returns undefined - * which will result in the global http(s) Agent being used. + * Returns a dispatcher for the given request. Proxied requests and requests + * with a socket limit get a cached dedicated dispatcher; everything else + * uses undici's global dispatcher, which pools and keeps connections alive + * by default. * @private * @param {string} uri The request uri * @param {Options} reqOpts The request options - * @returns {HttpAnyAgent|undefined} + * @returns {Dispatcher} */ -export function getAgent( - uri: string, - reqOpts: Options, -): HttpAnyAgent | undefined { - const isHttp = uri.startsWith('http://'); +export function getDispatcher(uri: string, reqOpts: Options): Dispatcher { const proxy = reqOpts.proxy || process.env.HTTP_PROXY || @@ -74,31 +90,36 @@ export function getAgent( process.env.HTTPS_PROXY || process.env.https_proxy; - const poolOptions = Object.assign({}, reqOpts.pool); - const manuallyProvidedProxy = !!reqOpts.proxy; const shouldUseProxy = manuallyProvidedProxy || shouldUseProxyForURI(uri); - if (proxy && shouldUseProxy) { - // tslint:disable-next-line variable-name - const {HttpProxyAgent} = require('http-proxy-agent'); - const {HttpsProxyAgent} = require('https-proxy-agent'); + // `pool.maxSockets` historically only took effect for proxied requests + // and keep-alive (`forever`) agents; other agent options have no undici + // equivalent and are ignored + const maxSockets = reqOpts.pool?.maxSockets; + const connections = + typeof maxSockets === 'number' && Number.isFinite(maxSockets) + ? maxSockets + : null; - const Agent = isHttp ? HttpProxyAgent : HttpsProxyAgent; - return new Agent(proxy, poolOptions); + if (proxy && shouldUseProxy) { + const key = `proxy:${proxy}:${connections}`; + if (!pool.has(key)) { + pool.set( + key, + new ProxyAgent({uri: proxy, ...(connections !== null && {connections})}) + ); + } + return withRedirects(pool.get(key)!); } - let key = isHttp ? 'http' : 'https'; - - if (reqOpts.forever) { - key += ':forever'; - + if (reqOpts.forever && connections !== null) { + const key = `agent:${connections}`; if (!pool.has(key)) { - // tslint:disable-next-line variable-name - const Agent = isHttp ? HTTPAgent : HTTPSAgent; - pool.set(key, new Agent({...poolOptions, keepAlive: true})); + pool.set(key, new Agent({connections})); } + return withRedirects(pool.get(key)!); } - return pool.get(key); + return withRedirects(getGlobalDispatcher()); } diff --git a/core/packages/teeny-request/src/index.ts b/core/packages/teeny-request/src/index.ts index 931861ad798c..5201b9304491 100644 --- a/core/packages/teeny-request/src/index.ts +++ b/core/packages/teeny-request/src/index.ts @@ -16,19 +16,21 @@ */ import {Agent, AgentOptions as HttpsAgentOptions} from 'https'; -import {AgentOptions as HttpAgentOptions} from 'http'; -import type * as f from 'node-fetch' with {'resolution-mode': 'import'}; +import {AgentOptions as HttpAgentOptions, STATUS_CODES} from 'http'; import {PassThrough, Readable, pipeline} from 'stream'; -import {getAgent} from './agents'; +import {promisify} from 'util'; +import * as zlib from 'zlib'; +import {Dispatcher, request as undiciRequest} from 'undici'; +import {getDispatcher} from './agents'; import {TeenyStatistics} from './TeenyStatistics'; import {randomUUID} from 'crypto'; // eslint-disable-next-line @typescript-eslint/no-var-requires const streamEvents = require('stream-events'); -import type nodeFetch from 'node-fetch' with {'resolution-mode': 'import'}; - -const fetch = (...args: Parameters) => - import('node-fetch').then(({default: fetch}) => fetch(...args)); +const gunzip = promisify(zlib.gunzip); +const inflate = promisify(zlib.inflate); +const inflateRaw = promisify(zlib.inflateRaw); +const brotliDecompress = promisify(zlib.brotliDecompress); export interface CoreOptions { method?: string; @@ -90,49 +92,67 @@ interface Headers { [index: string]: any; } +interface UndiciRequestOptions { + method: string; + headers: Headers; + body?: string | Buffer | Readable; + dispatcher: Dispatcher; + headersTimeout?: number; + bodyTimeout?: number; +} + /** - * Convert options from Request to Fetch format + * Set a header, replacing any casing variant of it. * @private - * @param reqOpts Request options */ -function requestToFetchOptions(reqOpts: Options) { - const options: f.RequestInit = { - method: reqOpts.method || 'GET', - ...(reqOpts.timeout && {timeout: reqOpts.timeout}), - ...(typeof reqOpts.gzip === 'boolean' && {compress: reqOpts.gzip}), - }; +function setHeader(headers: Headers, name: string, value: string) { + for (const key of Object.keys(headers)) { + if (key.toLowerCase() === name.toLowerCase()) { + delete headers[key]; + } + } + headers[name] = value; +} - if (typeof reqOpts.json === 'object') { - // Add Content-type: application/json header - reqOpts.headers = reqOpts.headers || {}; - if (reqOpts.headers instanceof globalThis.Headers) { - reqOpts.headers.set('Content-Type', 'application/json'); - } else { - reqOpts.headers['Content-Type'] = 'application/json'; +/** + * Check whether a header is set, in any casing. + * @private + */ +function hasHeader(headers: Headers, name: string) { + return Object.keys(headers).some( + key => key.toLowerCase() === name.toLowerCase() + ); +} + +/** + * Convert options from Request to undici format + * @private + * @param reqOpts Request options + */ +function requestToUndiciOptions(reqOpts: Options) { + let headers: Headers = {}; + if (reqOpts.headers instanceof globalThis.Headers) { + for (const pair of reqOpts.headers.entries()) { + headers[pair[0]] = pair[1]; } + } else if (reqOpts.headers) { + headers = {...reqOpts.headers}; + } - // Set body to JSON representation of value - options.body = JSON.stringify(reqOpts.json); + let body: string | Buffer | Readable | undefined; + if (typeof reqOpts.json === 'object') { + setHeader(headers, 'Content-Type', 'application/json'); + body = JSON.stringify(reqOpts.json); } else { if (Buffer.isBuffer(reqOpts.body)) { - options.body = reqOpts.body; + body = reqOpts.body; } else if (typeof reqOpts.body !== 'string') { - options.body = JSON.stringify(reqOpts.body); + body = JSON.stringify(reqOpts.body); } else { - options.body = reqOpts.body; + body = reqOpts.body; } } - if (reqOpts.headers instanceof globalThis.Headers) { - options.headers = {}; - for (const pair of reqOpts.headers.entries()) { - options.headers[pair[0]] = pair[1]; - } - } else { - // eslint-disable-next-line @typescript-eslint/no-explicit-any - options.headers = reqOpts.headers as any; - } - let uri = ((reqOpts as OptionsWithUri).uri || (reqOpts as OptionsWithUrl).url) as string; @@ -147,37 +167,148 @@ function requestToFetchOptions(reqOpts: Options) { uri = uri + '?' + params; } - options.agent = getAgent(uri, reqOpts); + const options: UndiciRequestOptions = { + method: reqOpts.method || 'GET', + // copied so that `userHeaders` stays as the caller provided them + headers: {...headers}, + body, + dispatcher: getDispatcher(uri, reqOpts), + ...(reqOpts.timeout && { + headersTimeout: reqOpts.timeout, + bodyTimeout: reqOpts.timeout, + }), + }; - return {uri, options}; + return {uri, options, userHeaders: headers}; +} + +/** + * Surface the underlying system error code (e.g. ECONNRESET) that undici + * wraps in its own error types, since downstream retry logic keys off + * `err.code`. + * @private + */ +function normalizeError(err: Error): Error { + const error = err as Error & {code?: unknown; cause?: {code?: unknown}}; + const causeCode = error?.cause?.code; + if ( + typeof causeCode === 'string' && + (error.code === undefined || String(error.code).startsWith('UND_')) + ) { + error.code = causeCode; + } + return error; } /** - * Convert a response from `fetch` to `request` format. + * Convert a response from `undici` to `request` format. * @private - * @param opts The `request` options used to create the request. - * @param res The Fetch response + * @param uri The request uri. + * @param userHeaders The request headers as provided by the caller. + * @param res The undici response * @returns A `request` response object */ -function fetchToRequestResponse(opts: f.RequestInit, res: f.Response) { +function undiciToRequestResponse( + uri: string, + userHeaders: Headers, + res: Dispatcher.ResponseData +) { const request = {} as Request; - request.agent = (opts.agent as Agent) || false; - request.headers = (opts.headers || {}) as Headers; - request.href = res.url; - // headers need to be converted from a map to an obj - const resHeaders = {} as Headers; - res.headers.forEach((value, key) => (resHeaders[key] = value)); + // connection pooling is managed by undici dispatchers, so there is no + // per-request http.Agent to expose + request.agent = false; + request.headers = userHeaders; + const history = (res.context as {history?: URL[]} | undefined)?.history; + request.href = history?.length ? String(history[history.length - 1]) : uri; + const resHeaders = {...res.headers} as Headers; const response = Object.assign(res.body as {}, { - statusCode: res.status, - statusMessage: res.statusText, + statusCode: res.statusCode, + statusMessage: STATUS_CODES[res.statusCode] || '', request, - body: res.body, headers: resHeaders, toJSON: () => ({headers: resHeaders}), + }) as unknown as Response; + // undici's response body has a getter-only `body` property (the web + // stream accessor), so it cannot be set through Object.assign + Object.defineProperty(response, 'body', { + value: res.body, + writable: true, + enumerable: true, + configurable: true, }); - return response as Response; + return response; +} + +/** + * Read the response body into a string, decompressing it if requested + * (undici, unlike fetch, hands back the raw bytes). + * @private + */ +async function readResponseBody( + res: Dispatcher.ResponseData, + decompress: boolean +): Promise { + const raw = Buffer.from(await res.body.arrayBuffer()); + if (!decompress || raw.length === 0) { + return raw.toString(); + } + const encoding = String(res.headers['content-encoding'] || '').toLowerCase(); + if (encoding === 'gzip' || encoding === 'x-gzip') { + return (await gunzip(raw)).toString(); + } + if (encoding === 'br') { + return (await brotliDecompress(raw)).toString(); + } + if (encoding === 'deflate') { + try { + return (await inflate(raw)).toString(); + } catch { + // some servers send raw deflate data without the zlib wrapper + return (await inflateRaw(raw)).toString(); + } + } + return raw.toString(); +} + +/** + * Read a callback-mode response and invoke the callback with it. + * @private + */ +function handleCallbackResponse( + uri: string, + userHeaders: Headers, + res: Dispatcher.ResponseData, + decompress: boolean, + callback: RequestCallback +) { + const header = String(res.headers['content-type'] || ''); + const response = undiciToRequestResponse(uri, userHeaders, res); + readResponseBody(res, decompress).then( + text => { + if ( + (header === 'application/json' || + header === 'application/json; charset=utf-8') && + response.statusCode !== 204 + ) { + try { + const json = JSON.parse(text); + response.body = json; + callback(null, response, json); + } catch (err) { + callback(err as Error, response, text); + } + return; + } + + response.body = text; + callback(null, response, text); + }, + err => { + callback(normalizeError(err), response, undefined); + } + ); } /** @@ -214,9 +345,17 @@ function teenyRequest(reqOpts: Options): Request; function teenyRequest(reqOpts: Options, callback: RequestCallback): void; function teenyRequest( reqOpts: Options, - callback?: RequestCallback, + callback?: RequestCallback ): Request | void { - const {uri, options} = requestToFetchOptions(reqOpts); + const {uri, options, userHeaders} = requestToUndiciOptions(reqOpts); + + // Callback mode transparently decompresses unless the caller opted out, + // like node-fetch did. Stream mode never does: consumers rely on getting + // the raw bytes (e.g. for integrity validation). + const decompress = reqOpts.gzip !== false && callback !== undefined; + if (decompress && !hasHeader(options.headers, 'Accept-Encoding')) { + options.headers['Accept-Encoding'] = 'gzip, deflate, br'; + } const multipart = reqOpts.multipart as RequestPart[]; if (reqOpts.multipart && multipart.length === 2) { @@ -225,48 +364,24 @@ function teenyRequest( throw new Error('Multipart without callback is not implemented.'); } const boundary: string = randomUUID(); - (options.headers as Headers)['Content-Type'] = - `multipart/related; boundary=${boundary}`; + setHeader( + options.headers, + 'Content-Type', + `multipart/related; boundary=${boundary}` + ); options.body = createMultipartStream(boundary, multipart); // Multipart upload teenyRequest.stats.requestStarting(); - fetch(uri, options).then( + undiciRequest(uri, options).then( res => { teenyRequest.stats.requestFinished(); - const header = res.headers.get('content-type'); - const response = fetchToRequestResponse(options, res); - const body = response.body; - if ( - header === 'application/json' || - header === 'application/json; charset=utf-8' - ) { - res.json().then( - json => { - response.body = json; - callback(null, response, json); - }, - (err: Error) => { - callback(err, response, body); - }, - ); - return; - } - - res.text().then( - text => { - response.body = text; - callback(null, response, text); - }, - err => { - callback(err, response, body); - }, - ); + handleCallbackResponse(uri, userHeaders, res, decompress, callback); }, err => { teenyRequest.stats.requestFinished(); - callback(err, null!, null); - }, + callback(normalizeError(err), null!, null); + } ); return; } @@ -274,88 +389,60 @@ function teenyRequest( if (callback === undefined) { // Stream mode const requestStream = streamEvents(new PassThrough()); - // eslint-disable-next-line @typescript-eslint/no-explicit-any - let responseStream: any; + let responseStream: Readable | undefined; + let piped = false; + const pipeResponse = () => { + piped = true; + pipeline(responseStream!, requestStream, () => {}); + }; requestStream.once('reading', () => { if (responseStream) { - pipeline(responseStream, requestStream, () => {}); + pipeResponse(); } else { - requestStream.once('response', () => { - pipeline(responseStream, requestStream, () => {}); - }); + requestStream.once('response', pipeResponse); + } + }); + // a consumer tearing the stream down without reading it must abort + // the in-flight request, or the socket would be left occupied + requestStream.once('close', () => { + if (!piped && responseStream) { + responseStream.destroy(); } }); - options.compress = false; teenyRequest.stats.requestStarting(); - fetch(uri, options).then( + undiciRequest(uri, options).then( res => { teenyRequest.stats.requestFinished(); responseStream = res.body; responseStream.on('error', (err: Error) => { - requestStream.emit('error', err); + requestStream.emit('error', normalizeError(err)); }); - const response = fetchToRequestResponse(options, res); + const response = undiciToRequestResponse(uri, userHeaders, res); requestStream.emit('response', response); }, err => { teenyRequest.stats.requestFinished(); - requestStream.emit('error', err); - }, + requestStream.emit('error', normalizeError(err)); + } ); - // fetch doesn't supply the raw HTTP stream, instead it - // returns a PassThrough piped from the HTTP response - // stream. return requestStream as Request; } // GET or POST with callback teenyRequest.stats.requestStarting(); - fetch(uri, options).then( + undiciRequest(uri, options).then( res => { teenyRequest.stats.requestFinished(); - const header = res.headers.get('content-type'); - const response = fetchToRequestResponse(options, res); - const body = response.body; - if ( - header === 'application/json' || - header === 'application/json; charset=utf-8' - ) { - if (response.statusCode === 204) { - // Probably a DELETE - callback(null, response, body); - return; - } - res.json().then( - json => { - response.body = json; - callback(null, response, json); - }, - err => { - callback(err, response, body); - }, - ); - return; - } - - res.text().then( - text => { - const response = fetchToRequestResponse(options, res); - response.body = text; - callback(null, response, text); - }, - err => { - callback(err, response, body); - }, - ); + handleCallbackResponse(uri, userHeaders, res, decompress, callback); }, err => { teenyRequest.stats.requestFinished(); - callback(err, null!, null); - }, + callback(normalizeError(err), null!, null); + } ); return; } diff --git a/core/packages/teeny-request/test/agents.ts b/core/packages/teeny-request/test/agents.ts index 0f8deee0e76a..c45326141daf 100644 --- a/core/packages/teeny-request/test/agents.ts +++ b/core/packages/teeny-request/test/agents.ts @@ -16,23 +16,10 @@ */ import assert from 'assert'; -import {describe, it, afterEach} from 'mocha'; -import * as http from 'http'; -import * as https from 'https'; +import {describe, it, afterEach, beforeEach} from 'mocha'; import * as sinon from 'sinon'; -import {getAgent, pool} from '../src/agents'; - -// eslint-disable-next-line @typescript-eslint/no-var-requires -let HttpProxyAgent = require('http-proxy-agent'); -// eslint-disable-next-line @typescript-eslint/no-var-requires -let HttpsProxyAgent = require('https-proxy-agent'); - -if (HttpProxyAgent.HttpProxyAgent) { - HttpProxyAgent = HttpProxyAgent.HttpProxyAgent; -} -if (HttpsProxyAgent.HttpsProxyAgent) { - HttpsProxyAgent = HttpsProxyAgent.HttpsProxyAgent; -} +import {Agent, ProxyAgent} from 'undici'; +import {getDispatcher, pool} from '../src/agents'; describe('agents', () => { const httpUri = 'http://example.com'; @@ -44,105 +31,72 @@ describe('agents', () => { pool.clear(); }); - describe('getAgent', () => { + function pooledProxyAgent(): ProxyAgent | undefined { + return [...pool.values()].find( + dispatcher => dispatcher instanceof ProxyAgent + ) as ProxyAgent | undefined; + } + + describe('getDispatcher', () => { const defaultOptions = {uri: httpUri}; - it('should return undefined by default', () => { - const agent = getAgent(httpUri, defaultOptions); - assert.strictEqual(agent, undefined); + it('should use the global dispatcher by default', () => { + const dispatcher = getDispatcher(httpUri, defaultOptions); + assert.ok(dispatcher); + assert.strictEqual(pool.size, 0); + }); + + it('should return the same dispatcher for repeated default requests', () => { + const dispatcher1 = getDispatcher(httpUri, defaultOptions); + const dispatcher2 = getDispatcher(httpsUri, defaultOptions); + assert.strictEqual(dispatcher1, dispatcher2); }); describe('proxy', () => { - const envVars = [ - 'http_proxy', - 'https_proxy', - 'HTTP_PROXY', - 'HTTPS_PROXY', - ]; + const envVars = ['http_proxy', 'https_proxy', 'HTTP_PROXY', 'HTTPS_PROXY']; const noProxyEnvVars = ['no_proxy', 'NO_PROXY']; + const proxy = 'https://hello.there:8080'; - describe('http', () => { - const uri = httpUri; - const proxy = 'http://hello.there:8080'; - const proxyExpected = { - hostname: 'hello.there', - port: '8080', - protocol: 'http:', - }; - - it('should respect the proxy option', () => { - const options = Object.assign({proxy}, defaultOptions); - const agent = getAgent(uri, options); - assert(agent instanceof HttpProxyAgent); - - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const {proxy: proxyActual}: any = agent!; - assert.strictEqual(proxyActual.protocol, proxyExpected.protocol); - assert.strictEqual(proxyActual.hostname, proxyExpected.hostname); - assert.strictEqual(proxyActual.port, proxyExpected.port); - }); - - envVars.forEach(envVar => { - it(`should respect the ${envVar} env var`, () => { - process.env[envVar] = proxy; - const agent = getAgent(uri, defaultOptions); - assert(agent instanceof HttpProxyAgent); - delete process.env[envVar]; - }); - }); + it('should respect the proxy option', () => { + const options = Object.assign({proxy}, defaultOptions); + const dispatcher = getDispatcher(httpsUri, options); + assert.ok(dispatcher); + assert.ok(pooledProxyAgent()); }); - describe('https', () => { - const uri = httpsUri; - const proxy = 'https://hello.there:8080'; - const proxyExpected = { - hostname: 'hello.there', - port: '8080', - protocol: 'https:', - }; - - it('should respect the proxy option', () => { - const options = Object.assign({proxy}, defaultOptions); - const agent = getAgent(uri, options); - assert(agent instanceof HttpsProxyAgent); - - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const {proxy: proxyActual}: any = agent!; - assert.strictEqual(proxyActual.protocol, proxyExpected.protocol); - assert.strictEqual(proxyActual.hostname, proxyExpected.hostname); - assert.strictEqual(proxyActual.port, proxyExpected.port); - }); + it('should cache the proxy dispatcher', () => { + const options = Object.assign({proxy}, defaultOptions); + const dispatcher1 = getDispatcher(httpsUri, options); + const dispatcher2 = getDispatcher(httpsUri, options); + assert.strictEqual(dispatcher1, dispatcher2); + assert.strictEqual(pool.size, 1); + }); - envVars.forEach(envVar => { - it(`should respect the ${envVar} env var`, () => { - process.env[envVar] = proxy; - const agent = getAgent(uri, defaultOptions); - assert(agent instanceof HttpsProxyAgent); - delete process.env[envVar]; - }); + envVars.forEach(envVar => { + it(`should respect the ${envVar} env var`, () => { + sandbox.stub(process, 'env').value({[envVar]: proxy}); + getDispatcher(httpsUri, defaultOptions); + assert.ok(pooledProxyAgent()); }); }); describe('no_proxy', () => { - const uri = httpsUri; - const proxy = 'https://hello.there:8080'; - beforeEach(() => { sandbox.stub(process, 'env').value({}); }); - noProxyEnvVars.forEach(noProxEnvVar => { - it(`should respect the proxy option, even if is in ${noProxEnvVar} env var`, () => { - process.env[noProxEnvVar] = new URL(uri).hostname; + noProxyEnvVars.forEach(noProxyEnvVar => { + it(`should respect the proxy option, even if in ${noProxyEnvVar} env var`, () => { + process.env[noProxyEnvVar] = new URL(httpsUri).hostname; const options = Object.assign({proxy}, defaultOptions); - const agent = getAgent(uri, options); - assert(agent instanceof HttpsProxyAgent); + getDispatcher(httpsUri, options); + assert.ok(pooledProxyAgent()); }); }); - noProxyEnvVars.forEach(noProxEnvVar => { + noProxyEnvVars.forEach(noProxyEnvVar => { envVars.forEach(envVar => { const root = 'example.com'; const subDomain = 'abc.' + root; @@ -158,13 +112,12 @@ describe('agents', () => { ]; for (const {name, value} of cases) { - it(`should respect the ${noProxEnvVar} env var > ${envVar}': ${name}`, () => { + it(`should respect the ${noProxyEnvVar} env var > ${envVar}': ${name}`, () => { process.env[envVar] = proxy; - process.env[noProxEnvVar] = value; - const agent = getAgent(uri.toString(), defaultOptions); - assert(!(agent instanceof HttpProxyAgent)); - assert(!(agent instanceof HttpsProxyAgent)); + process.env[noProxyEnvVar] = value; + getDispatcher(uri.toString(), defaultOptions); + assert.strictEqual(pooledProxyAgent(), undefined); }); } }); @@ -173,102 +126,70 @@ describe('agents', () => { }); describe('forever', () => { - describe('http', () => { - const uri = httpUri; - const options = Object.assign({forever: true}, defaultOptions); - - it('should return an http Agent', () => { - const agent = getAgent(uri, options)!; - assert(agent instanceof http.Agent); - }); - - it('should cache the agent', () => { - const agent1 = getAgent(uri, options); - const agent2 = getAgent(uri, options); - assert.strictEqual(agent1, agent2); - }); - }); - - describe('https', () => { - const uri = httpsUri; + it('should use the global dispatcher, which keeps connections alive', () => { const options = Object.assign({forever: true}, defaultOptions); - - it('should return an http Agent', () => { - const agent = getAgent(uri, options)!; - assert(agent instanceof https.Agent); - }); - - it('should cache the agent', () => { - const agent1 = getAgent(uri, options); - const agent2 = getAgent(uri, options); - assert.strictEqual(agent1, agent2); - }); + const dispatcher = getDispatcher(httpUri, options); + assert.ok(dispatcher); + assert.strictEqual(pool.size, 0); }); }); describe('pool', () => { - describe('http', () => { - const uri = httpUri; - - it('should pass AgentOptions from pool config when providing agent', () => { - const options = Object.assign( - { - forever: true, - pool: { - maxSockets: 1000, - }, + it('should create a dedicated dispatcher for a socket limit', () => { + const options = Object.assign( + { + forever: true, + pool: { + maxSockets: 1000, }, - defaultOptions, - ); - const agent = getAgent(uri, options); - assert.strictEqual(agent!.maxSockets, 1000); - }); + }, + defaultOptions + ); + getDispatcher(httpUri, options); + assert.ok([...pool.values()].some(d => d instanceof Agent)); + }); - it('should not set global AgentOptions from only pool config', () => { - const options = Object.assign( - { - pool: { - maxSockets: 1000, - }, + it('should cache the dispatcher for a socket limit', () => { + const options = Object.assign( + { + forever: true, + pool: { + maxSockets: 1000, }, - defaultOptions, - ); - const agent = getAgent(uri, options); - assert.strictEqual(agent, undefined); - assert.notStrictEqual(http.globalAgent.maxSockets, 1000); - }); + }, + defaultOptions + ); + const dispatcher1 = getDispatcher(httpUri, options); + const dispatcher2 = getDispatcher(httpUri, options); + assert.strictEqual(dispatcher1, dispatcher2); + assert.strictEqual(pool.size, 1); }); - describe('https', () => { - const uri = httpsUri; - - it('should pass AgentOptions from pool config when providing agent', () => { - const options = Object.assign( - { - forever: true, - pool: { - maxSockets: 1000, - }, + it('should ignore pool config without forever or proxy', () => { + const options = Object.assign( + { + pool: { + maxSockets: 1000, }, - defaultOptions, - ); - const agent = getAgent(uri, options); - assert.strictEqual(agent!.maxSockets, 1000); - }); + }, + defaultOptions + ); + getDispatcher(httpUri, options); + assert.strictEqual(pool.size, 0); + }); - it('should not set global AgentOptions from only pool config', () => { - const options = Object.assign( - { - pool: { - maxSockets: 1000, - }, + it('should ignore an unlimited socket limit', () => { + const options = Object.assign( + { + forever: true, + pool: { + maxSockets: Infinity, }, - defaultOptions, - ); - const agent = getAgent(uri, options); - assert.strictEqual(agent, undefined); - assert.notStrictEqual(https.globalAgent.maxSockets, 1000); - }); + }, + defaultOptions + ); + getDispatcher(httpUri, options); + assert.strictEqual(pool.size, 0); }); }); }); diff --git a/core/packages/teeny-request/test/index.ts b/core/packages/teeny-request/test/index.ts index e34d697a2a58..052a8f423e0c 100644 --- a/core/packages/teeny-request/test/index.ts +++ b/core/packages/teeny-request/test/index.ts @@ -16,36 +16,80 @@ */ import assert from 'assert'; -import {describe, it, afterEach, beforeEach} from 'mocha'; -import nock from 'nock'; +import {describe, it, before, after, afterEach, beforeEach} from 'mocha'; +import * as http from 'http'; +import {AddressInfo} from 'net'; import {Readable} from 'stream'; +import * as zlib from 'zlib'; import * as sinon from 'sinon'; +import {getGlobalDispatcher} from 'undici'; import {teenyRequest} from '../src'; import {TeenyStatistics, TeenyStatisticsWarning} from '../src/TeenyStatistics'; import {pool} from '../src/agents'; -// eslint-disable-next-line @typescript-eslint/no-var-requires -const HttpProxyAgent = require('http-proxy-agent'); -// eslint-disable-next-line @typescript-eslint/no-var-requires -const HttpsProxyAgent = require('https-proxy-agent'); - -nock.disableNetConnect(); -const uri = 'https://example.com'; - -function mockJson() { - return nock(uri).get('/').reply(200, {hello: '🌍'}); +interface ReceivedRequest { + method?: string; + url?: string; + headers: http.IncomingHttpHeaders; + body: Buffer; } -function mockError() { - return nock(uri).get('/').replyWithError('mock err'); -} +type Handler = ( + req: http.IncomingMessage, + res: http.ServerResponse, + body: Buffer +) => void; + +const jsonHandler: Handler = (req, res) => { + res.setHeader('content-type', 'application/json'); + res.end(JSON.stringify({hello: '🌍'})); +}; describe('teeny', () => { const sandbox = sinon.createSandbox(); let emitWarnStub: sinon.SinonStub; let statsStub: sinon.SinonStubbedInstance; + let server: http.Server; + let uri: string; + let deadUri: string; + let handler: Handler; + let received: ReceivedRequest[] = []; + + before(async () => { + server = http.createServer((req, res) => { + const chunks: Buffer[] = []; + req.on('data', chunk => chunks.push(chunk)); + req.on('end', () => { + const body = Buffer.concat(chunks); + received.push({ + method: req.method, + url: req.url, + headers: req.headers, + body, + }); + handler(req, res, body); + }); + }); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + uri = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + + // grab a port with nothing listening on it, for connection failures + const dead = http.createServer(); + await new Promise(resolve => dead.listen(0, '127.0.0.1', resolve)); + deadUri = `http://127.0.0.1:${(dead.address() as AddressInfo).port}`; + await new Promise(resolve => dead.close(() => resolve())); + }); + + after(async () => { + server.closeAllConnections(); + await new Promise(resolve => server.close(() => resolve())); + await getGlobalDispatcher().close(); + }); beforeEach(() => { + handler = jsonHandler; + received = []; + emitWarnStub = sandbox.stub(process, 'emitWarning'); // don't mask other process warns @@ -62,55 +106,50 @@ describe('teeny', () => { pool.clear(); sandbox.restore(); teenyRequest.resetStats(); - nock.cleanAll(); }); - it('should get JSON', async () => { - const scope = mockJson(); + it('should get JSON', done => { teenyRequest({uri}, (error, response, body) => { assert.ifError(error); assert.strictEqual(response.statusCode, 200); assert.ok(body.hello); - scope.done(); - // done(); + done(); }); }); - it('should set defaults', async () => { - const scope = mockJson(); + it('should set defaults', done => { const defaultRequest = teenyRequest.defaults({timeout: 60000}); defaultRequest({uri}, (error, response, body) => { assert.ifError(error); assert.strictEqual(response.statusCode, 200); assert.ok(body.hello); - scope.done(); + done(); }); }); - it('response event emits object compatible with request module', async () => { + it('response event emits object compatible with request module', done => { const reqHeaders = {fruit: 'banana'}; - const resHeaders = {veggies: 'carrots'}; - const scope = nock(uri).get('/').reply(202, 'ok', resHeaders); + handler = (req, res) => { + res.setHeader('veggies', 'carrots'); + res.statusCode = 202; + res.end('ok'); + }; const reqStream = teenyRequest({uri, headers: reqHeaders}); reqStream .on('response', res => { assert.strictEqual(res.statusCode, 202); assert.strictEqual(res.headers.veggies, 'carrots'); assert.deepStrictEqual(res.request.headers, reqHeaders); - assert.deepStrictEqual(res.toJSON(), { - headers: resHeaders, - }); + assert.strictEqual(res.toJSON().headers.veggies, 'carrots'); assert(res instanceof Readable); - scope.done(); + done(); }) - .on('error', err => { - throw err; - }); + .on('error', done); + reqStream.resume(); }); - it('should include the request in the response', async () => { + it('should include the request in the response', done => { const path = '/?dessert=pie'; - const scope = nock(uri).get(path).reply(202); const headers = {dinner: 'tacos'}; const url = `${uri}${path}`; teenyRequest({url, headers}, (error, response) => { @@ -118,129 +157,91 @@ describe('teeny', () => { const req = response.request; assert.deepStrictEqual(req.headers, headers); assert.strictEqual(req.href, url); - scope.done(); + assert.strictEqual(received[0].url, path); + done(); }); }); - it('should not wrap the error', async () => { - const scope = nock(uri) - .get('/') - .reply(200, '🚨', {'content-type': 'application/json'}); + it('should not wrap the error', done => { + handler = (req, res) => { + res.setHeader('content-type', 'application/json'); + res.end('🚨'); + }; teenyRequest({uri}, err => { assert.ok(err); - assert.ok(err!.message.match(/^invalid json response body/)); - scope.done(); + assert.ok(err!.message.match(/JSON/)); + done(); }); }); - it('should include headers in the response', async () => { - const headers = {dinner: 'tacos'}; - const body = {hello: '🌍'}; - const scope = nock(uri).get('/').reply(200, body, headers); + it('should include headers in the response', done => { + handler = (req, res) => { + res.setHeader('dinner', 'tacos'); + res.setHeader('content-type', 'application/json'); + res.end(JSON.stringify({hello: '🌍'})); + }; teenyRequest({uri}, (err, res) => { assert.ifError(err); - assert.strictEqual(headers['dinner'], res.headers['dinner']); - scope.done(); + assert.strictEqual(res.headers['dinner'], 'tacos'); + done(); }); }); it('should accept fetch Headers', done => { const body = {dish: '🍕'}; - const scope = nock(uri) - .post('/') - .matchHeader('dinner', 'pizza') - .matchHeader('content-type', 'application/json') - .reply(200, body, {country: 'Italy'}); - + handler = (req, res) => { + res.setHeader('country', 'Italy'); + res.end(); + }; const headers = new Headers(); headers.set('dinner', 'pizza'); teenyRequest({uri, headers, json: body, method: 'POST'}, (err, res) => { assert.ifError(err); assert.strictEqual(res.headers['country'], 'Italy'); - assert.strictEqual(res.headers['content-type'], 'application/json'); - scope.done(); + assert.strictEqual(received[0].headers['dinner'], 'pizza'); + assert.strictEqual(received[0].headers['content-type'], 'application/json'); + assert.strictEqual(received[0].body.toString(), JSON.stringify(body)); done(); }); }); - it('should accept the forever option', async () => { - const scope = nock(uri).get('/').reply(200); + it('should accept the forever option', done => { teenyRequest({uri, forever: true}, (err, res) => { assert.ifError(err); - // eslint-disable-next-line @typescript-eslint/no-explicit-any - assert.strictEqual((res.request.agent as any).keepAlive, true); - scope.done(); + assert.strictEqual(res.request.agent, false); + done(); }); }); - it('should allow setting compress/gzip to true', async () => { - const reqheaders = { - 'Accept-Encoding': 'gzip,deflate', + it('should request and decompress gzip responses by default', done => { + const payload = JSON.stringify({hello: '🌍'}); + handler = (req, res) => { + assert.ok(String(req.headers['accept-encoding']).includes('gzip')); + res.setHeader('content-type', 'application/json'); + res.setHeader('content-encoding', 'gzip'); + res.end(zlib.gzipSync(payload)); }; - - const scope = nock(uri, {reqheaders}).get('/').reply(200); - - teenyRequest({uri, gzip: true}, err => { + teenyRequest({uri, gzip: true}, (err, res, body) => { assert.ifError(err); - scope.done(); + assert.strictEqual(res.statusCode, 200); + assert.deepStrictEqual(body, {hello: '🌍'}); + done(); }); }); - it('should allow setting compress/gzip to false', async () => { - const badheaders = ['Accept-Encoding']; - - const scope = nock(uri, {badheaders}).get('/').reply(200); - + it('should allow setting compress/gzip to false', done => { + handler = (req, res) => { + assert.strictEqual(req.headers['accept-encoding'], undefined); + res.end('ok'); + }; teenyRequest({uri, gzip: false}, err => { assert.ifError(err); - scope.done(); - }); - }); - - const envVars = ['http_proxy', 'https_proxy', 'HTTP_PROXY', 'HTTPS_PROXY']; - for (const v of envVars) { - it(`should respect ${v} environment variable for proxy config`, () => { - sandbox.stub(process, 'env').value({[v]: 'https://fake.proxy'}); - const expectedBody = {hello: '🌎'}; - const scope = nock(uri).get('/').reply(200, expectedBody); - teenyRequest({uri}, (err, res, body) => { - scope.done(); - assert.ifError(err); - assert.deepStrictEqual(expectedBody, body); - assert.ok(res.request.agent instanceof HttpsProxyAgent); - return; - }); - }); - } - - it('should create http proxy if upstream scheme is http', async () => { - sandbox.stub(process, 'env').value({http_proxy: 'https://fake.proxy'}); - const expectedBody = {hello: '🌎'}; - const scope = nock('http://example.com').get('/').reply(200, expectedBody); - teenyRequest({uri: 'http://example.com'}, (err, res, body) => { - scope.done(); - assert.ifError(err); - assert.deepStrictEqual(expectedBody, body); - assert.ok(res.request.agent instanceof HttpProxyAgent); - return; - }); - }); - - it('should use proxy if set in request options', async () => { - const expectedBody = {hello: '🌎'}; - const scope = nock(uri).get('/').reply(200, expectedBody); - teenyRequest({uri, proxy: 'https://fake.proxy'}, (err, res, body) => { - scope.done(); - assert.ifError(err); - assert.deepStrictEqual(expectedBody, body); - assert.ok(res.request.agent instanceof HttpsProxyAgent); - return; + done(); }); }); // see: https://github.com/googleapis/nodejs-storage/issues/798 it('should not throw exception when piped through pumpify', async () => { - const scope = mockJson(); const stream = teenyRequest({uri}); // set the encoding for the returned stream stream.setEncoding('utf8'); @@ -252,34 +253,27 @@ describe('teeny', () => { } assert.deepStrictEqual(JSON.parse(content.join('')), {hello: '🌍'}); - scope.done(); }); - it('should emit response event when called without callback', async () => { - const scope = mockJson(); - teenyRequest({uri}).on('response', res => { + it('should emit response event when called without callback', done => { + const stream = teenyRequest({uri}); + stream.on('response', res => { assert.ok(res); - scope.done(); - return; + done(); }); + stream.resume(); }); - it('should pipe response stream to user', () => { - const scope = mockJson(); + it('should pipe response stream to user', done => { teenyRequest({uri}) - .on('error', err => { - throw err; - }) - .on('data', () => { - scope.done(); + .on('error', done) + .once('data', () => { + done(); }); }); - it('should not pipe response stream to user unless they ask for it', async () => { - const scope = mockJson(); - const stream = teenyRequest({uri}).on('error', err => { - throw err; - }); + it('should not pipe response stream to user unless they ask for it', done => { + const stream = teenyRequest({uri}).on('error', done); stream.on('response', responseStream => { // We are using an internal property of Readable to get the number of // active readers. The property changed from `pipesCount: number` in @@ -288,16 +282,52 @@ describe('teeny', () => { responseStream.body._readableState.pipesCount ?? responseStream.body._readableState.pipes?.length; assert.strictEqual(numPipes, 0); - stream.on('data', () => { + stream.once('data', () => { numPipes = responseStream.body._readableState.pipesCount ?? responseStream.body._readableState.pipes?.length; assert.strictEqual(numPipes, 1); - scope.done(); + done(); }); }); }); + it('should deliver raw bytes in stream mode, even when compressed', done => { + const compressed = zlib.gzipSync('raw bytes for integrity validation'); + handler = (req, res) => { + res.setHeader('content-encoding', 'gzip'); + res.end(compressed); + }; + const stream = teenyRequest({ + uri, + gzip: true, + headers: {'accept-encoding': 'gzip'}, + }).on('error', done); + const chunks: Buffer[] = []; + stream.on('data', (chunk: Buffer) => chunks.push(chunk)); + stream.on('end', () => { + assert.ok(Buffer.concat(chunks).equals(compressed)); + done(); + }); + }); + + // see: https://github.com/googleapis/google-cloud-node/issues/9185 + it('should not emit MaxListenersExceededWarning in stream mode', async () => { + handler = (req, res) => { + res.end('x'.repeat(1024 * 1024)); + }; + const stream = teenyRequest({uri}); + for await (const chunk of stream) { + void chunk; + } + const maxListenersWarned = emitWarnStub + .getCalls() + .some(call => + String(call.args[0]).includes('MaxListenersExceededWarning') + ); + assert.strictEqual(maxListenersWarned, false); + }); + it('should expose TeenyStatistics instance', () => { assert.ok(teenyRequest.stats instanceof TeenyStatistics); }); @@ -318,101 +348,116 @@ describe('teeny', () => { assert.deepStrictEqual(newOptions, {concurrentRequests: 42}); }); - it('should emit warning on too many concurrent requests', () => { + it('should emit warning on too many concurrent requests', done => { statsStub.setOptions.restore(); statsStub.requestStarting.restore(); teenyRequest.stats.setOptions({concurrentRequests: 1}); - const scope = mockJson(); teenyRequest({uri}, () => { assert.ok(emitWarnStub.calledOnce); - scope.done(); + done(); }); }); - it('should track stats, callback mode, success', () => { - const scope = mockJson(); + it('should track stats, callback mode, success', done => { teenyRequest({uri}, () => { assert.ok(statsStub.requestStarting.calledOnceWithExactly()); assert.ok(statsStub.requestFinished.calledOnceWithExactly()); - scope.done(); + done(); }); }); - it('should track stats, callback mode, failure', () => { - const scope = mockError(); - teenyRequest({uri}, err => { + it('should track stats, callback mode, failure', done => { + teenyRequest({uri: deadUri}, err => { assert.ok(err); assert.ok(statsStub.requestStarting.calledOnceWithExactly()); assert.ok(statsStub.requestFinished.calledOnceWithExactly()); - scope.done(); + done(); }); }); - it('should track stats, stream mode, success', () => { - const scope = mockJson(); + it('should track stats, stream mode, success', done => { const readable = teenyRequest({uri}); assert.ok(statsStub.requestStarting.calledOnceWithExactly()); readable.once('response', () => { assert.ok(statsStub.requestFinished.calledOnceWithExactly()); - scope.done(); + done(); }); + readable.resume(); }); - it('should track stats, stream mode, failure', () => { - const scope = mockError(); - const readable = teenyRequest({uri}); + it('should track stats, stream mode, failure', done => { + const readable = teenyRequest({uri: deadUri}); assert.ok(statsStub.requestStarting.calledOnceWithExactly()); readable.once('error', err => { assert.ok(err); assert.ok(statsStub.requestFinished.calledOnceWithExactly()); - scope.done(); + done(); }); }); - it('should accept a Buffer as the body of a request', () => { - const scope = nock(uri).post('/', 'hello').reply(200, '🌍'); + it('should surface the system error code on connection failures', done => { + teenyRequest({uri: deadUri}, err => { + assert.ok(err); + assert.strictEqual( + (err as Error & {code?: string}).code, + 'ECONNREFUSED' + ); + done(); + }); + }); + + it('should accept a Buffer as the body of a request', done => { + handler = (req, res) => { + res.end('🌍'); + }; teenyRequest( {uri, method: 'POST', body: Buffer.from('hello')}, (error, response, body) => { assert.ifError(error); assert.strictEqual(response.statusCode, 200); assert.strictEqual(body, '🌍'); - scope.done(); - }, + assert.strictEqual(received[0].body.toString(), 'hello'); + done(); + } ); }); - it('should accept a plain string as the body of a request', () => { - const scope = nock(uri).post('/', 'hello').reply(200, '🌍'); + it('should accept a plain string as the body of a request', done => { + handler = (req, res) => { + res.end('🌍'); + }; teenyRequest( {uri, method: 'POST', body: 'hello'}, (error, response, body) => { assert.ifError(error); assert.strictEqual(response.statusCode, 200); assert.strictEqual(body, '🌍'); - scope.done(); - }, + assert.strictEqual(received[0].body.toString(), 'hello'); + done(); + } ); }); - it('should accept json as the body of a request', () => { - const body = {hello: '🌍'}; - const scope = nock(uri).post('/', JSON.stringify(body)).reply(200, '👋'); - teenyRequest({uri, method: 'POST', json: body}, (error, response, body) => { + it('should accept json as the body of a request', done => { + handler = (req, res) => { + res.end('👋'); + }; + const json = {hello: '🌍'}; + teenyRequest({uri, method: 'POST', json}, (error, response, body) => { assert.ifError(error); assert.strictEqual(response.statusCode, 200); assert.strictEqual(body, '👋'); - scope.done(); + assert.strictEqual(received[0].body.toString(), JSON.stringify(json)); + done(); }); }); // TODO multipart is broken with 2 strings // see: https://github.com/googleapis/teeny-request/issues/168 it.skip('should track stats, multipart mode, success', done => { - const scope = mockJson(); teenyRequest( { method: 'POST', @@ -423,27 +468,25 @@ describe('teeny', () => { () => { assert.ok(statsStub.requestStarting.calledOnceWithExactly()); assert.ok(statsStub.requestFinished.calledOnceWithExactly()); - scope.done(); done(); - }, + } ); }); - it.skip('should track stats, multipart mode, failure', () => { - const scope = mockError(); + it.skip('should track stats, multipart mode, failure', done => { teenyRequest( { method: 'POST', headers: {}, multipart: [{body: 'foo'}, {body: 'bar'}], - uri, + uri: deadUri, }, err => { assert.ok(err); assert.ok(statsStub.requestStarting.calledOnceWithExactly()); assert.ok(statsStub.requestFinished.calledOnceWithExactly()); - scope.done(); - }, + done(); + } ); }); @@ -453,7 +496,7 @@ describe('teeny', () => { teenyRequest({uri: ''}); }, /Missing uri or url in reqOpts/, - 'Did not throw with expected message', + 'Did not throw with expected message' ); }); @@ -463,7 +506,7 @@ describe('teeny', () => { teenyRequest({url: ''}); }, /Missing uri or url in reqOpts/, - 'Did not throw with expected message', + 'Did not throw with expected message' ); }); });