implement register

xconnio · Oct 5, 2024 · d4033f2 · d4033f2
1 parent 5f1a99a
commit d4033f2
Show file tree

Hide file tree

Showing 2 changed files with 79 additions and 3 deletions.
diff --git a/lib/session.ts b/lib/session.ts
@@ -2,14 +2,18 @@ import {
     WAMPSession,
     SessionScopeIDGenerator,
     Message,
-    Call,
+    Call, CallFields,
     Result as ResultMsg,
-    Error as ErrorMsg, CallFields
+    Error as ErrorMsg,
+    Register, RegisterFields,
+    Registered,
+    Yield, YieldFields,
+    Invocation as InvocationMsg
 } from "wampproto";
 
 import {wampErrorString} from "./helpers";
-import {IBaseSession, Result} from "./types";
 import {ApplicationError, ProtocolError} from "./exception";
+import {IBaseSession, Result, Invocation, RegisterRequest, Registration} from "./types";
 
 
 export class Session {
@@ -21,6 +25,8 @@ export class Session {
         resolve: (value?: Result) => void,
         reject: (reason?: ApplicationError) => void
     }> = new Map();
+    private _registerRequests: Map<number, RegisterRequest> = new Map();
+    private _registrations: Map<number, (invocation: Invocation) => Result> = new Map();
 
     constructor(baseSession: IBaseSession) {
         this._baseSession = baseSession;
@@ -46,6 +52,24 @@ export class Session {
         if (message instanceof ResultMsg) {
             const promiseHandler = this._callRequests.get(message.requestID);
             promiseHandler?.resolve(new Result(message.args, message.kwargs, message.options));
+        } else if (message instanceof Registered) {
+            const request = this._registerRequests.get(message.requestID);
+            if (request) {
+                this._registrations.set(message.registrationID, request.endpoint);
+                request.promise.resolve(new Registration(message.registrationID));
+                this._registerRequests.delete(message.requestID);
+            }
+        } else if (message instanceof InvocationMsg) {
+            const endpoint = this._registrations.get(message.registrationID);
+            if (endpoint) {
+                const result = endpoint(new Invocation(message.args, message.kwargs, message.details));
+                this._baseSession.send(this._wampSession.sendMessage(new Yield(new YieldFields(
+                    message.requestID,
+                    result.args,
+                    result.kwargs,
+                    result.details,
+                ))));
+            }
         } else if (message instanceof ErrorMsg) {
             switch (message.messageType) {
                 case Call.TYPE: {
@@ -58,6 +82,14 @@ export class Session {
                     this._callRequests.delete(message.requestID);
                     break;
                 }
+                case Register.TYPE: {
+                    const registerRequest = this._registerRequests.get(message.requestID);
+                    registerRequest?.promise.reject(
+                        new ApplicationError(message.uri, {args: message.args, kwargs: message.kwargs})
+                    );
+                    this._registerRequests.delete(message.requestID);
+                    break;
+                }
                 default:
                     throw new ProtocolError(wampErrorString(message));
             }
@@ -92,4 +124,25 @@ export class Session {
 
         return promise;
     }
+
+    async register(
+        procedure: string,
+        endpoint: (invocation: Invocation) => Result,
+        options?: { [key: string]: any } | null
+    ): Promise<Registration> {
+        const register = new Register(new RegisterFields(this._nextID, procedure, options));
+        let promiseHandler: {
+            resolve: (value: Registration | PromiseLike<Registration>) => void;
+            reject: (reason?: ApplicationError) => void
+        };
+
+        const promise = new Promise<Registration>((resolve, reject) => {
+            promiseHandler = {resolve, reject};
+        });
+
+        const request = new RegisterRequest(promiseHandler, endpoint);
+        this._registerRequests.set(register.requestID, request);
+        this._baseSession.send(this._wampSession.sendMessage(register));
+        return promise;
+    }
 }
diff --git a/lib/types.ts b/lib/types.ts
@@ -1,5 +1,7 @@
 import {Serializer, Message, SessionDetails} from "wampproto";
 
+import {ApplicationError} from "./exception";
+
 
 export abstract class IBaseSession {
     id(): number {
@@ -131,3 +133,24 @@ export class Result {
         this.details = details || {};
     }
 }
+
+export class Registration {
+    constructor(public readonly registrationID: number) {
+    }
+}
+
+export class RegisterRequest {
+    constructor(
+        public readonly promise: { resolve: (value?: Registration) => void, reject: (reason?: ApplicationError) => void },
+        public readonly endpoint: (invocation: Invocation) => Result) {
+    }
+}
+
+export class Invocation {
+    constructor(
+        public readonly args: any[] = [],
+        public readonly kwargs: { [key: string]: any } = {},
+        public readonly details: { [key: string]: any }
+    ) {
+    }
+}