add support for custom transport

Author: Mahad-10

lib/joiner.ts

@@ -1,6 +1,6 @@
 import {Joiner, ClientAuthenticator, Serializer, JSONSerializer} from 'wampproto';
 
-import {BaseSession} from './types';
+import {BaseSession, Peer, WebSocketPeer} from './types';
 import {getSubProtocol} from './helpers';
 
 
@@ -21,47 +21,45 @@ export class WAMPSessionJoiner {
     }
 
     async join(uri: string, realm: string): Promise<BaseSession> {
-        await ensureGlobalWebSocket()
+        await ensureGlobalWebSocket();
         const ws = new WebSocket(uri, [getSubProtocol(this._serializer)]);
+        await new Promise((resolve, reject) => {
+            ws.addEventListener("open", resolve);
+            ws.addEventListener("error", reject);
+        });
+
+        const peer = new WebSocketPeer(ws);
+        return joinPeer(peer, realm, this._serializer, this._authenticator);
+    }
+}
 
-        const joiner = new Joiner(realm, this._serializer, this._authenticator);
 
-        ws.addEventListener('open', () => {
-            ws.send(joiner.sendHello());
-        });
+export async function joinPeer(
+    peer: Peer,
+    realm: string,
+    serializer: Serializer,
+    authenticator?: ClientAuthenticator
+): Promise<BaseSession> {
 
-        return new Promise<BaseSession>((resolve, reject) => {
-            const wsMessageHandler = async (event: MessageEvent) => {
-                try {
-                    let data = event.data;
+    const joiner = new Joiner(realm, serializer, authenticator);
 
-                    if (event.data instanceof Blob) {
-                        data = new Uint8Array(await event.data.arrayBuffer());
-                    }
+    const hello = joiner.sendHello();
+    peer.send(hello as Uint8Array);
 
-                    const toSend = await joiner.receive(data);
-                    if (!toSend) {
-                        ws.removeEventListener('message', wsMessageHandler);
-                        ws.removeEventListener('close', closeHandler);
+    // eslint-disable-next-line no-constant-condition
+    while (true) {
+        const msgBytes = await peer.receive();
 
-                        const baseSession = new BaseSession(ws, wsMessageHandler, joiner.getSessionDetails(), this._serializer);
-                        resolve(baseSession);
-                    } else {
-                        ws.send(toSend);
-                    }
-                } catch (error) {
-                    reject(error);
-                }
-            };
+        const toSend = await joiner.receive(msgBytes);
 
-            const closeHandler = () => {
-                ws.removeEventListener('message', wsMessageHandler);
-                reject(new Error('Connection closed before handshake completed'));
-            };
+        if (toSend === null) {
+            return new BaseSession(
+                peer,
+                joiner.getSessionDetails(),
+                serializer
+            );
+        }
 
-            ws.addEventListener('message', wsMessageHandler);
-            ws.addEventListener('error', (error) => reject(error));
-            ws.addEventListener('close', closeHandler);
-        });
+        peer.send(toSend as Uint8Array);
     }
 }

lib/types.ts

@@ -59,125 +59,87 @@ export abstract class IBaseSession {
 }
 
 export class BaseSession extends IBaseSession {
-    private readonly _ws: WebSocket;
-    private readonly _wsMessageHandler: any;
-    private readonly sessionDetails: SessionDetails;
+    private readonly _peer: Peer;
     private readonly _serializer: Serializer;
+    private readonly _sessionDetails: SessionDetails;
+
     private _disconnectCallbacks: Array<(reason?: string) => Promise<void>> = [];
-    private _queue: any[] = [];
-    private _waiting: {resolve: (value: any) => void; reject: (reason?: any) => void;}[] = [];
-    private _wsClosed = false;
 
     constructor(
-        ws: WebSocket,
-        wsMessageHandler: any,
+        peer: Peer,
         sessionDetails: SessionDetails,
         serializer: Serializer
     ) {
         super();
-        this._ws = ws;
-        this._wsMessageHandler = wsMessageHandler;
-        this.sessionDetails = sessionDetails;
+        this._peer = peer;
         this._serializer = serializer;
-
-        this._ws.binaryType = "arraybuffer";
-        this._ws.addEventListener("message", (event: MessageEvent) => {
-            const data = event.data instanceof ArrayBuffer
-                ? new Uint8Array(event.data)
-                : event.data;
-
-            if (this._waiting.length > 0) {
-                const waiter = this._waiting.shift()!;
-                waiter.resolve(data);
-            } else {
-                this._queue.push(data);
-            }
-        });
-
-        // close cleanly on abrupt client disconnect
-        this._ws.addEventListener("close", async () => {
-            this._wsClosed = true;
-
-            while (this._waiting.length > 0) {
-                const waiter = this._waiting.shift()!;
-                waiter.reject(new SessionClosedError());
-            }
-
-            if (this._disconnectCallbacks.length > 0) {
-                await Promise.all(this._disconnectCallbacks.map(cb => cb()));
-            }
-            await this.close();
-        });
+        this._sessionDetails = sessionDetails;
+
+        if (peer.onDisconnect) {
+            peer.onDisconnect(async () => {
+                if (this._disconnectCallbacks.length > 0) {
+                    await Promise.all(
+                        this._disconnectCallbacks.map(cb => cb())
+                    );
+                }
+            });
+        }
     }
 
     id(): number {
-        return this.sessionDetails.sessionID;
+        return this._sessionDetails.sessionID;
     }
 
     realm(): string {
-        return this.sessionDetails.realm;
+        return this._sessionDetails.realm;
     }
 
     authid(): string {
-        return this.sessionDetails.authid;
+        return this._sessionDetails.authid;
     }
 
     authrole(): string {
-        return this.sessionDetails.authrole;
+        return this._sessionDetails.authrole;
     }
 
     serializer(): Serializer {
         return this._serializer;
     }
 
     send(data: any): void {
-        this._ws.send(data);
-    }
-
-    sendMessage(msg: Message): void {
-        this.send(this._serializer.serialize(msg));
+        this._peer.send(data);
     }
 
     async receive(): Promise<any> {
-        if (this._wsClosed) {
-            throw new Error("Session closed");
-        }
-
-        if (this._queue.length > 0) {
-            return this._queue.shift();
-        }
+        return this._peer.receive();
+    }
 
-        return new Promise((resolve, reject) => {
-            this._waiting.push({ resolve, reject });
-        });
+    sendMessage(msg: Message): void {
+        const bytes = this._serializer.serialize(msg);
+        this._peer.send(bytes as Uint8Array);
     }
 
     async receiveMessage(): Promise<Message> {
-        return this._serializer.deserialize(await this.receive());
+        const data = await this._peer.receive();
+        return this._serializer.deserialize(data);
     }
 
     async close(): Promise<void> {
-        if (this._wsMessageHandler) {
-            this._ws.removeEventListener("message", this._wsMessageHandler);
-            this._ws.removeEventListener("close", this._wsMessageHandler);
-        }
-
-        this._ws.close();
+        await this._peer.close();
     }
 
     isConnected(): boolean {
-        return this._ws.readyState === WebSocket.OPEN;
+        return this._peer.isConnected();
     }
 
     onDisconnect(callback: (reason?: string) => Promise<void>): void {
         this._disconnectCallbacks.push(callback);
     }
 
     getSessionDetails(): SessionDetails {
-        return this.sessionDetails;
+        return this._sessionDetails;
     }
 }
-
 export class Result {
     args: any[];
     kwargs: { [key: string]: any };
@@ -315,3 +277,105 @@ export class Progress {
 }
 
 export class SessionClosedError extends Error {}
+
+export abstract class Peer {
+    abstract send(data: Uint8Array): void;
+    abstract receive(): Promise<Uint8Array>;
+    abstract close(): Promise<void>;
+    abstract isConnected(): boolean;
+    onDisconnect?(callback: () => Promise<void>): void;
+}
+
+export class WebSocketPeer extends Peer {
+    private readonly _ws: WebSocket;
+    private _queue: Uint8Array[] = [];
+    private _waiting: {
+        resolve: (data: Uint8Array) => void,
+        reject: (err: Error) => void
+    }[] = [];
+
+    private _disconnectHandlers: (() => Promise<void>)[] = [];
+    private _closed = false;
+
+    constructor(ws: WebSocket) {
+        super();
+
+        this._ws = ws;
+        this._ws.binaryType = "arraybuffer";
+        this._bindEvents();
+    }
+
+    private _bindEvents() {
+        this._ws.addEventListener("message", (event: MessageEvent) => {
+            const data =
+                event.data instanceof ArrayBuffer
+                    ? new Uint8Array(event.data)
+                    : new Uint8Array(event.data);
+
+            if (this._waiting.length > 0) {
+                const waiter = this._waiting.shift()!;
+                waiter.resolve(data);
+            } else {
+                this._queue.push(data);
+            }
+
+        });
+
+        this._ws.addEventListener("close", () => {
+            this._handleDisconnect();
+        });
+
+        this._ws.addEventListener("error", () => {
+            this._handleDisconnect();
+        });
+    }
+
+    private async _handleDisconnect() {
+        if (this._closed) return;
+        this._closed = true;
+
+        while (this._waiting.length > 0) {
+            const waiter = this._waiting.shift();
+            waiter?.reject(new Error("WebSocket closed"));
+        }
+
+        for (const cb of this._disconnectHandlers) {
+            await cb();
+        }
+    }
+
+    onDisconnect(callback: () => Promise<void>): void {
+        this._disconnectHandlers.push(callback);
+    }
+
+    isConnected(): boolean {
+        return this._ws.readyState === WebSocket.OPEN;
+    }
+
+    send(data: Uint8Array): void {
+        this._ws.send(data);
+    }
+
+    async receive(): Promise<Uint8Array> {
+        if (this._closed) {
+            throw new Error("WebSocket closed in receive");
+        }
+
+        if (this._queue.length > 0) {
+            return this._queue.shift()!;
+        }
+
+        return new Promise((resolve, reject) => {
+            this._waiting.push({ resolve, reject });
+        });
+    }
+
+    async close(): Promise<void> {
+        if (this._closed) return;
+
+        this._closed = true;
+        this._ws.close();
+
+        await this._handleDisconnect();
+    }
+}