Merge pull request #34 from xconnio/connected-function-disconnect-callback

Add function to check connection state, handle goodbye and disconnect callback

Author: Mahad-10

examples/reconnect.ts

@@ -0,0 +1,55 @@
+import { connectAnonymous } from "../lib";
+
+async function sleep(ms: number): Promise<void> {
+    return new Promise(resolve => setTimeout(resolve, ms));
+}
+
+async function main() {
+    let retryDelay = 1000; // 1 second
+    const maxDelay = 30000; // 30 seconds
+
+    for (; ;) {
+        try {
+            const session = await connectAnonymous("ws://localhost:8080/ws", "realm1");
+            console.log("Connected successfully");
+
+            // Reset backoff after successful connection
+            retryDelay = 1000;
+
+            // Register multiple disconnect callbacks
+            session.onDisconnect(async () => {
+                console.log("Callback 1: disconnection event.");
+            });
+
+            session.onDisconnect(async () => {
+                console.log("Callback 2: disconnection event.");
+                await sleep(500);
+            });
+
+            session.onDisconnect(async () => {
+                console.log("Callback 3: disconnection event.");
+            });
+
+            // Wait for disconnect
+            const disconnected = new Promise<void>(resolve => {
+                session.onDisconnect(async () => {
+                    console.log("Disconnected from router!");
+                    resolve();
+                });
+            });
+
+            await disconnected;
+            console.log("Retrying connection...");
+
+        } catch (err) {
+            console.error(`Failed to connect: ${err}`);
+            console.log(`Retrying in ${retryDelay / 1000}s...`);
+            await sleep(retryDelay);
+
+            // Exponential backoff
+            retryDelay = Math.min(retryDelay * 2, maxDelay);
+        }
+    }
+}
+
+main().catch(err => console.error(`Fatal error: ${err}`));

lib/session.ts

@@ -18,11 +18,12 @@ import {
     Event as EventMsg,
     Unsubscribe, UnsubscribeFields,
     Unsubscribed,
-    Error, ErrorFields
+    Error, ErrorFields,
+    Goodbye, GoodbyeFields
 } from "wampproto";
 
 import {wampErrorString} from "./helpers";
-import {ERROR_RUNTIME_ERROR} from "./wamp";
+import {ERROR_RUNTIME_ERROR, CLOSE_CLOSE_REALM} from "./wamp";
 import {ApplicationError, ProtocolError} from "./exception";
 import {
     IBaseSession,
@@ -42,6 +43,7 @@ export class Session {
     private _baseSession: IBaseSession;
     private _wampSession: WAMPSession;
     private _idGen: SessionScopeIDGenerator = new SessionScopeIDGenerator();
+    private _disconnectCallbacks: Array<() => Promise<void>> = [];
 
     private _callRequests: Map<number, {
         resolve: (value: Result) => void,
@@ -58,9 +60,24 @@ export class Session {
     private _subscriptions: Map<number, (event: Event) => void> = new Map();
     private _unsubscribeRequests: Map<number, UnsubscribeRequest> = new Map();
 
+    private _goodbyeRequest = (() => {
+        let resolve!: () => void;
+        let isCompleted = false;
+        const promise = new Promise<void>((res) => {
+            resolve = () => {
+                if (!isCompleted) {
+                    isCompleted = true;
+                    res();
+                }
+            };
+        });
+        return { promise, resolve, isCompleted };
+    })();
+
     constructor(baseSession: IBaseSession) {
         this._baseSession = baseSession;
         this._wampSession = new WAMPSession(baseSession.serializer());
+        this._baseSession.onDisconnect(async () => { await this.markDisconnected();});
 
         (async () => {
             for (; ;) {
@@ -70,12 +87,34 @@ export class Session {
         })();
     }
 
+    onDisconnect(callback: () => Promise<void>): void {
+        this._disconnectCallbacks.push(callback);
+    }
+
     private get _nextID(): number {
         return this._idGen.next();
     }
 
     async close(): Promise<void> {
-        await this._baseSession.close();
+        const goodbye = new Goodbye(new GoodbyeFields({}, CLOSE_CLOSE_REALM));
+        const data = this._wampSession.sendMessage(goodbye)
+        this._baseSession.send(data)
+
+        return Promise.race([
+            this._goodbyeRequest.promise,
+            new Promise<void>((resolve) =>
+                setTimeout(async () => {
+                    await this._baseSession.close();
+                    resolve();
+                }, 10_000)
+            )
+        ]).finally(async () => {
+            await this._baseSession.close();
+        });
+    }
+
+    isConnected(): boolean {
+        return this._baseSession.isConnected();
     }
 
     private async _processIncomingMessage(message: Message): Promise<void> {
@@ -200,11 +239,23 @@ export class Session {
                 default:
                     throw new ProtocolError(wampErrorString(message));
             }
+        } else if (message instanceof Goodbye) {
+            await this.markDisconnected()
         } else {
             throw new ProtocolError(`Unexpected message type ${typeof message}`);
         }
     }
 
+    private async markDisconnected() {
+        if (this._disconnectCallbacks.length > 0) {
+            await Promise.all(this._disconnectCallbacks.map(cb => cb()));
+        }
+
+        if (this._goodbyeRequest && !this._goodbyeRequest.isCompleted) {
+            this._goodbyeRequest.resolve();
+        }
+    }
+
     async call(
         procedure: string,
         args?: any[] | null,

lib/types.ts

@@ -44,13 +44,22 @@ export abstract class IBaseSession {
     async close(): Promise<void> {
         throw new Error("UnimplementedError");
     }
+
+    isConnected(): boolean {
+        throw new Error("UnimplementedError");
+    }
+
+    onDisconnect(callback: () => Promise<void>): void {
+        throw new Error("UnimplementedError");
+    }
 }
 
 export class BaseSession extends IBaseSession {
     private readonly _ws: WebSocket;
     private readonly _wsMessageHandler: any;
     private readonly sessionDetails: SessionDetails;
     private readonly _serializer: Serializer;
+    private _disconnectCallbacks: Array<() => Promise<void>> = [];
 
     constructor(
         ws: WebSocket,
@@ -66,6 +75,9 @@ export class BaseSession extends IBaseSession {
 
         // close cleanly on abrupt client disconnect
         this._ws.addEventListener("close", async () => {
+            if (this._disconnectCallbacks.length > 0) {
+                await Promise.all(this._disconnectCallbacks.map(cb => cb()));
+            }
             await this.close();
         });
     }
@@ -126,6 +138,14 @@ export class BaseSession extends IBaseSession {
 
         this._ws.close();
     }
+
+    isConnected(): boolean {
+        return this._ws.readyState === WebSocket.OPEN;
+    }
+
+    onDisconnect(callback: () => Promise<void>): void {
+        this._disconnectCallbacks.push(callback);
+    }
 }
 
 export class Result {

lib/wamp.ts

@@ -1 +1,2 @@
+export const CLOSE_CLOSE_REALM = "wamp.close.close_realm"
 export const ERROR_RUNTIME_ERROR = "wamp.error.runtime_error"