fix: 修复配额说明重复和undefined问题

- 在editStorageForm中初始化oss_storage_quota_value和oss_quota_unit
- 删除重复的旧配额说明块,保留新的当前配额设置显示

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-22 19:39:53 +08:00
commit 4350113979
7649 changed files with 897277 additions and 0 deletions

View File

@@ -0,0 +1 @@
export const NODEJS_TIMEOUT_ERROR_CODES = ["ECONNRESET", "EPIPE", "ETIMEDOUT"];

View File

@@ -0,0 +1,9 @@
const getTransformedHeaders = (headers) => {
const transformedHeaders = {};
for (const name of Object.keys(headers)) {
const headerValues = headers[name];
transformedHeaders[name] = Array.isArray(headerValues) ? headerValues.join(",") : headerValues;
}
return transformedHeaders;
};
export { getTransformedHeaders };

View File

@@ -0,0 +1,3 @@
export * from "./node-http-handler";
export * from "./node-http2-handler";
export * from "./stream-collector";

View File

@@ -0,0 +1,224 @@
import { HttpResponse } from "@smithy/protocol-http";
import { buildQueryString } from "@smithy/querystring-builder";
import { Agent as hAgent, request as hRequest } from "http";
import { Agent as hsAgent, request as hsRequest } from "https";
import { NODEJS_TIMEOUT_ERROR_CODES } from "./constants";
import { getTransformedHeaders } from "./get-transformed-headers";
import { setConnectionTimeout } from "./set-connection-timeout";
import { setRequestTimeout } from "./set-request-timeout";
import { setSocketKeepAlive } from "./set-socket-keep-alive";
import { setSocketTimeout } from "./set-socket-timeout";
import { timing } from "./timing";
import { writeRequestBody } from "./write-request-body";
export const DEFAULT_REQUEST_TIMEOUT = 0;
export class NodeHttpHandler {
config;
configProvider;
socketWarningTimestamp = 0;
externalAgent = false;
metadata = { handlerProtocol: "http/1.1" };
static create(instanceOrOptions) {
if (typeof instanceOrOptions?.handle === "function") {
return instanceOrOptions;
}
return new NodeHttpHandler(instanceOrOptions);
}
static checkSocketUsage(agent, socketWarningTimestamp, logger = console) {
const { sockets, requests, maxSockets } = agent;
if (typeof maxSockets !== "number" || maxSockets === Infinity) {
return socketWarningTimestamp;
}
const interval = 15_000;
if (Date.now() - interval < socketWarningTimestamp) {
return socketWarningTimestamp;
}
if (sockets && requests) {
for (const origin in sockets) {
const socketsInUse = sockets[origin]?.length ?? 0;
const requestsEnqueued = requests[origin]?.length ?? 0;
if (socketsInUse >= maxSockets && requestsEnqueued >= 2 * maxSockets) {
logger?.warn?.(`@smithy/node-http-handler:WARN - socket usage at capacity=${socketsInUse} and ${requestsEnqueued} additional requests are enqueued.
See https://docs.aws.amazon.com/sdk-for-javascript/v3/developer-guide/node-configuring-maxsockets.html
or increase socketAcquisitionWarningTimeout=(millis) in the NodeHttpHandler config.`);
return Date.now();
}
}
}
return socketWarningTimestamp;
}
constructor(options) {
this.configProvider = new Promise((resolve, reject) => {
if (typeof options === "function") {
options()
.then((_options) => {
resolve(this.resolveDefaultConfig(_options));
})
.catch(reject);
}
else {
resolve(this.resolveDefaultConfig(options));
}
});
}
resolveDefaultConfig(options) {
const { requestTimeout, connectionTimeout, socketTimeout, socketAcquisitionWarningTimeout, httpAgent, httpsAgent, throwOnRequestTimeout, } = options || {};
const keepAlive = true;
const maxSockets = 50;
return {
connectionTimeout,
requestTimeout,
socketTimeout,
socketAcquisitionWarningTimeout,
throwOnRequestTimeout,
httpAgent: (() => {
if (httpAgent instanceof hAgent || typeof httpAgent?.destroy === "function") {
this.externalAgent = true;
return httpAgent;
}
return new hAgent({ keepAlive, maxSockets, ...httpAgent });
})(),
httpsAgent: (() => {
if (httpsAgent instanceof hsAgent || typeof httpsAgent?.destroy === "function") {
this.externalAgent = true;
return httpsAgent;
}
return new hsAgent({ keepAlive, maxSockets, ...httpsAgent });
})(),
logger: console,
};
}
destroy() {
this.config?.httpAgent?.destroy();
this.config?.httpsAgent?.destroy();
}
async handle(request, { abortSignal, requestTimeout } = {}) {
if (!this.config) {
this.config = await this.configProvider;
}
return new Promise((_resolve, _reject) => {
const config = this.config;
let writeRequestBodyPromise = undefined;
const timeouts = [];
const resolve = async (arg) => {
await writeRequestBodyPromise;
timeouts.forEach(timing.clearTimeout);
_resolve(arg);
};
const reject = async (arg) => {
await writeRequestBodyPromise;
timeouts.forEach(timing.clearTimeout);
_reject(arg);
};
if (abortSignal?.aborted) {
const abortError = new Error("Request aborted");
abortError.name = "AbortError";
reject(abortError);
return;
}
const isSSL = request.protocol === "https:";
const headers = request.headers ?? {};
const expectContinue = (headers.Expect ?? headers.expect) === "100-continue";
let agent = isSSL ? config.httpsAgent : config.httpAgent;
if (expectContinue && !this.externalAgent) {
agent = new (isSSL ? hsAgent : hAgent)({
keepAlive: false,
maxSockets: Infinity,
});
}
timeouts.push(timing.setTimeout(() => {
this.socketWarningTimestamp = NodeHttpHandler.checkSocketUsage(agent, this.socketWarningTimestamp, config.logger);
}, config.socketAcquisitionWarningTimeout ?? (config.requestTimeout ?? 2000) + (config.connectionTimeout ?? 1000)));
const queryString = buildQueryString(request.query || {});
let auth = undefined;
if (request.username != null || request.password != null) {
const username = request.username ?? "";
const password = request.password ?? "";
auth = `${username}:${password}`;
}
let path = request.path;
if (queryString) {
path += `?${queryString}`;
}
if (request.fragment) {
path += `#${request.fragment}`;
}
let hostname = request.hostname ?? "";
if (hostname[0] === "[" && hostname.endsWith("]")) {
hostname = request.hostname.slice(1, -1);
}
else {
hostname = request.hostname;
}
const nodeHttpsOptions = {
headers: request.headers,
host: hostname,
method: request.method,
path,
port: request.port,
agent,
auth,
};
const requestFunc = isSSL ? hsRequest : hRequest;
const req = requestFunc(nodeHttpsOptions, (res) => {
const httpResponse = new HttpResponse({
statusCode: res.statusCode || -1,
reason: res.statusMessage,
headers: getTransformedHeaders(res.headers),
body: res,
});
resolve({ response: httpResponse });
});
req.on("error", (err) => {
if (NODEJS_TIMEOUT_ERROR_CODES.includes(err.code)) {
reject(Object.assign(err, { name: "TimeoutError" }));
}
else {
reject(err);
}
});
if (abortSignal) {
const onAbort = () => {
req.destroy();
const abortError = new Error("Request aborted");
abortError.name = "AbortError";
reject(abortError);
};
if (typeof abortSignal.addEventListener === "function") {
const signal = abortSignal;
signal.addEventListener("abort", onAbort, { once: true });
req.once("close", () => signal.removeEventListener("abort", onAbort));
}
else {
abortSignal.onabort = onAbort;
}
}
const effectiveRequestTimeout = requestTimeout ?? config.requestTimeout;
timeouts.push(setConnectionTimeout(req, reject, config.connectionTimeout));
timeouts.push(setRequestTimeout(req, reject, effectiveRequestTimeout, config.throwOnRequestTimeout, config.logger ?? console));
timeouts.push(setSocketTimeout(req, reject, config.socketTimeout));
const httpAgent = nodeHttpsOptions.agent;
if (typeof httpAgent === "object" && "keepAlive" in httpAgent) {
timeouts.push(setSocketKeepAlive(req, {
keepAlive: httpAgent.keepAlive,
keepAliveMsecs: httpAgent.keepAliveMsecs,
}));
}
writeRequestBodyPromise = writeRequestBody(req, request, effectiveRequestTimeout, this.externalAgent).catch((e) => {
timeouts.forEach(timing.clearTimeout);
return _reject(e);
});
});
}
updateHttpClientConfig(key, value) {
this.config = undefined;
this.configProvider = this.configProvider.then((config) => {
return {
...config,
[key]: value,
};
});
}
httpHandlerConfigs() {
return this.config ?? {};
}
}

View File

@@ -0,0 +1,87 @@
import http2 from "http2";
import { NodeHttp2ConnectionPool } from "./node-http2-connection-pool";
export class NodeHttp2ConnectionManager {
constructor(config) {
this.config = config;
if (this.config.maxConcurrency && this.config.maxConcurrency <= 0) {
throw new RangeError("maxConcurrency must be greater than zero.");
}
}
config;
sessionCache = new Map();
lease(requestContext, connectionConfiguration) {
const url = this.getUrlString(requestContext);
const existingPool = this.sessionCache.get(url);
if (existingPool) {
const existingSession = existingPool.poll();
if (existingSession && !this.config.disableConcurrency) {
return existingSession;
}
}
const session = http2.connect(url);
if (this.config.maxConcurrency) {
session.settings({ maxConcurrentStreams: this.config.maxConcurrency }, (err) => {
if (err) {
throw new Error("Fail to set maxConcurrentStreams to " +
this.config.maxConcurrency +
"when creating new session for " +
requestContext.destination.toString());
}
});
}
session.unref();
const destroySessionCb = () => {
session.destroy();
this.deleteSession(url, session);
};
session.on("goaway", destroySessionCb);
session.on("error", destroySessionCb);
session.on("frameError", destroySessionCb);
session.on("close", () => this.deleteSession(url, session));
if (connectionConfiguration.requestTimeout) {
session.setTimeout(connectionConfiguration.requestTimeout, destroySessionCb);
}
const connectionPool = this.sessionCache.get(url) || new NodeHttp2ConnectionPool();
connectionPool.offerLast(session);
this.sessionCache.set(url, connectionPool);
return session;
}
deleteSession(authority, session) {
const existingConnectionPool = this.sessionCache.get(authority);
if (!existingConnectionPool) {
return;
}
if (!existingConnectionPool.contains(session)) {
return;
}
existingConnectionPool.remove(session);
this.sessionCache.set(authority, existingConnectionPool);
}
release(requestContext, session) {
const cacheKey = this.getUrlString(requestContext);
this.sessionCache.get(cacheKey)?.offerLast(session);
}
destroy() {
for (const [key, connectionPool] of this.sessionCache) {
for (const session of connectionPool) {
if (!session.destroyed) {
session.destroy();
}
connectionPool.remove(session);
}
this.sessionCache.delete(key);
}
}
setMaxConcurrentStreams(maxConcurrentStreams) {
if (maxConcurrentStreams && maxConcurrentStreams <= 0) {
throw new RangeError("maxConcurrentStreams must be greater than zero.");
}
this.config.maxConcurrency = maxConcurrentStreams;
}
setDisableConcurrentStreams(disableConcurrentStreams) {
this.config.disableConcurrency = disableConcurrentStreams;
}
getUrlString(request) {
return request.destination.toString();
}
}

View File

@@ -0,0 +1,32 @@
export class NodeHttp2ConnectionPool {
sessions = [];
constructor(sessions) {
this.sessions = sessions ?? [];
}
poll() {
if (this.sessions.length > 0) {
return this.sessions.shift();
}
}
offerLast(session) {
this.sessions.push(session);
}
contains(session) {
return this.sessions.includes(session);
}
remove(session) {
this.sessions = this.sessions.filter((s) => s !== session);
}
[Symbol.iterator]() {
return this.sessions[Symbol.iterator]();
}
destroy(connection) {
for (const session of this.sessions) {
if (session === connection) {
if (!session.destroyed) {
session.destroy();
}
}
}
}
}

View File

@@ -0,0 +1,170 @@
import { HttpResponse } from "@smithy/protocol-http";
import { buildQueryString } from "@smithy/querystring-builder";
import { constants } from "http2";
import { getTransformedHeaders } from "./get-transformed-headers";
import { NodeHttp2ConnectionManager } from "./node-http2-connection-manager";
import { writeRequestBody } from "./write-request-body";
export class NodeHttp2Handler {
config;
configProvider;
metadata = { handlerProtocol: "h2" };
connectionManager = new NodeHttp2ConnectionManager({});
static create(instanceOrOptions) {
if (typeof instanceOrOptions?.handle === "function") {
return instanceOrOptions;
}
return new NodeHttp2Handler(instanceOrOptions);
}
constructor(options) {
this.configProvider = new Promise((resolve, reject) => {
if (typeof options === "function") {
options()
.then((opts) => {
resolve(opts || {});
})
.catch(reject);
}
else {
resolve(options || {});
}
});
}
destroy() {
this.connectionManager.destroy();
}
async handle(request, { abortSignal, requestTimeout } = {}) {
if (!this.config) {
this.config = await this.configProvider;
this.connectionManager.setDisableConcurrentStreams(this.config.disableConcurrentStreams || false);
if (this.config.maxConcurrentStreams) {
this.connectionManager.setMaxConcurrentStreams(this.config.maxConcurrentStreams);
}
}
const { requestTimeout: configRequestTimeout, disableConcurrentStreams } = this.config;
const effectiveRequestTimeout = requestTimeout ?? configRequestTimeout;
return new Promise((_resolve, _reject) => {
let fulfilled = false;
let writeRequestBodyPromise = undefined;
const resolve = async (arg) => {
await writeRequestBodyPromise;
_resolve(arg);
};
const reject = async (arg) => {
await writeRequestBodyPromise;
_reject(arg);
};
if (abortSignal?.aborted) {
fulfilled = true;
const abortError = new Error("Request aborted");
abortError.name = "AbortError";
reject(abortError);
return;
}
const { hostname, method, port, protocol, query } = request;
let auth = "";
if (request.username != null || request.password != null) {
const username = request.username ?? "";
const password = request.password ?? "";
auth = `${username}:${password}@`;
}
const authority = `${protocol}//${auth}${hostname}${port ? `:${port}` : ""}`;
const requestContext = { destination: new URL(authority) };
const session = this.connectionManager.lease(requestContext, {
requestTimeout: this.config?.sessionTimeout,
disableConcurrentStreams: disableConcurrentStreams || false,
});
const rejectWithDestroy = (err) => {
if (disableConcurrentStreams) {
this.destroySession(session);
}
fulfilled = true;
reject(err);
};
const queryString = buildQueryString(query || {});
let path = request.path;
if (queryString) {
path += `?${queryString}`;
}
if (request.fragment) {
path += `#${request.fragment}`;
}
const req = session.request({
...request.headers,
[constants.HTTP2_HEADER_PATH]: path,
[constants.HTTP2_HEADER_METHOD]: method,
});
session.ref();
req.on("response", (headers) => {
const httpResponse = new HttpResponse({
statusCode: headers[":status"] || -1,
headers: getTransformedHeaders(headers),
body: req,
});
fulfilled = true;
resolve({ response: httpResponse });
if (disableConcurrentStreams) {
session.close();
this.connectionManager.deleteSession(authority, session);
}
});
if (effectiveRequestTimeout) {
req.setTimeout(effectiveRequestTimeout, () => {
req.close();
const timeoutError = new Error(`Stream timed out because of no activity for ${effectiveRequestTimeout} ms`);
timeoutError.name = "TimeoutError";
rejectWithDestroy(timeoutError);
});
}
if (abortSignal) {
const onAbort = () => {
req.close();
const abortError = new Error("Request aborted");
abortError.name = "AbortError";
rejectWithDestroy(abortError);
};
if (typeof abortSignal.addEventListener === "function") {
const signal = abortSignal;
signal.addEventListener("abort", onAbort, { once: true });
req.once("close", () => signal.removeEventListener("abort", onAbort));
}
else {
abortSignal.onabort = onAbort;
}
}
req.on("frameError", (type, code, id) => {
rejectWithDestroy(new Error(`Frame type id ${type} in stream id ${id} has failed with code ${code}.`));
});
req.on("error", rejectWithDestroy);
req.on("aborted", () => {
rejectWithDestroy(new Error(`HTTP/2 stream is abnormally aborted in mid-communication with result code ${req.rstCode}.`));
});
req.on("close", () => {
session.unref();
if (disableConcurrentStreams) {
session.destroy();
}
if (!fulfilled) {
rejectWithDestroy(new Error("Unexpected error: http2 request did not get a response"));
}
});
writeRequestBodyPromise = writeRequestBody(req, request, effectiveRequestTimeout);
});
}
updateHttpClientConfig(key, value) {
this.config = undefined;
this.configProvider = this.configProvider.then((config) => {
return {
...config,
[key]: value,
};
});
}
httpHandlerConfigs() {
return this.config ?? {};
}
destroySession(session) {
if (!session.destroyed) {
session.destroy();
}
}
}

View File

@@ -0,0 +1,21 @@
import { Readable } from "stream";
export class ReadFromBuffers extends Readable {
buffersToRead;
numBuffersRead = 0;
errorAfter;
constructor(options) {
super(options);
this.buffersToRead = options.buffers;
this.errorAfter = typeof options.errorAfter === "number" ? options.errorAfter : -1;
}
_read() {
if (this.errorAfter !== -1 && this.errorAfter === this.numBuffersRead) {
this.emit("error", new Error("Mock Error"));
return;
}
if (this.numBuffersRead >= this.buffersToRead.length) {
return this.push(null);
}
return this.push(this.buffersToRead[this.numBuffersRead++]);
}
}

View File

@@ -0,0 +1,88 @@
import { readFileSync } from "fs";
import { createServer as createHttpServer } from "http";
import { createServer as createHttp2Server } from "http2";
import { createServer as createHttpsServer } from "https";
import { join } from "path";
import { Readable } from "stream";
import { timing } from "./timing";
const fixturesDir = join(__dirname, "..", "fixtures");
const setResponseHeaders = (response, headers) => {
for (const [key, value] of Object.entries(headers)) {
response.setHeader(key, value);
}
};
const setResponseBody = (response, body) => {
if (body instanceof Readable) {
body.pipe(response);
}
else {
response.end(body);
}
};
export const createResponseFunction = (httpResp) => (request, response) => {
response.statusCode = httpResp.statusCode;
if (httpResp.reason) {
response.statusMessage = httpResp.reason;
}
setResponseHeaders(response, httpResp.headers);
setResponseBody(response, httpResp.body);
};
export const createResponseFunctionWithDelay = (httpResp, delay) => (request, response) => {
response.statusCode = httpResp.statusCode;
if (httpResp.reason) {
response.statusMessage = httpResp.reason;
}
setResponseHeaders(response, httpResp.headers);
timing.setTimeout(() => setResponseBody(response, httpResp.body), delay);
};
export const createContinueResponseFunction = (httpResp) => (request, response) => {
response.writeContinue();
timing.setTimeout(() => {
createResponseFunction(httpResp)(request, response);
}, 100);
};
export const createMockHttpsServer = () => {
const server = createHttpsServer({
key: readFileSync(join(fixturesDir, "test-server-key.pem")),
cert: readFileSync(join(fixturesDir, "test-server-cert.pem")),
});
return server;
};
export const createMockHttpServer = () => {
const server = createHttpServer();
return server;
};
export const createMockHttp2Server = () => {
const server = createHttp2Server();
return server;
};
export const createMirrorResponseFunction = (httpResp) => (request, response) => {
const bufs = [];
request.on("data", (chunk) => {
bufs.push(chunk);
});
request.on("end", () => {
response.statusCode = httpResp.statusCode;
setResponseHeaders(response, httpResp.headers);
setResponseBody(response, Buffer.concat(bufs));
});
request.on("error", (err) => {
response.statusCode = 500;
setResponseHeaders(response, httpResp.headers);
setResponseBody(response, err.message);
});
};
export const getResponseBody = (response) => {
return new Promise((resolve, reject) => {
const bufs = [];
response.body.on("data", function (d) {
bufs.push(d);
});
response.body.on("end", function () {
resolve(Buffer.concat(bufs).toString());
});
response.body.on("error", (err) => {
reject(err);
});
});
};

View File

@@ -0,0 +1,36 @@
import { timing } from "./timing";
const DEFER_EVENT_LISTENER_TIME = 1000;
export const setConnectionTimeout = (request, reject, timeoutInMs = 0) => {
if (!timeoutInMs) {
return -1;
}
const registerTimeout = (offset) => {
const timeoutId = timing.setTimeout(() => {
request.destroy();
reject(Object.assign(new Error(`@smithy/node-http-handler - the request socket did not establish a connection with the server within the configured timeout of ${timeoutInMs} ms.`), {
name: "TimeoutError",
}));
}, timeoutInMs - offset);
const doWithSocket = (socket) => {
if (socket?.connecting) {
socket.on("connect", () => {
timing.clearTimeout(timeoutId);
});
}
else {
timing.clearTimeout(timeoutId);
}
};
if (request.socket) {
doWithSocket(request.socket);
}
else {
request.on("socket", doWithSocket);
}
};
if (timeoutInMs < 2000) {
registerTimeout(0);
return 0;
}
return timing.setTimeout(registerTimeout.bind(null, DEFER_EVENT_LISTENER_TIME), DEFER_EVENT_LISTENER_TIME);
};

View File

@@ -0,0 +1,21 @@
import { timing } from "./timing";
export const setRequestTimeout = (req, reject, timeoutInMs = 0, throwOnRequestTimeout, logger) => {
if (timeoutInMs) {
return timing.setTimeout(() => {
let msg = `@smithy/node-http-handler - [${throwOnRequestTimeout ? "ERROR" : "WARN"}] a request has exceeded the configured ${timeoutInMs} ms requestTimeout.`;
if (throwOnRequestTimeout) {
const error = Object.assign(new Error(msg), {
name: "TimeoutError",
code: "ETIMEDOUT",
});
req.destroy(error);
reject(error);
}
else {
msg += ` Init client requestHandler with throwOnRequestTimeout=true to turn this into an error.`;
logger?.warn?.(msg);
}
}, timeoutInMs);
}
return -1;
};

View File

@@ -0,0 +1,22 @@
import { timing } from "./timing";
const DEFER_EVENT_LISTENER_TIME = 3000;
export const setSocketKeepAlive = (request, { keepAlive, keepAliveMsecs }, deferTimeMs = DEFER_EVENT_LISTENER_TIME) => {
if (keepAlive !== true) {
return -1;
}
const registerListener = () => {
if (request.socket) {
request.socket.setKeepAlive(keepAlive, keepAliveMsecs || 0);
}
else {
request.on("socket", (socket) => {
socket.setKeepAlive(keepAlive, keepAliveMsecs || 0);
});
}
};
if (deferTimeMs === 0) {
registerListener();
return 0;
}
return timing.setTimeout(registerListener, deferTimeMs);
};

View File

@@ -0,0 +1,23 @@
import { timing } from "./timing";
const DEFER_EVENT_LISTENER_TIME = 3000;
export const setSocketTimeout = (request, reject, timeoutInMs = 0) => {
const registerTimeout = (offset) => {
const timeout = timeoutInMs - offset;
const onTimeout = () => {
request.destroy();
reject(Object.assign(new Error(`@smithy/node-http-handler - the request socket timed out after ${timeoutInMs} ms of inactivity (configured by client requestHandler).`), { name: "TimeoutError" }));
};
if (request.socket) {
request.socket.setTimeout(timeout, onTimeout);
request.on("close", () => request.socket?.removeListener("timeout", onTimeout));
}
else {
request.setTimeout(timeout, onTimeout);
}
};
if (0 < timeoutInMs && timeoutInMs < 6000) {
registerTimeout(0);
return 0;
}
return timing.setTimeout(registerTimeout.bind(null, timeoutInMs === 0 ? 0 : DEFER_EVENT_LISTENER_TIME), DEFER_EVENT_LISTENER_TIME);
};

View File

@@ -0,0 +1,8 @@
import { Writable } from "stream";
export class Collector extends Writable {
bufferedBytes = [];
_write(chunk, encoding, callback) {
this.bufferedBytes.push(chunk);
callback();
}
}

View File

@@ -0,0 +1,41 @@
import { Collector } from "./collector";
export const streamCollector = (stream) => {
if (isReadableStreamInstance(stream)) {
return collectReadableStream(stream);
}
return new Promise((resolve, reject) => {
const collector = new Collector();
stream.pipe(collector);
stream.on("error", (err) => {
collector.end();
reject(err);
});
collector.on("error", reject);
collector.on("finish", function () {
const bytes = new Uint8Array(Buffer.concat(this.bufferedBytes));
resolve(bytes);
});
});
};
const isReadableStreamInstance = (stream) => typeof ReadableStream === "function" && stream instanceof ReadableStream;
async function collectReadableStream(stream) {
const chunks = [];
const reader = stream.getReader();
let isDone = false;
let length = 0;
while (!isDone) {
const { done, value } = await reader.read();
if (value) {
chunks.push(value);
length += value.length;
}
isDone = done;
}
const collected = new Uint8Array(length);
let offset = 0;
for (const chunk of chunks) {
collected.set(chunk, offset);
offset += chunk.length;
}
return collected;
}

View File

@@ -0,0 +1,21 @@
import { Readable } from "stream";
export class ReadFromBuffers extends Readable {
buffersToRead;
numBuffersRead = 0;
errorAfter;
constructor(options) {
super(options);
this.buffersToRead = options.buffers;
this.errorAfter = typeof options.errorAfter === "number" ? options.errorAfter : -1;
}
_read(size) {
if (this.errorAfter !== -1 && this.errorAfter === this.numBuffersRead) {
this.emit("error", new Error("Mock Error"));
return;
}
if (this.numBuffersRead >= this.buffersToRead.length) {
return this.push(null);
}
return this.push(this.buffersToRead[this.numBuffersRead++]);
}
}

View File

@@ -0,0 +1,4 @@
export const timing = {
setTimeout: (cb, ms) => setTimeout(cb, ms),
clearTimeout: (timeoutId) => clearTimeout(timeoutId),
};

View File

@@ -0,0 +1,56 @@
import { Readable } from "stream";
import { timing } from "./timing";
const MIN_WAIT_TIME = 6_000;
export async function writeRequestBody(httpRequest, request, maxContinueTimeoutMs = MIN_WAIT_TIME, externalAgent = false) {
const headers = request.headers ?? {};
const expect = headers.Expect || headers.expect;
let timeoutId = -1;
let sendBody = true;
if (!externalAgent && expect === "100-continue") {
sendBody = await Promise.race([
new Promise((resolve) => {
timeoutId = Number(timing.setTimeout(() => resolve(true), Math.max(MIN_WAIT_TIME, maxContinueTimeoutMs)));
}),
new Promise((resolve) => {
httpRequest.on("continue", () => {
timing.clearTimeout(timeoutId);
resolve(true);
});
httpRequest.on("response", () => {
timing.clearTimeout(timeoutId);
resolve(false);
});
httpRequest.on("error", () => {
timing.clearTimeout(timeoutId);
resolve(false);
});
}),
]);
}
if (sendBody) {
writeBody(httpRequest, request.body);
}
}
function writeBody(httpRequest, body) {
if (body instanceof Readable) {
body.pipe(httpRequest);
return;
}
if (body) {
if (Buffer.isBuffer(body) || typeof body === "string") {
httpRequest.end(body);
return;
}
const uint8 = body;
if (typeof uint8 === "object" &&
uint8.buffer &&
typeof uint8.byteOffset === "number" &&
typeof uint8.byteLength === "number") {
httpRequest.end(Buffer.from(uint8.buffer, uint8.byteOffset, uint8.byteLength));
return;
}
httpRequest.end(Buffer.from(body));
return;
}
httpRequest.end();
}