push.rocks/smartrequest
3
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(client): Fix CI configuration, prevent socket hangs with auto-drain, and apply various client/core TypeScript fixes and test updates

Browse Source
This commit is contained in:
Jürgen Kunz
2025-08-18 00:21:14 +00:00
parent 9b9c8fd618
commit ee750dea58
34 changed files with 2144 additions and 892 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ts/core_node/index.ts
View File

@@ -1,3 +1,3 @@
// Core exports
export * from './response.js';
export { CoreRequest } from './request.js';
export { CoreRequest } from './request.js';

2
ts/core_node/plugins.ts
View File

@@ -17,4 +17,4 @@ import { HttpAgent, HttpsAgent } from 'agentkeepalive';
const agentkeepalive = { HttpAgent, HttpsAgent };
import formData from 'form-data';
export { agentkeepalive, formData };
export { agentkeepalive, formData };

56
ts/core_node/request.ts
View File

@@ -29,21 +29,33 @@ const httpsAgentKeepAliveFalse = new plugins.agentkeepalive.HttpsAgent({
/**
* Node.js implementation of Core Request class that handles all HTTP/HTTPS requests
*/
export class CoreRequest extends AbstractCoreRequest<types.ICoreRequestOptions, CoreResponse> {
export class CoreRequest extends AbstractCoreRequest<
types.ICoreRequestOptions,
CoreResponse
> {
private requestDataFunc: ((req: plugins.http.ClientRequest) => void) | null;
constructor(
url: string,
options: types.ICoreRequestOptions = {},
requestDataFunc: ((req: plugins.http.ClientRequest) => void) | null = null
requestDataFunc: ((req: plugins.http.ClientRequest) => void) | null = null,
) {
super(url, options);
this.requestDataFunc = requestDataFunc;
// Check for unsupported fetch-specific options
if (options.credentials || options.mode || options.cache || options.redirect ||
options.referrer || options.referrerPolicy || options.integrity) {
throw new Error('Fetch API specific options (credentials, mode, cache, redirect, referrer, referrerPolicy, integrity) are not supported in Node.js implementation');
if (
options.credentials ||
options.mode ||
options.cache ||
options.redirect ||
options.referrer ||
options.referrerPolicy ||
options.integrity
) {
throw new Error(
'Fetch API specific options (credentials, mode, cache, redirect, referrer, referrerPolicy, integrity) are not supported in Node.js implementation',
);
}
}
@@ -65,7 +77,7 @@ export class CoreRequest extends AbstractCoreRequest<types.ICoreRequestOptions,
const parsedUrl = plugins.smarturl.Smarturl.createFromUrl(this.url, {
searchParams: this.options.queryParams || {},
});
this.options.hostname = parsedUrl.hostname;
if (parsedUrl.port) {
this.options.port = parseInt(parsedUrl.port, 10);
@@ -74,7 +86,9 @@ export class CoreRequest extends AbstractCoreRequest<types.ICoreRequestOptions,
// Handle unix socket URLs
if (CoreRequest.isUnixSocket(this.url)) {
const { socketPath, path } = CoreRequest.parseUnixSocketUrl(this.options.path);
const { socketPath, path } = CoreRequest.parseUnixSocketUrl(
this.options.path,
);
this.options.socketPath = socketPath;
this.options.path = path;
}
@@ -83,18 +97,25 @@ export class CoreRequest extends AbstractCoreRequest<types.ICoreRequestOptions,
if (!this.options.agent) {
// Only use keep-alive agents if explicitly requested
if (this.options.keepAlive === true) {
this.options.agent = parsedUrl.protocol === 'https:' ? httpsAgent : httpAgent;
this.options.agent =
parsedUrl.protocol === 'https:' ? httpsAgent : httpAgent;
} else if (this.options.keepAlive === false) {
this.options.agent = parsedUrl.protocol === 'https:' ? httpsAgentKeepAliveFalse : httpAgentKeepAliveFalse;
this.options.agent =
parsedUrl.protocol === 'https:'
? httpsAgentKeepAliveFalse
: httpAgentKeepAliveFalse;
}
// If keepAlive is undefined, don't set any agent (more fetch-like behavior)
}
// Determine request module
const requestModule = parsedUrl.protocol === 'https:' ? plugins.https : plugins.http;
const requestModule =
parsedUrl.protocol === 'https:' ? plugins.https : plugins.http;
if (!requestModule) {
throw new Error(`The request to ${this.url} is missing a viable protocol. Must be http or https`);
throw new Error(
`The request to ${this.url} is missing a viable protocol. Must be http or https`,
);
}
// Perform the request
@@ -119,11 +140,12 @@ export class CoreRequest extends AbstractCoreRequest<types.ICoreRequestOptions,
});
} else {
// Write body as-is - caller is responsible for serialization
const bodyData = typeof this.options.requestBody === 'string'
? this.options.requestBody
: this.options.requestBody instanceof Buffer
const bodyData =
typeof this.options.requestBody === 'string'
? this.options.requestBody
: JSON.stringify(this.options.requestBody); // Still stringify for backward compatibility
: this.options.requestBody instanceof Buffer
? this.options.requestBody
: JSON.stringify(this.options.requestBody); // Still stringify for backward compatibility
request.write(bodyData);
request.end();
}
@@ -155,7 +177,7 @@ export class CoreRequest extends AbstractCoreRequest<types.ICoreRequestOptions,
*/
static async create(
url: string,
options: types.ICoreRequestOptions = {}
options: types.ICoreRequestOptions = {},
): Promise<CoreResponse> {
const request = new CoreRequest(url, options);
return request.fire();

51
ts/core_node/response.ts
View File

@@ -5,7 +5,10 @@ import { CoreResponse as AbstractCoreResponse } from '../core_base/response.js';
/**
* Node.js implementation of Core Response class that provides a fetch-like API
*/
export class CoreResponse<T = any> extends AbstractCoreResponse<T> implements types.INodeResponse<T> {
export class CoreResponse<T = any>
extends AbstractCoreResponse<T>
implements types.INodeResponse<T>
{
private incomingMessage: plugins.http.IncomingMessage;
private bodyBufferPromise: Promise<Buffer> | null = null;
private _autoDrainTimeout: NodeJS.Immediate | null = null;
@@ -17,7 +20,11 @@ export class CoreResponse<T = any> extends AbstractCoreResponse<T> implements ty
public readonly headers: plugins.http.IncomingHttpHeaders;
public readonly url: string;
constructor(incomingMessage: plugins.http.IncomingMessage, url: string, options: types.ICoreRequestOptions = {}) {
constructor(
incomingMessage: plugins.http.IncomingMessage,
url: string,
options: types.ICoreRequestOptions = {},
) {
super();
this.incomingMessage = incomingMessage;
this.url = url;
@@ -25,14 +32,16 @@ export class CoreResponse<T = any> extends AbstractCoreResponse<T> implements ty
this.statusText = incomingMessage.statusMessage || '';
this.ok = this.status >= 200 && this.status < 300;
this.headers = incomingMessage.headers;
// Auto-drain unconsumed streams to prevent socket hanging
// This prevents keep-alive sockets from timing out when response bodies aren't consumed
// Default to true if not specified
if (options.autoDrain !== false) {
this._autoDrainTimeout = setImmediate(() => {
if (!this.consumed && !this.incomingMessage.readableEnded) {
console.log(`Auto-draining unconsumed response body for ${this.url} (status: ${this.status})`);
console.log(
`Auto-draining unconsumed response body for ${this.url} (status: ${this.status})`,
);
this.incomingMessage.resume(); // Drain without processing
}
});
@@ -48,7 +57,7 @@ export class CoreResponse<T = any> extends AbstractCoreResponse<T> implements ty
clearImmediate(this._autoDrainTimeout);
this._autoDrainTimeout = null;
}
super.ensureNotConsumed();
}
@@ -57,22 +66,22 @@ export class CoreResponse<T = any> extends AbstractCoreResponse<T> implements ty
*/
private async collectBody(): Promise<Buffer> {
this.ensureNotConsumed();
if (this.bodyBufferPromise) {
return this.bodyBufferPromise;
}
this.bodyBufferPromise = new Promise<Buffer>((resolve, reject) => {
const chunks: Buffer[] = [];
this.incomingMessage.on('data', (chunk: Buffer) => {
chunks.push(chunk);
});
this.incomingMessage.on('end', () => {
resolve(Buffer.concat(chunks));
});
this.incomingMessage.on('error', reject);
});
@@ -85,7 +94,7 @@ export class CoreResponse<T = any> extends AbstractCoreResponse<T> implements ty
async json(): Promise<T> {
const buffer = await this.collectBody();
const text = buffer.toString('utf-8');
try {
return JSON.parse(text);
} catch (error) {
@@ -106,7 +115,10 @@ export class CoreResponse<T = any> extends AbstractCoreResponse<T> implements ty
*/
async arrayBuffer(): Promise<ArrayBuffer> {
const buffer = await this.collectBody();
return buffer.buffer.slice(buffer.byteOffset, buffer.byteOffset + buffer.byteLength);
return buffer.buffer.slice(
buffer.byteOffset,
buffer.byteOffset + buffer.byteLength,
);
}
/**
@@ -114,13 +126,13 @@ export class CoreResponse<T = any> extends AbstractCoreResponse<T> implements ty
*/
stream(): ReadableStream<Uint8Array> | null {
this.ensureNotConsumed();
// Convert Node.js stream to web stream
// In Node.js 16.5+ we can use Readable.toWeb()
if (this.incomingMessage.readableEnded || this.incomingMessage.destroyed) {
return null;
}
// Create a web ReadableStream from the Node.js stream
const nodeStream = this.incomingMessage;
return new ReadableStream<Uint8Array>({
@@ -128,22 +140,22 @@ export class CoreResponse<T = any> extends AbstractCoreResponse<T> implements ty
nodeStream.on('data', (chunk) => {
controller.enqueue(new Uint8Array(chunk));
});
nodeStream.on('end', () => {
controller.close();
});
nodeStream.on('error', (err) => {
controller.error(err);
});
},
cancel() {
nodeStream.destroy();
}
},
});
}
/**
* Get response as a Node.js readable stream
*/
@@ -158,5 +170,4 @@ export class CoreResponse<T = any> extends AbstractCoreResponse<T> implements ty
raw(): plugins.http.IncomingMessage {
return this.incomingMessage;
}
}
}

7
ts/core_node/types.ts
View File

@@ -7,7 +7,8 @@ export * from '../core_base/types.js';
/**
* Extended IncomingMessage with body property (legacy compatibility)
*/
export interface IExtendedIncomingMessage<T = any> extends plugins.http.IncomingMessage {
export interface IExtendedIncomingMessage<T = any>
extends plugins.http.IncomingMessage {
body: T;
}
@@ -17,7 +18,7 @@ export interface IExtendedIncomingMessage<T = any> extends plugins.http.Incoming
export interface INodeResponse<T = any> extends baseTypes.ICoreResponse<T> {
// Node.js specific methods
streamNode(): NodeJS.ReadableStream; // Returns Node.js style stream
// Legacy compatibility
raw(): plugins.http.IncomingMessage;
}
}