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

Refactor smartsocket implementation for improved WebSocket handling and message protocol

Browse Source 
- Updated test files to use new testing library and reduced test cycles for efficiency.
- Removed dependency on smartexpress and integrated direct WebSocket handling.
- Enhanced Smartsocket and SmartsocketClient classes to support new message types and authentication flow.
- Implemented a new message interface for structured communication between client and server.
- Added external server support for smartserve with appropriate WebSocket hooks.
- Improved connection management and error handling in SocketConnection and SocketRequest classes.
- Cleaned up code and removed deprecated socket.io references in favor of native WebSocket.
This commit is contained in:
Jürgen Kunz
2025-12-03 02:20:38 +00:00
parent ee59471e14
commit 1d62c9c695
14 changed files with 3901 additions and 3007 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
ts/interfaces/index.ts
View File

@@ -1,2 +1,3 @@
export * from './connection.js';
export * from './tag.js';
export * from './message.js';

67
ts/interfaces/message.ts Normal file
View File

@@ -0,0 +1,67 @@
/**
* Message types for the smartsocket protocol
*/
export type TMessageType =
| 'authRequest' // Server requests authentication from client
| 'auth' // Client provides authentication data
| 'authResponse' // Server responds to authentication
| 'serverReady' // Server signals it's fully ready
| 'function' // Function call request
| 'functionResponse' // Function call response
| 'tagUpdate'; // Tag store synchronization
/**
* Base message interface for all smartsocket messages
*/
export interface ISocketMessage<T = any> {
type: TMessageType;
id?: string; // For request/response correlation
payload: T;
}
/**
* Authentication request payload (server -> client)
*/
export interface IAuthRequestPayload {
serverAlias: string;
}
/**
* Authentication data payload (client -> server)
*/
export interface IAuthPayload {
alias: string;
}
/**
* Authentication response payload (server -> client)
*/
export interface IAuthResponsePayload {
success: boolean;
error?: string;
}
/**
* Function call payload
*/
export interface IFunctionCallPayload {
funcName: string;
funcData: any;
}
/**
* Tag update payload
*/
export interface ITagUpdatePayload {
tags: { [key: string]: any };
}
/**
* Helper type for creating typed messages
*/
export type TAuthRequestMessage = ISocketMessage<IAuthRequestPayload>;
export type TAuthMessage = ISocketMessage<IAuthPayload>;
export type TAuthResponseMessage = ISocketMessage<IAuthResponsePayload>;
export type TFunctionMessage = ISocketMessage<IFunctionCallPayload>;
export type TFunctionResponseMessage = ISocketMessage<IFunctionCallPayload>;
export type TTagUpdateMessage = ISocketMessage<ITagUpdatePayload>;

90
ts/smartsocket.classes.smartsocket.ts
View File

@@ -26,7 +26,6 @@ export class Smartsocket {
public alias: string;
public smartenv = new plugins.smartenv.Smartenv();
public options: ISmartsocketConstructorOptions;
public io: pluginsTyped.socketIo.Server;
public socketConnections = new plugins.lik.ObjectMap<SocketConnection>();
public socketFunctions = new plugins.lik.ObjectMap<SocketFunction<any>>();
public socketRequests = new plugins.lik.ObjectMap<SocketRequest<any>>();
@@ -40,26 +39,62 @@ export class Smartsocket {
this.alias = plugins.isounique.uni(this.options.alias);
}
public async setExternalServer(serverType: 'smartexpress', serverArg: any) {
await this.socketServer.setExternalServer(serverType, serverArg);
/**
* Set an external server (smartserve) for WebSocket handling
*/
public async setExternalServer(
serverType: 'smartserve',
serverArg: any,
websocketHooks?: pluginsTyped.ISmartserveWebSocketHooks
) {
await this.socketServer.setExternalServer(serverType, serverArg, websocketHooks);
}
/**
* starts smartsocket
*/
public async start() {
const socketIoModule = await this.smartenv.getSafeNodeModule('socket.io');
this.io = new socketIoModule.Server(await this.socketServer.getServerForSocketIo(), {
cors: {
allowedHeaders: '*',
methods: '*',
origin: '*',
},
});
await this.socketServer.start();
this.io.on('connection', (socketArg) => {
this._handleSocketConnection(socketArg);
}
/**
* Handle a new WebSocket connection
* Called by SocketServer when a new connection is established
*/
public async handleNewConnection(socket: WebSocket | pluginsTyped.ws.WebSocket) {
const socketConnection: SocketConnection = new SocketConnection({
alias: undefined,
authenticated: false,
side: 'server',
smartsocketHost: this,
socket: socket,
});
logger.log('info', 'Socket connected. Trying to authenticate...');
this.socketConnections.add(socketConnection);
// Handle disconnection
const handleClose = () => {
this.socketConnections.remove(socketConnection);
socketConnection.eventSubject.next('disconnected');
};
socket.addEventListener('close', handleClose);
socket.addEventListener('error', handleClose);
try {
await socketConnection.authenticate();
await socketConnection.listenToFunctionRequests();
// Signal that the server is ready
socketConnection.sendMessage({
type: 'serverReady',
payload: {},
});
} catch (err) {
logger.log('warn', `Authentication failed: ${err}`);
this.socketConnections.remove(socketConnection);
}
}
/**
@@ -77,10 +112,9 @@ export class Smartsocket {
}
});
this.socketConnections.wipe();
this.io.close();
// stop the corresponging server
this.socketServer.stop();
// stop the corresponding server
await this.socketServer.stop();
}
// communication
@@ -110,28 +144,4 @@ export class Smartsocket {
public addSocketFunction(socketFunction: SocketFunction<any>) {
this.socketFunctions.add(socketFunction);
}
/**
* the standard handler for new socket connections
*/
private async _handleSocketConnection(socketArg: pluginsTyped.socketIo.Socket) {
const socketConnection: SocketConnection = new SocketConnection({
alias: undefined,
authenticated: false,
side: 'server',
smartsocketHost: this,
socket: socketArg,
});
logger.log('info', 'Socket connected. Trying to authenticate...');
this.socketConnections.add(socketConnection);
const disconnectSubscription = socketConnection.eventSubject.subscribe((eventArg) => {
if (eventArg === 'disconnected') {
this.socketConnections.remove(socketConnection);
disconnectSubscription.unsubscribe();
}
});
await socketConnection.authenticate();
await socketConnection.listenToFunctionRequests();
await socketConnection.socket.emit('serverFullyReactive');
}
}

230
ts/smartsocket.classes.smartsocketclient.ts
View File

@@ -16,7 +16,7 @@ import { logger } from './smartsocket.logging.js';
export interface ISmartsocketClientOptions {
port: number;
url: string;
alias: string; // an alias makes it easier to identify this client in a multo client environment
alias: string; // an alias makes it easier to identify this client in a multi client environment
autoReconnect?: boolean;
maxRetries?: number; // maximum number of reconnection attempts
initialBackoffDelay?: number; // initial backoff delay in ms
@@ -97,115 +97,162 @@ export class SmartsocketClient {
this.socketFunctions.add(socketFunction);
}
private isReconnecting = false;
/**
* connect the client to the server
*/
public async connect() {
// Reset retry counters on new connection attempt
this.currentRetryCount = 0;
this.currentBackoffDelay = this.initialBackoffDelay;
// Only reset retry counters on fresh connection (not during auto-reconnect)
if (!this.isReconnecting) {
this.currentRetryCount = 0;
this.currentBackoffDelay = this.initialBackoffDelay;
}
this.isReconnecting = false;
const done = plugins.smartpromise.defer();
const smartenvInstance = new plugins.smartenv.Smartenv();
const socketIoClient: any = await smartenvInstance.getEnvAwareModule({
nodeModuleName: 'socket.io-client',
webUrlArg: 'https://cdn.jsdelivr.net/npm/socket.io-client@4/dist/socket.io.js',
getFunction: () => {
const socketIoBrowserModule = (globalThis as any).io;
// console.log('loaded socket.io for browser');
return socketIoBrowserModule;
},
});
// console.log(socketIoClient);
logger.log('info', 'trying to connect...');
const socketUrl = `${this.serverUrl}:${this.serverPort}`;
// Construct WebSocket URL
const protocol = this.serverUrl.startsWith('https') ? 'wss' : 'ws';
const host = this.serverUrl.replace(/^https?:\/\//, '');
const socketUrl = `${protocol}://${host}:${this.serverPort}`;
// Get WebSocket implementation (native in browser, ws in Node)
let WebSocketClass: typeof WebSocket;
if (typeof WebSocket !== 'undefined') {
// Browser environment
WebSocketClass = WebSocket;
} else {
// Node.js environment
const wsModule = await smartenvInstance.getSafeNodeModule('ws');
WebSocketClass = wsModule.default || wsModule;
}
const socket = new WebSocketClass(socketUrl);
this.currentSocket = socket;
this.socketConnection = new SocketConnection({
alias: this.alias,
authenticated: false,
side: 'client',
smartsocketHost: this,
socket: await socketIoClient
.connect(socketUrl, {
multiplex: true,
rememberUpgrade: true,
autoConnect: false,
reconnectionAttempts: 0,
rejectUnauthorized: socketUrl.startsWith('https://localhost') ? false : true,
})
.open(),
socket: socket as any,
});
// Increment attempt ID to invalidate any pending timers from previous attempts
this.connectionAttemptId++;
const currentAttemptId = this.connectionAttemptId;
const timer = new plugins.smarttime.Timer(5000);
timer.start();
timer.completed.then(() => {
this.updateStatus('timedOut');
logger.log('warn', 'connection to server timed out.');
this.disconnect(true);
// Only fire timeout if this is still the current connection attempt
if (currentAttemptId === this.connectionAttemptId && this.eventStatus !== 'connected') {
this.updateStatus('timedOut');
logger.log('warn', 'connection to server timed out.');
this.disconnect(true);
}
});
// authentication flow
this.socketConnection.socket.on('requestAuth', (dataArg: interfaces.IRequestAuthPayload) => {
// Handle connection open
socket.addEventListener('open', () => {
timer.reset();
logger.log('info', `server ${dataArg.serverAlias} requested authentication`);
});
// lets register the authenticated event
this.socketConnection.socket.on('authenticated', async () => {
this.remoteShortId = dataArg.serverAlias;
logger.log('info', 'client is authenticated');
this.socketConnection.authenticated = true;
await this.socketConnection.listenToFunctionRequests();
});
// Handle messages
socket.addEventListener('message', async (event: MessageEvent | { data: string }) => {
try {
const data = typeof event.data === 'string' ? event.data : event.data.toString();
const message: interfaces.ISocketMessage = JSON.parse(data);
this.socketConnection.socket.on('serverFullyReactive', async () => {
// lets take care of retagging
const oldTagStore = this.tagStore;
this.tagStoreSubscription?.unsubscribe();
for (const keyArg of Object.keys(this.tagStore)) {
this.socketConnection.addTag(this.tagStore[keyArg]);
switch (message.type) {
case 'authRequest':
timer.reset();
const authRequestPayload = message.payload as interfaces.IAuthRequestPayload;
logger.log('info', `server ${authRequestPayload.serverAlias} requested authentication`);
this.remoteShortId = authRequestPayload.serverAlias;
// Send authentication data
this.socketConnection.sendMessage({
type: 'auth',
payload: { alias: this.alias },
});
break;
case 'authResponse':
const authResponse = message.payload as interfaces.IAuthResponsePayload;
if (authResponse.success) {
logger.log('info', 'client is authenticated');
this.socketConnection.authenticated = true;
} else {
logger.log('warn', `authentication failed: ${authResponse.error}`);
await this.disconnect();
}
break;
case 'serverReady':
// Set up function request listening
await this.socketConnection.listenToFunctionRequests();
// Handle retagging
const oldTagStore = this.tagStore;
this.tagStoreSubscription?.unsubscribe();
for (const keyArg of Object.keys(this.tagStore)) {
this.socketConnection.addTag(this.tagStore[keyArg]);
}
this.tagStoreSubscription = this.socketConnection.tagStoreObservable.subscribe(
(tagStoreArg) => {
this.tagStore = tagStoreArg;
}
);
for (const tag of Object.keys(oldTagStore)) {
await this.addTag(oldTagStore[tag]);
}
this.updateStatus('connected');
done.resolve();
break;
default:
// Other messages are handled by SocketConnection
this.socketConnection.handleMessage(message);
break;
}
this.tagStoreSubscription = this.socketConnection.tagStoreObservable.subscribe(
(tagStoreArg) => {
this.tagStore = tagStoreArg;
}
} catch (err) {
// Not a valid JSON message, ignore
}
});
// Handle disconnection and errors
const closeHandler = async () => {
// Only handle close if this is still the current socket and we're not already disconnecting
if (this.currentSocket === socket && !this.disconnectRunning) {
logger.log(
'info',
`SocketConnection with >alias ${this.alias} on >side client disconnected`
);
await this.disconnect(true);
}
};
for (const tag of Object.keys(oldTagStore)) {
await this.addTag(oldTagStore[tag]);
}
this.updateStatus('connected');
done.resolve();
});
const errorHandler = async () => {
if (this.currentSocket === socket && !this.disconnectRunning) {
await this.disconnect(true);
}
};
// lets register the forbidden event
this.socketConnection.socket.on('forbidden', async () => {
logger.log('warn', `disconnecting due to being forbidden to use the ressource`);
await this.disconnect();
});
socket.addEventListener('close', closeHandler);
socket.addEventListener('error', errorHandler);
// lets provide the actual auth data
this.socketConnection.socket.emit('dataAuth', {
alias: this.alias,
});
});
// handle connection
this.socketConnection.socket.on('connect', async () => {});
// handle disconnection and errors
this.socketConnection.socket.on('disconnect', async () => {
await this.disconnect(true);
});
this.socketConnection.socket.on('reconnect_failed', async () => {
await this.disconnect(true);
});
this.socketConnection.socket.on('connect_error', async () => {
await this.disconnect(true);
});
return done.promise;
}
private disconnectRunning = false;
private currentSocket: WebSocket | null = null;
private connectionAttemptId = 0; // Increment on each connect attempt to invalidate old timers
/**
* disconnect from the server
@@ -217,11 +264,16 @@ export class SmartsocketClient {
this.disconnectRunning = true;
this.updateStatus('disconnecting');
this.tagStoreSubscription?.unsubscribe();
// Store reference to current socket before cleanup
const socketToClose = this.currentSocket;
this.currentSocket = null;
if (this.socketConnection) {
await this.socketConnection.disconnect();
this.socketConnection = undefined;
logger.log('ok', 'disconnected socket!');
} else {
} else if (!socketToClose) {
this.disconnectRunning = false;
logger.log('warn', 'tried to disconnect, without a SocketConnection');
return;
@@ -232,28 +284,29 @@ export class SmartsocketClient {
if (this.autoReconnect && useAutoReconnectSetting && this.eventStatus !== 'connecting') {
this.updateStatus('connecting');
// Check if we've exceeded the maximum number of retries
if (this.currentRetryCount >= this.maxRetries) {
logger.log('warn', `Maximum reconnection attempts (${this.maxRetries}) reached. Giving up.`);
this.disconnectRunning = false;
return;
}
// Increment retry counter
this.currentRetryCount++;
// Calculate backoff with jitter (±20% randomness)
const jitter = this.currentBackoffDelay * 0.2 * (Math.random() * 2 - 1);
const delay = Math.min(this.currentBackoffDelay + jitter, this.maxBackoffDelay);
logger.log('info', `Reconnect attempt ${this.currentRetryCount}/${this.maxRetries} in ${Math.round(delay)}ms`);
// Apply exponential backoff for next time (doubling with each attempt)
this.currentBackoffDelay = Math.min(this.currentBackoffDelay * 2, this.maxBackoffDelay);
await plugins.smartdelay.delayFor(delay);
this.disconnectRunning = false;
this.isReconnecting = true;
await this.connect();
} else {
this.disconnectRunning = false;
@@ -279,7 +332,6 @@ export class SmartsocketClient {
functionNameArg: T['method'],
dataArg: T['request']
): Promise<T['response']> {
const done = plugins.smartpromise.defer();
const socketRequest = new SocketRequest<T>(this, {
side: 'requesting',
originSocketConnection: this.socketConnection,
@@ -299,14 +351,14 @@ export class SmartsocketClient {
this.eventSubject.next(statusArg);
}
this.eventStatus = statusArg;
// Reset reconnection state when connection is successful
if (statusArg === 'connected') {
this.currentRetryCount = 0;
this.currentBackoffDelay = this.initialBackoffDelay;
}
}
/**
* Resets the reconnection state
*/

235
ts/smartsocket.classes.socketconnection.ts
View File

@@ -26,7 +26,7 @@ export interface ISocketConnectionConstructorOptions {
authenticated: boolean;
side: TSocketConnectionSide;
smartsocketHost: Smartsocket | SmartsocketClient;
socket: pluginsTyped.socketIo.Socket | pluginsTyped.socketIoClient.Socket;
socket: WebSocket | pluginsTyped.ws.WebSocket;
}
/**
@@ -47,7 +47,7 @@ export class SocketConnection {
public side: TSocketConnectionSide;
public authenticated: boolean = false;
public smartsocketRef: Smartsocket | SmartsocketClient;
public socket: pluginsTyped.socketIo.Socket | pluginsTyped.socketIoClient.Socket;
public socket: WebSocket | pluginsTyped.ws.WebSocket;
public eventSubject = new plugins.smartrx.rxjs.Subject<interfaces.TConnectionStatus>();
public eventStatus: interfaces.TConnectionStatus = 'new';
@@ -65,20 +65,94 @@ export class SocketConnection {
// standard behaviour that is always true
allSocketConnections.add(this);
}
// handle connection
this.socket.on('connect', async () => {
this.updateStatus('connected');
});
this.socket.on('disconnect', async () => {
logger.log(
'info',
`SocketConnection with >alias ${this.alias} on >side ${this.side} disconnected`
);
await this.disconnect();
allSocketConnections.remove(this);
this.eventSubject.next('disconnected');
});
/**
* Sends a message through the socket
*/
public sendMessage(message: interfaces.ISocketMessage): void {
if (this.socket.readyState === 1) { // WebSocket.OPEN
this.socket.send(JSON.stringify(message));
}
}
/**
* Handles incoming messages
*/
public handleMessage(messageData: interfaces.ISocketMessage): void {
switch (messageData.type) {
case 'function':
this.handleFunctionCall(messageData);
break;
case 'functionResponse':
this.handleFunctionResponse(messageData);
break;
case 'tagUpdate':
this.handleTagUpdate(messageData);
break;
default:
// Authentication messages are handled by the server/client classes
break;
}
}
private handleFunctionCall(messageData: interfaces.ISocketMessage): void {
const requestData: ISocketRequestDataObject<any> = {
funcCallData: {
funcName: messageData.payload.funcName,
funcDataArg: messageData.payload.funcData,
},
shortId: messageData.id,
};
const referencedFunction: SocketFunction<any> =
this.smartsocketRef.socketFunctions.findSync((socketFunctionArg) => {
return socketFunctionArg.name === requestData.funcCallData.funcName;
});
if (referencedFunction) {
const localSocketRequest = new SocketRequest(this.smartsocketRef, {
side: 'responding',
originSocketConnection: this,
shortId: requestData.shortId,
funcCallData: requestData.funcCallData,
});
localSocketRequest.createResponse();
} else {
logger.log('warn', `function ${requestData.funcCallData.funcName} not found or out of scope`);
}
}
private handleFunctionResponse(messageData: interfaces.ISocketMessage): void {
const responseData: ISocketRequestDataObject<any> = {
funcCallData: {
funcName: messageData.payload.funcName,
funcDataArg: messageData.payload.funcData,
},
shortId: messageData.id,
};
const targetSocketRequest = SocketRequest.getSocketRequestById(
this.smartsocketRef,
responseData.shortId
);
if (targetSocketRequest) {
targetSocketRequest.handleResponse(responseData);
}
}
private handleTagUpdate(messageData: interfaces.ISocketMessage): void {
const tagStoreArg = messageData.payload.tags as interfaces.TTagStore;
if (!plugins.smartjson.deepEqualObjects(this.tagStore, tagStoreArg)) {
this.tagStore = tagStoreArg;
// Echo back to confirm
this.sendMessage({
type: 'tagUpdate',
payload: { tags: this.tagStore },
});
this.tagStoreObservable.next(this.tagStore);
}
this.remoteTagStoreObservable.next(tagStoreArg);
}
/**
@@ -99,7 +173,10 @@ export class SocketConnection {
done.resolve();
}
});
this.socket.emit('updateTagStore', this.tagStore);
this.sendMessage({
type: 'tagUpdate',
payload: { tags: this.tagStore },
});
await done.promise;
}
@@ -117,36 +194,68 @@ export class SocketConnection {
public async removeTagById(tagIdArg: interfaces.ITag['id']) {
delete this.tagStore[tagIdArg];
this.tagStoreObservable.next(this.tagStore);
this.socket.emit('updateTagStore', this.tagStore);
this.sendMessage({
type: 'tagUpdate',
payload: { tags: this.tagStore },
});
}
// authenticating --------------------------
/**
* authenticate the socket
* authenticate the socket (server side)
*/
public authenticate() {
const done = plugins.smartpromise.defer();
this.socket.on('dataAuth', async (dataArg: ISocketConnectionAuthenticationObject) => {
logger.log('info', 'received authentication data...');
this.socket.removeAllListeners('dataAuth');
if (dataArg.alias) {
// TODO: authenticate password
this.alias = dataArg.alias;
this.authenticated = true;
this.socket.emit('authenticated');
logger.log('ok', `socket with >>alias ${this.alias} is authenticated!`);
done.resolve(this);
} else {
this.authenticated = false;
await this.disconnect();
done.reject('a socket tried to connect, but could not authenticated.');
public authenticate(): Promise<SocketConnection> {
const done = plugins.smartpromise.defer<SocketConnection>();
// Set up message handler for authentication
const messageHandler = (event: MessageEvent | { data: string }) => {
try {
const data = typeof event.data === 'string' ? event.data : event.data.toString();
const message: interfaces.ISocketMessage = JSON.parse(data);
if (message.type === 'auth') {
const authData = message.payload as interfaces.IAuthPayload;
logger.log('info', 'received authentication data...');
if (authData.alias) {
this.alias = authData.alias;
this.authenticated = true;
// Send authentication response
this.sendMessage({
type: 'authResponse',
payload: { success: true },
});
logger.log('ok', `socket with >>alias ${this.alias} is authenticated!`);
done.resolve(this);
} else {
this.authenticated = false;
this.sendMessage({
type: 'authResponse',
payload: { success: false, error: 'No alias provided' },
});
this.disconnect();
done.reject('a socket tried to connect, but could not authenticate.');
}
}
} catch (err) {
// Not a valid message, ignore
}
});
const requestAuthPayload: interfaces.IRequestAuthPayload = {
serverAlias: this.smartsocketRef.alias,
};
this.socket.emit('requestAuth', requestAuthPayload);
this.socket.addEventListener('message', messageHandler as any);
// Request authentication
const requestAuthPayload: interfaces.TAuthRequestMessage = {
type: 'authRequest',
payload: {
serverAlias: (this.smartsocketRef as Smartsocket).alias,
},
};
this.sendMessage(requestAuthPayload);
return done.promise;
}
@@ -158,43 +267,18 @@ export class SocketConnection {
public listenToFunctionRequests() {
const done = plugins.smartpromise.defer();
if (this.authenticated) {
this.socket.on('function', (dataArg: ISocketRequestDataObject<any>) => {
// check if requested function is available to the socket's scope
// logger.log('info', 'function request received');
const referencedFunction: SocketFunction<any> =
this.smartsocketRef.socketFunctions.findSync((socketFunctionArg) => {
return socketFunctionArg.name === dataArg.funcCallData.funcName;
});
if (referencedFunction) {
// logger.log('ok', 'function in access scope');
const localSocketRequest = new SocketRequest(this.smartsocketRef, {
side: 'responding',
originSocketConnection: this,
shortId: dataArg.shortId,
funcCallData: dataArg.funcCallData,
});
localSocketRequest.createResponse(); // takes care of creating response and sending it back
} else {
logger.log('warn', 'function not existent or out of access scope');
// Set up message handler for all messages
const messageHandler = (event: MessageEvent | { data: string }) => {
try {
const data = typeof event.data === 'string' ? event.data : event.data.toString();
const message: interfaces.ISocketMessage = JSON.parse(data);
this.handleMessage(message);
} catch (err) {
// Not a valid JSON message, ignore
}
});
this.socket.on('functionResponse', (dataArg: ISocketRequestDataObject<any>) => {
// logger.log('info', `received response for request with id ${dataArg.shortId}`);
const targetSocketRequest = SocketRequest.getSocketRequestById(
this.smartsocketRef,
dataArg.shortId
);
targetSocketRequest.handleResponse(dataArg);
});
};
this.socket.on('updateTagStore', async (tagStoreArg: interfaces.TTagStore) => {
if (!plugins.smartjson.deepEqualObjects(this.tagStore, tagStoreArg)) {
this.tagStore = tagStoreArg;
this.socket.emit('updateTagStore', this.tagStore);
this.tagStoreObservable.next(this.tagStore);
}
this.remoteTagStoreObservable.next(tagStoreArg);
});
this.socket.addEventListener('message', messageHandler as any);
logger.log(
'info',
@@ -211,7 +295,10 @@ export class SocketConnection {
// disconnecting ----------------------
public async disconnect() {
this.socket.disconnect(true);
if (this.socket.readyState === 1 || this.socket.readyState === 0) {
this.socket.close();
}
allSocketConnections.remove(this);
this.updateStatus('disconnected');
}

29
ts/smartsocket.classes.socketrequest.ts
View File

@@ -1,4 +1,5 @@
import * as plugins from './smartsocket.plugins.js';
import * as interfaces from './interfaces/index.js';
// import interfaces
import {
@@ -78,11 +79,15 @@ export class SocketRequest<T extends plugins.typedrequestInterfaces.ITypedReques
* dispatches a socketrequest from the requesting to the receiving side
*/
public dispatch(): Promise<ISocketFunctionCallDataResponse<T>> {
const requestData: ISocketRequestDataObject<T> = {
funcCallData: this.funcCallData,
shortId: this.shortid,
const message: interfaces.ISocketMessage<interfaces.IFunctionCallPayload> = {
type: 'function',
id: this.shortid,
payload: {
funcName: this.funcCallData.funcName,
funcData: this.funcCallData.funcDataArg,
},
};
this.originSocketConnection.socket.emit('function', requestData);
this.originSocketConnection.sendMessage(message);
return this.done.promise;
}
@@ -90,7 +95,6 @@ export class SocketRequest<T extends plugins.typedrequestInterfaces.ITypedReques
* handles the response that is received by the requesting side
*/
public async handleResponse(responseDataArg: ISocketRequestDataObject<T>) {
// logger.log('info', 'handling response!');
this.done.resolve(responseDataArg.funcCallData);
this.smartsocketRef.socketRequests.remove(this);
}
@@ -110,16 +114,19 @@ export class SocketRequest<T extends plugins.typedrequestInterfaces.ITypedReques
logger.log('error', `There is no SocketFunction defined for ${this.funcCallData.funcName}`);
return;
}
// logger.log('info', `invoking ${targetSocketFunction.name}`);
targetSocketFunction
.invoke(this.funcCallData, this.originSocketConnection)
.then((resultData) => {
// logger.log('info', 'got resultData. Sending it to requesting party.');
const responseData: ISocketRequestDataObject<T> = {
funcCallData: resultData,
shortId: this.shortid,
const message: interfaces.ISocketMessage<interfaces.IFunctionCallPayload> = {
type: 'functionResponse',
id: this.shortid,
payload: {
funcName: resultData.funcName,
funcData: resultData.funcDataArg,
},
};
this.originSocketConnection.socket.emit('functionResponse', responseData);
this.originSocketConnection.sendMessage(message);
this.smartsocketRef.socketRequests.remove(this);
});
}

218
ts/smartsocket.classes.socketserver.ts
View File

@@ -6,16 +6,23 @@ import { Smartsocket } from './smartsocket.classes.smartsocket.js';
import { logger } from './smartsocket.logging.js';
/**
* class socketServer
* handles the attachment of socketIo to whatever server is in play
* class SocketServer
* handles the WebSocket server, either standalone or integrated with smartserve
*/
export class SocketServer {
private smartsocket: Smartsocket;
private httpServerDeferred: plugins.smartpromise.Deferred<any>;
private httpServer: pluginsTyped.http.Server | pluginsTyped.https.Server;
private wsServer: pluginsTyped.ws.WebSocketServer;
/**
* wether httpServer is standalone
* whether we're using an external server (smartserve)
*/
private externalServerMode = false;
private externalServer: any = null;
private externalWebSocketHooks: pluginsTyped.ISmartserveWebSocketHooks = null;
/**
* whether httpServer is standalone
*/
private standaloneServer = false;
@@ -24,71 +31,200 @@ export class SocketServer {
}
/**
* starts the server with another server
* also works with an express style server
* Set an external server (smartserve) for WebSocket handling
*/
public async setExternalServer(
serverType: 'smartexpress',
serverArg: pluginsTyped.typedserver.servertools.Server
serverType: 'smartserve',
serverArg: any,
websocketHooks?: pluginsTyped.ISmartserveWebSocketHooks
) {
this.httpServerDeferred = plugins.smartpromise.defer();
await serverArg.startedPromise;
this.httpServer = serverArg.httpServer;
this.httpServerDeferred.resolve();
if (serverType !== 'smartserve') {
throw new Error(`Unsupported server type: ${serverType}. Only 'smartserve' is supported.`);
}
this.externalServerMode = true;
this.externalServer = serverArg;
this.externalWebSocketHooks = websocketHooks || null;
}
/**
* gets the server for socket.io
*/
public async getServerForSocketIo() {
if (this.httpServerDeferred) {
await this.httpServerDeferred.promise;
}
if (this.httpServer) {
return this.httpServer;
} else {
const httpModule = await this.smartsocket.smartenv.getSafeNodeModule('http');
this.httpServer = new httpModule.Server();
this.standaloneServer = true;
return this.httpServer;
}
}
/**
* starts listening to incoming sockets:
* starts listening to incoming websocket connections
*/
public async start() {
const done = plugins.smartpromise.defer();
// handle http servers
// in case an external server has been set "this.standaloneServer" should be false
if (this.httpServer && this.standaloneServer) {
if (this.externalServerMode) {
// Using external smartserve server
// The smartserve server should be configured with websocket hooks
// that call our handleNewConnection method
logger.log('info', 'Using external smartserve server for WebSocket handling');
// If smartserve provides a way to get the underlying http server for upgrade,
// we could attach ws to it. For now, we expect smartserve to handle WS
// and call us back via the hooks.
done.resolve();
} else {
// Standalone mode - create our own HTTP server and WebSocket server
const httpModule = await this.smartsocket.smartenv.getSafeNodeModule('http');
const wsModule = await this.smartsocket.smartenv.getSafeNodeModule('ws');
if (!this.smartsocket.options.port) {
logger.log('error', 'there should be a port specifed for smartsocket!');
logger.log('error', 'there should be a port specified for smartsocket!');
throw new Error('there should be a port specified for smartsocket');
}
this.httpServer = httpModule.createServer();
this.standaloneServer = true;
// Create WebSocket server attached to HTTP server
this.wsServer = new wsModule.WebSocketServer({ server: this.httpServer });
this.wsServer.on('connection', (ws: pluginsTyped.ws.WebSocket) => {
this.smartsocket.handleNewConnection(ws);
});
this.httpServer.listen(this.smartsocket.options.port, () => {
logger.log(
'success',
`Server started in standalone mode on ${this.smartsocket.options.port}`
`Server started in standalone mode on port ${this.smartsocket.options.port}`
);
done.resolve();
});
} else {
done.resolve();
}
// nothing else to do if express server is set
await done.promise;
return;
}
/**
* closes the server
*/
public async stop() {
if (this.httpServer) {
this.httpServer.close();
const done = plugins.smartpromise.defer<void>();
let resolved = false;
if (this.wsServer) {
// Close all WebSocket connections
this.wsServer.clients.forEach((client) => {
client.terminate();
});
this.wsServer.close();
this.wsServer = null;
}
if (this.httpServer && this.standaloneServer) {
const resolveOnce = () => {
if (!resolved) {
resolved = true;
this.httpServer = null;
this.standaloneServer = false;
done.resolve();
}
};
this.httpServer.close(() => {
resolveOnce();
});
// Add a timeout in case close callback doesn't fire
const timeoutId = setTimeout(() => {
resolveOnce();
}, 2000);
// Ensure timeout doesn't keep process alive
if (timeoutId.unref) {
timeoutId.unref();
}
} else {
done.resolve();
}
await done.promise;
}
/**
* Returns WebSocket hooks for integration with smartserve
* Call this to get hooks that you can pass to smartserve's websocket config
*/
public getSmartserveWebSocketHooks(): pluginsTyped.ISmartserveWebSocketHooks {
return {
onOpen: async (peer: pluginsTyped.ISmartserveWebSocketPeer) => {
// Create a wrapper that adapts ISmartserveWebSocketPeer to WebSocket-like interface
const wsLikeSocket = this.createWsLikeFromPeer(peer);
await this.smartsocket.handleNewConnection(wsLikeSocket as any);
// Call external hooks if provided
if (this.externalWebSocketHooks?.onOpen) {
await this.externalWebSocketHooks.onOpen(peer);
}
},
onMessage: async (peer: pluginsTyped.ISmartserveWebSocketPeer, message: pluginsTyped.ISmartserveWebSocketMessage) => {
// Messages are handled by SocketConnection via the adapter
// But we still call external hooks if provided
if (this.externalWebSocketHooks?.onMessage) {
await this.externalWebSocketHooks.onMessage(peer, message);
}
},
onClose: async (peer: pluginsTyped.ISmartserveWebSocketPeer, code: number, reason: string) => {
if (this.externalWebSocketHooks?.onClose) {
await this.externalWebSocketHooks.onClose(peer, code, reason);
}
},
onError: async (peer: pluginsTyped.ISmartserveWebSocketPeer, error: Error) => {
if (this.externalWebSocketHooks?.onError) {
await this.externalWebSocketHooks.onError(peer, error);
}
},
};
}
/**
* Creates a WebSocket-like object from a smartserve peer
* This allows our SocketConnection to work with both native WebSocket and smartserve peers
*/
private createWsLikeFromPeer(peer: pluginsTyped.ISmartserveWebSocketPeer): any {
const messageListeners: Array<(event: any) => void> = [];
const closeListeners: Array<() => void> = [];
const errorListeners: Array<() => void> = [];
// Store the adapter on the peer for message routing
peer.data.set('smartsocket_adapter', {
dispatchMessage: (data: string) => {
messageListeners.forEach((listener) => {
listener({ data });
});
},
dispatchClose: () => {
closeListeners.forEach((listener) => listener());
},
dispatchError: () => {
errorListeners.forEach((listener) => listener());
},
});
return {
readyState: peer.readyState,
send: (data: string) => peer.send(data),
close: (code?: number, reason?: string) => peer.close(code, reason),
addEventListener: (event: string, listener: any) => {
if (event === 'message') {
messageListeners.push(listener);
} else if (event === 'close') {
closeListeners.push(listener);
} else if (event === 'error') {
errorListeners.push(listener);
}
},
removeEventListener: (event: string, listener: any) => {
if (event === 'message') {
const idx = messageListeners.indexOf(listener);
if (idx >= 0) messageListeners.splice(idx, 1);
} else if (event === 'close') {
const idx = closeListeners.indexOf(listener);
if (idx >= 0) closeListeners.splice(idx, 1);
} else if (event === 'error') {
const idx = errorListeners.indexOf(listener);
if (idx >= 0) errorListeners.splice(idx, 1);
}
},
};
}
}

51
ts/smartsocket.pluginstyped.ts
View File

@@ -4,20 +4,43 @@ import type * as https from 'https';
export type { http, https };
// pushrocks scope
import type * as typedserver from '@api.global/typedserver';
// third party scope - ws types
import type * as wsTypes from 'ws';
export type { typedserver };
// third party scope
import type { Socket as ServerSocket, Server as ServerServer } from 'socket.io';
import type { Socket as ClientSocket, connect as ClientIo } from 'socket.io-client';
export namespace socketIo {
export type Socket = ServerSocket;
export type Server = ServerServer;
export namespace ws {
export type WebSocket = wsTypes.WebSocket;
export type WebSocketServer = wsTypes.WebSocketServer;
export type RawData = wsTypes.RawData;
}
export namespace socketIoClient {
export type Socket = ClientSocket;
export type connect = typeof ClientIo;
// smartserve compatibility interface (for setExternalServer)
// This mirrors the IWebSocketPeer interface from smartserve
export interface ISmartserveWebSocketPeer {
id: string;
url: string;
readyState: 0 | 1 | 2 | 3;
protocol: string;
extensions: string;
send(data: string): void;
sendBinary(data: Uint8Array | ArrayBuffer): void;
close(code?: number, reason?: string): void;
ping(data?: Uint8Array): void;
terminate(): void;
context: any;
data: Map<string, unknown>;
tags: Set<string>;
}
export interface ISmartserveWebSocketMessage {
type: 'text' | 'binary';
text?: string;
data?: Uint8Array;
size: number;
}
export interface ISmartserveWebSocketHooks {
onOpen?: (peer: ISmartserveWebSocketPeer) => void | Promise<void>;
onMessage?: (peer: ISmartserveWebSocketPeer, message: ISmartserveWebSocketMessage) => void | Promise<void>;
onClose?: (peer: ISmartserveWebSocketPeer, code: number, reason: string) => void | Promise<void>;
onError?: (peer: ISmartserveWebSocketPeer, error: Error) => void | Promise<void>;
}