update api spec

randilfernando · randilfernando · commit 3fb69431a1bc · 2026-01-21T21:21:49.000+05:30
diff --git a/client/README.md b/client/README.md
@@ -33,20 +33,20 @@ const client = new Client({
 
 client.connect();
 
-client.on('open', () => {
+client.onOpen(() => {
     console.log('Connected to server');
 
     // Create a new stream
     const stream = client.createStream('my-stream-1');
 
-    stream.on('data', (data) => {
+    stream.onData((data) => {
         console.log('Received data:', data);
     });
 
     stream.send({ message: 'Hello World' });
 });
 
-client.on('close', () => {
+client.onClose(() => {
     console.log('Disconnected from server');
 });
 ```
diff --git a/client/src/client.ts b/client/src/client.ts
@@ -1,4 +1,3 @@
-import { EventEmitter } from 'events';
 import WebSocket from 'ws';
 import { Message, MessageType } from './protocol';
 import { Stream } from './stream';
@@ -10,7 +9,7 @@ export interface ClientOptions {
     maxRetries?: number;
 }
 
-export class Client extends EventEmitter {
+export class Client {
     private ws: WebSocket | null = null;
     private options: ClientOptions;
     public streams: Map<string, Stream> = new Map();
@@ -19,15 +18,30 @@ export class Client extends EventEmitter {
     private shouldReconnect: boolean = true;
     private retryTimer: NodeJS.Timeout | null = null;
 
+    private openCallbacks: (() => void)[] = [];
+    private closeCallbacks: (() => void)[] = [];
+    private errorCallbacks: ((error: Error) => void)[] = [];
+
     constructor(options: ClientOptions) {
-        super();
         this.options = {
             retryInterval: 1000,
             maxRetries: Infinity,
             ...options
         };
     }
 
+    public onOpen(cb: () => void) {
+        this.openCallbacks.push(cb);
+    }
+
+    public onClose(cb: () => void) {
+        this.closeCallbacks.push(cb);
+    }
+
+    public onError(cb: (error: Error) => void) {
+        this.errorCallbacks.push(cb);
+    }
+
     public connect(): void {
         this.shouldReconnect = true;
         this.initiateConnection();
@@ -44,7 +58,7 @@ export class Client extends EventEmitter {
         this.ws.onopen = () => {
             console.log('Connected');
             this.isConnected = true;
-            this.emit('open');
+            this.openCallbacks.forEach(cb => cb());
             this.flushQueue();
 
             // Clear any existing Retry timer
@@ -69,6 +83,7 @@ export class Client extends EventEmitter {
         this.ws.onerror = (err) => {
             console.error('WebSocket Error:', err);
             // creating error usually triggers close, so cleanup logic runs there.
+            this.errorCallbacks.forEach(cb => cb(new Error(err.message)));
         };
     }
 
@@ -129,7 +144,7 @@ export class Client extends EventEmitter {
                 const stream = this.streams.get(header.streamId);
                 if (stream) {
                     if (header.type === MessageType.CLOSE_STREAM) {
-                        stream.emit('close');
+                        stream.dispatchClose();
                         this.streams.delete(header.streamId);
                     } else if (header.type === MessageType.DATA) {
                         stream.handleMessage(payload);
@@ -156,9 +171,9 @@ export class Client extends EventEmitter {
 
     private cleanup(): void {
         this.isConnected = false;
-        this.emit('close');
+        this.closeCallbacks.forEach(cb => cb());
         // Close all streams
-        this.streams.forEach(stream => stream.emit('close'));
+        this.streams.forEach(stream => stream.dispatchClose());
         this.streams.clear();
     }
 
diff --git a/client/src/stream.ts b/client/src/stream.ts
@@ -1,19 +1,28 @@
-import { EventEmitter } from 'events';
 import { MessageType } from './protocol';
 import { Client } from './client';
 
-export class Stream extends EventEmitter {
+export class Stream {
     public id: string;
     private client: Client;
     public isOpen: boolean;
 
+    private dataCallbacks: ((data: any) => void)[] = [];
+    private closeCallbacks: (() => void)[] = [];
+
     constructor(id: string, client: Client) {
-        super();
         this.id = id;
         this.client = client;
         this.isOpen = true;
     }
 
+    public onData(cb: (data: any) => void) {
+        this.dataCallbacks.push(cb);
+    }
+
+    public onClose(cb: () => void) {
+        this.closeCallbacks.push(cb);
+    }
+
     public send(data: any): void {
         if (!this.isOpen) {
             throw new Error('Stream is closed');
@@ -25,12 +34,19 @@ export class Stream extends EventEmitter {
         if (this.isOpen) {
             this.isOpen = false;
             this.client.send(this.id, MessageType.CLOSE_STREAM, null);
-            this.emit('close');
+            this.dispatchClose();
         }
     }
 
     // Called internally by Client
     public handleMessage(payload: any): void {
-        this.emit('data', payload);
+        this.dataCallbacks.forEach(cb => cb(payload));
+    }
+
+    public dispatchClose(): void {
+        if (this.isOpen) {
+            this.isOpen = false;
+        }
+        this.closeCallbacks.forEach(cb => cb());
     }
 }
diff --git a/server/README.md b/server/README.md
@@ -31,30 +31,30 @@ import { LinkServer } from '@marketrix-ai/link-server';
 
 const server = new LinkServer({ port: 8080 });
 
-server.on('listening', () => {
+server.onListening(() => {
     console.log('Server is listening on port 8080');
 });
 
-server.on('connection', (connection) => {
+server.onConnection((connection) => {
     console.log('Client connected:', connection.id);
 
     // Handle new streams created by the client
-    connection.on('stream', (stream) => {
+    connection.onStream((stream) => {
         console.log(`Stream ${stream.id} created`);
 
-        stream.on('data', (data) => {
+        stream.onData((data) => {
             console.log(`Received data on stream ${stream.id}:`, data);
             
             // Example: sending data back
             // stream.send({ message: 'Hello from server' });
         });
 
-        stream.on('close', () => {
+        stream.onClose(() => {
             console.log(`Stream ${stream.id} closed`);
         });
     });
 
-    connection.on('close', () => {
+    connection.onClose(() => {
         console.log('Client disconnected');
     });
 });
diff --git a/server/src/connection.ts b/server/src/connection.ts
@@ -1,20 +1,22 @@
-import { EventEmitter } from 'events';
 import WebSocket from 'ws';
 import { MessageType, Message } from './protocol';
 import { Stream } from './stream';
 
-export class Connection extends EventEmitter {
+export class Connection {
     private socket: WebSocket;
     public id: string | null;
     public streams: Map<string, Stream>;
 
+    private streamCallbacks: ((stream: Stream) => void)[] = [];
+    private closeCallbacks: (() => void)[] = [];
+    private errorCallbacks: ((error: Error) => void)[] = [];
+
     private pingTimer: NodeJS.Timeout | null = null;
     private pongTimeoutTimer: NodeJS.Timeout | null = null;
     private readonly PING_INTERVAL = 2000; // 2 seconds (TEST)
     private readonly PONG_TIMEOUT = 2000;  // 2 seconds (TEST)
 
     constructor(socket: WebSocket) {
-        super();
         this.socket = socket;
         this.id = null; // Will be set upon first message
         this.streams = new Map();
@@ -26,6 +28,18 @@ export class Connection extends EventEmitter {
         this.startPingLoop();
     }
 
+    public onStream(cb: (stream: Stream) => void) {
+        this.streamCallbacks.push(cb);
+    }
+
+    public onClose(cb: () => void) {
+        this.closeCallbacks.push(cb);
+    }
+
+    public onError(cb: (error: Error) => void) {
+        this.errorCallbacks.push(cb);
+    }
+
     private startPingLoop(): void {
         this.pingTimer = setInterval(() => {
             if (this.socket.readyState === WebSocket.OPEN) {
@@ -84,14 +98,14 @@ export class Connection extends EventEmitter {
                     if (!this.streams.has(streamId)) {
                         const stream = new Stream(streamId, this);
                         this.streams.set(streamId, stream);
-                        this.emit('stream', stream);
+                        this.streamCallbacks.forEach(cb => cb(stream));
                     }
                 }
 
                 const stream = this.streams.get(streamId);
                 if (stream) {
                     if (type === MessageType.CLOSE_STREAM) {
-                        stream.emit('close');
+                        stream.dispatchClose();
                         this.streams.delete(streamId);
                     } else if (type === MessageType.DATA) {
                         stream.handleMessage(payload);
@@ -133,8 +147,8 @@ export class Connection extends EventEmitter {
         if (this.pingTimer) clearInterval(this.pingTimer);
         if (this.pongTimeoutTimer) clearTimeout(this.pongTimeoutTimer);
 
-        this.streams.forEach(stream => stream.emit('close'));
+        this.streams.forEach(stream => stream.dispatchClose());
         this.streams.clear();
-        this.emit('close');
+        this.closeCallbacks.forEach(cb => cb());
     }
 }
diff --git a/server/src/server.ts b/server/src/server.ts
@@ -1,4 +1,3 @@
-import { EventEmitter } from 'events';
 import WebSocket from 'ws';
 import { Connection } from './connection';
 
@@ -7,30 +6,44 @@ export interface LinkServerOptions extends WebSocket.ServerOptions {
     // for now we just extend standard WS server options
 }
 
-export class LinkServer extends EventEmitter {
+export class LinkServer {
     private wss: WebSocket.Server;
+    private connectionCallbacks: ((connection: Connection) => void)[] = [];
+    private errorCallbacks: ((error: Error) => void)[] = [];
+    private listeningCallbacks: (() => void)[] = [];
 
     constructor(options: LinkServerOptions) {
-        super();
         this.wss = new WebSocket.Server(options);
         this.init();
     }
 
     private init() {
         this.wss.on('connection', (ws: WebSocket) => {
             const connection = new Connection(ws);
-            this.emit('connection', connection);
+            this.connectionCallbacks.forEach(cb => cb(connection));
         });
 
         this.wss.on('error', (error) => {
-            this.emit('error', error);
+            this.errorCallbacks.forEach(cb => cb(error));
         });
 
         this.wss.on('listening', () => {
-            this.emit('listening');
+            this.listeningCallbacks.forEach(cb => cb());
         });
     }
 
+    public onConnection(cb: (connection: Connection) => void) {
+        this.connectionCallbacks.push(cb);
+    }
+
+    public onError(cb: (error: Error) => void) {
+        this.errorCallbacks.push(cb);
+    }
+
+    public onListening(cb: () => void) {
+        this.listeningCallbacks.push(cb);
+    }
+
     public close(cb?: (err?: Error) => void) {
         this.wss.close(cb);
     }
diff --git a/server/src/stream.ts b/server/src/stream.ts
@@ -1,19 +1,28 @@
-import { EventEmitter } from 'events';
 import { MessageType } from './protocol';
 import { Connection } from './connection';
 
-export class Stream extends EventEmitter {
+export class Stream {
     public id: string;
     private connection: Connection;
     public isOpen: boolean;
 
+    private dataCallbacks: ((data: any) => void)[] = [];
+    private closeCallbacks: (() => void)[] = [];
+
     constructor(id: string, connection: Connection) {
-        super();
         this.id = id;
         this.connection = connection;
         this.isOpen = true;
     }
 
+    public onData(cb: (data: any) => void) {
+        this.dataCallbacks.push(cb);
+    }
+
+    public onClose(cb: () => void) {
+        this.closeCallbacks.push(cb);
+    }
+
     public send(data: any): void {
         if (!this.isOpen) {
             throw new Error('Stream is closed');
@@ -25,12 +34,20 @@ export class Stream extends EventEmitter {
         if (this.isOpen) {
             this.isOpen = false;
             this.connection.send(this.id, MessageType.CLOSE_STREAM, null);
-            this.emit('close');
+            this.dispatchClose();
         }
     }
 
     // Called when data is received from the connection for this stream
     public handleMessage(payload: any): void {
-        this.emit('data', payload);
+        this.dataCallbacks.forEach(cb => cb(payload));
+    }
+
+    // Called internally or by Connection when stream is closed remotely
+    public dispatchClose(): void {
+        if (this.isOpen) {
+            this.isOpen = false;
+        }
+        this.closeCallbacks.forEach(cb => cb());
     }
 }
