import { decodeStream, Deferred, deferred, Disposable, encode, io,} from "./deps.ts";import * as message from "./message.ts";import { Indexer } from "./indexer.ts";import { ResponseWaiter } from "./response_waiter.ts";
const MSGID_THRESHOLD = 2 ** 32;
export interface Dispatcher { [key: string]: (this: Session, ...args: unknown[]) => Promise<unknown>;}
export type DispatcherFrom<T> = { [K in keyof T]: T[K] extends (...args: infer Args) => unknown ? (...args: { [K in keyof Args]: unknown }) => Promise<unknown> : never;};
export type SessionOptions = { responseTimeout?: number;
errorCallback?: (e: Error) => void;};
export class Session implements Disposable { #indexer: Indexer; #waiter: ResponseWaiter; #reader: Deno.Reader & Deno.Closer; #writer: Deno.Writer; #listener: Promise<void>; #closed: boolean; #closedSignal: Deferred<never>;
dispatcher: Dispatcher;
constructor( reader: Deno.Reader & Deno.Closer, writer: Deno.Writer, dispatcher: Dispatcher = {}, options: SessionOptions = {}, ) { this.dispatcher = dispatcher; this.#indexer = new Indexer(MSGID_THRESHOLD); this.#waiter = new ResponseWaiter(options.responseTimeout); this.#reader = reader; this.#writer = writer; this.#closed = false; this.#closedSignal = deferred(); this.#listener = this.listen().catch((e) => { if (options.errorCallback) { options.errorCallback(e); } else { console.error(`Unexpected error occured in session: ${e}`); } }); }
private async send(data: Uint8Array): Promise<void> { await io.writeAll(this.#writer, data); }
private async dispatch( method: string, ...params: unknown[] ): Promise<unknown> { if (!Object.prototype.hasOwnProperty.call(this.dispatcher, method)) { const propertyNames = Object.getOwnPropertyNames(this.dispatcher); throw new Error( `No method '${method}' exists in ${JSON.stringify(propertyNames)}`, ); } return await this.dispatcher[method].apply(this, params); }
private async handleRequest(request: message.RequestMessage): Promise<void> { const [_, msgid, method, params] = request; const [result, error] = await (async () => { let result: message.MessageResult = null; let error: message.MessageError = null; try { result = await this.dispatch(method, ...params); } catch (e) { error = e.stack ?? e.toString(); } return [result, error]; })(); const response: message.ResponseMessage = [1, msgid, error, result]; await this.send(encode(response)); }
private handleResponse(response: message.ResponseMessage): void { if (!this.#waiter.provide(response)) { console.warn("Unexpected response message received", response); } }
private async handleNotification( notification: message.NotificationMessage, ): Promise<void> { const [_, method, params] = notification; try { await this.dispatch(method, ...params); } catch (e) { console.error(e); } }
private async listen(): Promise<void> { const iter = decodeStream(io.iter(this.#reader)); try { while (!this.#closed) { const { done, value } = await Promise.race([ this.#closedSignal, iter.next(), ]); if (done) { return; } if (message.isRequestMessage(value)) { this.handleRequest(value); } else if (message.isResponseMessage(value)) { this.handleResponse(value); } else if (message.isNotificationMessage(value)) { this.handleNotification(value); } else { console.warn(`Unexpected data received: ${value}`); continue; } } } catch (e) { if (e instanceof SessionClosedError) { return; } if (e instanceof Deno.errors.BadResource) { return; } throw e; } }
dispose() { this.close(); }
close(): void { this.#closed = true; this.#closedSignal.reject(new SessionClosedError()); }
waitClosed(): Promise<void> { return this.#listener; }
async call(method: string, ...params: unknown[]): Promise<unknown> { if (this.#closed) { throw new SessionClosedError(); } const msgid = this.#indexer.next(); const data: message.RequestMessage = [0, msgid, method, params]; const [_, response] = await Promise.race([ this.#closedSignal, Promise.all([this.send(encode(data)), this.#waiter.wait(msgid)]), ]); const [err, result] = response.slice(2); if (err) { const paramsStr = JSON.stringify(params); const errStr = typeof err === "string" ? err : JSON.stringify(err); throw new Error( `Failed to call '${method}' with ${paramsStr}: ${errStr}`, ); } return result; }
async notify(method: string, ...params: unknown[]): Promise<void> { if (this.#closed) { throw new SessionClosedError(); } const data: message.NotificationMessage = [2, method, params]; await Promise.race([this.#closedSignal, this.send(encode(data))]); }
clearDispatcher(): void { this.dispatcher = {}; }
extendDispatcher(dispatcher: Dispatcher): void { this.dispatcher = { ...this.dispatcher, ...dispatcher, }; }}
export class SessionClosedError extends Error { constructor() { super("The session is closed"); this.name = "SessionClosedError"; }}