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

feat(core): Add heartbeat grace/timeout options, client retry/wait-for-ready, server readiness and socket cleanup, transport socket options, helper utilities, and tests

Browse Source
This commit is contained in:
Jürgen Kunz
2025-08-25 13:37:31 +00:00
parent e3c1d35895
commit dd25ffd3e4
9 changed files with 780 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ts/00_commitinfo_data.ts
View File

@@ -3,6 +3,6 @@
*/
export const commitinfo = {
name: '@push.rocks/smartipc',
version: '2.0.3',
version: '2.1.0',
description: 'A library for node inter process communication, providing an easy-to-use API for IPC.'
}

26
ts/classes.ipcchannel.ts
View File

@@ -22,6 +22,10 @@ export interface IIpcChannelOptions extends IIpcTransportOptions {
heartbeatInterval?: number;
/** Heartbeat timeout in ms */
heartbeatTimeout?: number;
/** Initial grace period before heartbeat timeout in ms */
heartbeatInitialGracePeriodMs?: number;
/** Throw on heartbeat timeout (default: true, set false to emit event instead) */
heartbeatThrowOnTimeout?: boolean;
}
/**
@@ -46,6 +50,7 @@ export class IpcChannel<TRequest = any, TResponse = any> extends plugins.EventEm
private heartbeatTimer?: NodeJS.Timeout;
private heartbeatCheckTimer?: NodeJS.Timeout;
private lastHeartbeat: number = Date.now();
private connectionStartTime: number = Date.now();
private isReconnecting = false;
private isClosing = false;
@@ -203,6 +208,7 @@ export class IpcChannel<TRequest = any, TResponse = any> extends plugins.EventEm
this.stopHeartbeat();
this.lastHeartbeat = Date.now();
this.connectionStartTime = Date.now();
// Send heartbeat messages
this.heartbeatTimer = setInterval(() => {
@@ -214,9 +220,25 @@ export class IpcChannel<TRequest = any, TResponse = any> extends plugins.EventEm
// Check for heartbeat timeout
this.heartbeatCheckTimer = setInterval(() => {
const timeSinceLastHeartbeat = Date.now() - this.lastHeartbeat;
const timeSinceConnection = Date.now() - this.connectionStartTime;
const gracePeriod = this.options.heartbeatInitialGracePeriodMs || 0;
// Skip timeout check during initial grace period
if (timeSinceConnection < gracePeriod) {
return;
}
if (timeSinceLastHeartbeat > this.options.heartbeatTimeout!) {
this.emit('error', new Error('Heartbeat timeout'));
this.transport.disconnect().catch(() => {});
const error = new Error('Heartbeat timeout');
if (this.options.heartbeatThrowOnTimeout !== false) {
// Default behavior: emit error which may cause disconnect
this.emit('error', error);
this.transport.disconnect().catch(() => {});
} else {
// Emit heartbeatTimeout event instead of error
this.emit('heartbeatTimeout', error);
}
}
}, this.options.heartbeatTimeout! / 2);
}

141
ts/classes.ipcclient.ts
View File

@@ -5,11 +5,35 @@ import type { IIpcChannelOptions } from './classes.ipcchannel.js';
/**
* Options for IPC Client
*/
export interface IConnectRetryConfig {
/** Enable connection retry */
enabled: boolean;
/** Initial delay before first retry in ms */
initialDelay?: number;
/** Maximum delay between retries in ms */
maxDelay?: number;
/** Maximum number of attempts */
maxAttempts?: number;
/** Total timeout for all retry attempts in ms */
totalTimeout?: number;
}
export interface IClientConnectOptions {
/** Wait for server to be ready before attempting connection */
waitForReady?: boolean;
/** Timeout for waiting for server readiness in ms */
waitTimeout?: number;
}
export interface IIpcClientOptions extends IIpcChannelOptions {
/** Client identifier */
clientId?: string;
/** Client metadata */
metadata?: Record<string, any>;
/** Connection retry configuration */
connectRetry?: IConnectRetryConfig;
/** Registration timeout in ms (default: 5000) */
registerTimeoutMs?: number;
}
/**
@@ -35,34 +59,111 @@ export class IpcClient extends plugins.EventEmitter {
/**
* Connect to the server
*/
public async connect(): Promise<void> {
public async connect(connectOptions: IClientConnectOptions = {}): Promise<void> {
if (this.isConnected) {
return;
}
// Connect the channel
await this.channel.connect();
// Helper function to attempt registration
const attemptRegistration = async (): Promise<void> => {
const registerTimeoutMs = this.options.registerTimeoutMs || 5000;
try {
const response = await this.channel.request<any, any>(
'__register__',
{
clientId: this.clientId,
metadata: this.options.metadata
},
{ timeout: registerTimeoutMs }
);
// Register with the server
try {
const response = await this.channel.request<any, any>(
'__register__',
{
clientId: this.clientId,
metadata: this.options.metadata
},
{ timeout: 5000 }
);
if (!response.success) {
throw new Error(response.error || 'Registration failed');
}
if (!response.success) {
throw new Error(response.error || 'Registration failed');
this.isConnected = true;
this.emit('connect');
} catch (error) {
throw new Error(`Failed to register with server: ${error.message}`);
}
};
// Helper function to attempt connection with retry
const attemptConnection = async (): Promise<void> => {
const retryConfig = this.options.connectRetry;
const maxAttempts = retryConfig?.maxAttempts || 1;
const initialDelay = retryConfig?.initialDelay || 100;
const maxDelay = retryConfig?.maxDelay || 1500;
const totalTimeout = retryConfig?.totalTimeout || 15000;
const startTime = Date.now();
let lastError: Error | undefined;
let delay = initialDelay;
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
// Check total timeout
if (totalTimeout && Date.now() - startTime > totalTimeout) {
throw new Error(`Connection timeout after ${totalTimeout}ms: ${lastError?.message || 'Unknown error'}`);
}
try {
// Connect the channel
await this.channel.connect();
// Attempt registration
await attemptRegistration();
return; // Success!
} catch (error) {
lastError = error as Error;
// Disconnect channel for retry
await this.channel.disconnect().catch(() => {});
// If this isn't the last attempt and retry is enabled, wait before retrying
if (attempt < maxAttempts && retryConfig?.enabled) {
// Check if we have time for another attempt
if (totalTimeout && Date.now() - startTime + delay > totalTimeout) {
break; // Will timeout, don't wait
}
await new Promise(resolve => setTimeout(resolve, delay));
// Exponential backoff with max limit
delay = Math.min(delay * 2, maxDelay);
}
}
}
this.isConnected = true;
this.emit('connect');
} catch (error) {
await this.channel.disconnect();
throw new Error(`Failed to register with server: ${error.message}`);
// All attempts failed
throw lastError || new Error('Failed to connect to server');
};
// If waitForReady is specified, wait for server socket to exist first
if (connectOptions.waitForReady) {
const waitTimeout = connectOptions.waitTimeout || 10000;
const startTime = Date.now();
while (Date.now() - startTime < waitTimeout) {
try {
// Try to connect
await attemptConnection();
return; // Success!
} catch (error) {
// If it's a connection refused error, server might not be ready yet
if ((error as any).message?.includes('ECONNREFUSED') ||
(error as any).message?.includes('ENOENT')) {
await new Promise(resolve => setTimeout(resolve, 100));
continue;
}
// Other errors should be thrown
throw error;
}
}
throw new Error(`Server not ready after ${waitTimeout}ms`);
} else {
// Normal connection attempt
await attemptConnection();
}
}

31
ts/classes.ipcserver.ts
View File

@@ -5,11 +5,20 @@ import type { IIpcChannelOptions } from './classes.ipcchannel.js';
/**
* Options for IPC Server
*/
export interface IServerStartOptions {
/** When to consider server ready (default: 'socket-bound') */
readyWhen?: 'socket-bound' | 'accepting';
}
export interface IIpcServerOptions extends Omit<IIpcChannelOptions, 'autoReconnect' | 'reconnectDelay' | 'maxReconnectDelay' | 'reconnectMultiplier' | 'maxReconnectAttempts'> {
/** Maximum number of client connections */
maxClients?: number;
/** Client idle timeout in ms */
clientIdleTimeout?: number;
/** Automatically cleanup stale socket file on start (default: false) */
autoCleanupSocketFile?: boolean;
/** Socket file permissions mode (e.g. 0o600) */
socketMode?: number;
}
/**
@@ -32,6 +41,7 @@ export class IpcServer extends plugins.EventEmitter {
private messageHandlers = new Map<string, (payload: any, clientId: string) => any | Promise<any>>();
private primaryChannel?: IpcChannel;
private isRunning = false;
private isReady = false;
private clientIdleCheckTimer?: NodeJS.Timeout;
// Pub/sub tracking
@@ -50,7 +60,7 @@ export class IpcServer extends plugins.EventEmitter {
/**
* Start the server
*/
public async start(): Promise<void> {
public async start(options: IServerStartOptions = {}): Promise<void> {
if (this.isRunning) {
return;
}
@@ -196,6 +206,18 @@ export class IpcServer extends plugins.EventEmitter {
this.isRunning = true;
this.startClientIdleCheck();
this.emit('start');
// Handle readiness based on options
if (options.readyWhen === 'accepting') {
// Wait a bit to ensure handlers are fully set up
await new Promise(resolve => setTimeout(resolve, 10));
this.isReady = true;
this.emit('ready');
} else {
// Default: ready when socket is bound
this.isReady = true;
this.emit('ready');
}
}
/**
@@ -505,4 +527,11 @@ export class IpcServer extends plugins.EventEmitter {
uptime: this.primaryChannel ? Date.now() - (this.primaryChannel as any).connectedAt : undefined
};
}
/**
* Check if server is ready to accept connections
*/
public getIsReady(): boolean {
return this.isReady;
}
}

25
ts/classes.transports.ts
View File

@@ -34,6 +34,10 @@ export interface IIpcTransportOptions {
noDelay?: boolean;
/** Maximum message size in bytes (default: 8MB) */
maxMessageSize?: number;
/** Automatically cleanup stale socket file on start (default: false) */
autoCleanupSocketFile?: boolean;
/** Socket file permissions mode (e.g. 0o600) */
socketMode?: number;
}
/**
@@ -206,11 +210,13 @@ export class UnixSocketTransport extends IpcTransport {
*/
private async startServer(socketPath: string): Promise<void> {
return new Promise((resolve, reject) => {
// Clean up stale socket file if it exists
try {
plugins.fs.unlinkSync(socketPath);
} catch (error) {
// File doesn't exist, that's fine
// Clean up stale socket file if autoCleanupSocketFile is enabled
if (this.options.autoCleanupSocketFile) {
try {
plugins.fs.unlinkSync(socketPath);
} catch (error) {
// File doesn't exist, that's fine
}
}
this.server = plugins.net.createServer((socket) => {
@@ -247,6 +253,15 @@ export class UnixSocketTransport extends IpcTransport {
this.server.on('error', reject);
this.server.listen(socketPath, () => {
// Set socket permissions if specified
if (this.options.socketMode !== undefined && process.platform !== 'win32') {
try {
plugins.fs.chmodSync(socketPath, this.options.socketMode);
} catch (error) {
// Ignore permission errors, not critical
}
}
this.connected = true;
this.emit('connect');
resolve();

85
ts/index.ts
View File

@@ -7,7 +7,7 @@ import { IpcServer } from './classes.ipcserver.js';
import { IpcClient } from './classes.ipcclient.js';
import { IpcChannel } from './classes.ipcchannel.js';
import type { IIpcServerOptions } from './classes.ipcserver.js';
import type { IIpcClientOptions } from './classes.ipcclient.js';
import type { IIpcClientOptions, IConnectRetryConfig } from './classes.ipcclient.js';
import type { IIpcChannelOptions } from './classes.ipcchannel.js';
/**
@@ -17,6 +17,89 @@ export class SmartIpc {
/**
* Create an IPC server
*/
/**
* Wait for a server to become ready at the given socket path
*/
public static async waitForServer(options: {
socketPath: string;
timeoutMs?: number;
}): Promise<void> {
const timeout = options.timeoutMs || 10000;
const startTime = Date.now();
while (Date.now() - startTime < timeout) {
try {
// Try to connect as a temporary client
const testClient = new IpcClient({
id: `test-probe-${Date.now()}`,
socketPath: options.socketPath,
autoReconnect: false,
heartbeat: false
});
await testClient.connect();
await testClient.disconnect();
return; // Server is ready
} catch (error) {
// Server not ready yet, wait and retry
await new Promise(resolve => setTimeout(resolve, 100));
}
}
throw new Error(`Server not ready at ${options.socketPath} after ${timeout}ms`);
}
/**
* Helper to spawn a server process and connect a client
*/
public static async spawnAndConnect(options: {
serverScript: string;
socketPath: string;
clientId?: string;
spawnOptions?: any;
connectRetry?: IConnectRetryConfig;
timeoutMs?: number;
}): Promise<{
client: IpcClient;
serverProcess: any;
}> {
const { spawn } = await import('child_process');
// Spawn the server process
const serverProcess = spawn('node', [options.serverScript], {
detached: true,
stdio: 'pipe',
...options.spawnOptions
});
// Handle server process errors
serverProcess.on('error', (error: Error) => {
console.error('Server process error:', error);
});
// Wait for server to be ready
await SmartIpc.waitForServer({
socketPath: options.socketPath,
timeoutMs: options.timeoutMs || 10000
});
// Create and connect client
const client = new IpcClient({
id: options.clientId || 'test-client',
socketPath: options.socketPath,
connectRetry: options.connectRetry || {
enabled: true,
maxAttempts: 10,
initialDelay: 100,
maxDelay: 1000
}
});
await client.connect({ waitForReady: true });
return { client, serverProcess };
}
public static createServer(options: IIpcServerOptions): IpcServer {
return new IpcServer(options);
}