diff --git a/src/actors/errors.ts b/src/actors/errors.ts new file mode 100644 index 0000000..b50ca7e --- /dev/null +++ b/src/actors/errors.ts @@ -0,0 +1,9 @@ +export type ErrorCode = + | "NOT_FOUND" + | "METHOD_NOT_FOUND" + | "METHOD_NOT_INVOCABLE"; +export class ActorError extends Error { + constructor(msg: string, public code: ErrorCode, options?: ErrorOptions) { + super(msg, options); + } +} diff --git a/src/actors/proxy.ts b/src/actors/proxy.ts index b729e1b..8626caf 100644 --- a/src/actors/proxy.ts +++ b/src/actors/proxy.ts @@ -19,6 +19,26 @@ export type Fulfilled = ((result: R) => T | PromiseLike) | null; // deno-lint-ignore no-explicit-any export type Rejected = ((reason: any) => E | PromiseLike) | null; +export interface ActorInvoker< + // deno-lint-ignore no-explicit-any + TResponse = any, + TChannel extends DuplexChannel = DuplexChannel< + unknown, + unknown + >, +> { + invoke( + name: string, + method: string, + methodArgs: unknown[], + ): Promise; + + connect( + name: string, + method: string, + methodArgs: unknown[], + ): Promise; +} export class ActorAwaiter< TResponse, TChannel extends DuplexChannel, @@ -29,10 +49,10 @@ export class ActorAwaiter< DuplexChannel { ch: Promise | null = null; constructor( - protected fetcher: () => Promise< - TResponse - >, - protected ws: () => Promise, + protected actorName: string, + protected actorMethod: string, + protected methodArgs: unknown[], + protected invoker: ActorInvoker, ) { } async close() { @@ -46,7 +66,11 @@ export class ActorAwaiter< } private get channel(): Promise { - return this.ch ??= this.ws(); + return this.ch ??= this.invoker.connect( + this.actorName, + this.actorMethod, + this.methodArgs, + ); } async send(value: unknown): Promise { @@ -55,7 +79,12 @@ export class ActorAwaiter< } catch(onrejected: Rejected): Promise { - return this.fetcher().catch(onrejected); + return this.invoker.invoke( + this.actorName, + this.actorMethod, + this.methodArgs, + ) + .catch(onrejected); } then( @@ -65,7 +94,11 @@ export class ActorAwaiter< >, onrejected?: Rejected, ): Promise { - return this.fetcher().then(onfufilled).catch( + return this.invoker.invoke( + this.actorName, + this.actorMethod, + this.methodArgs, + ).then(onfufilled).catch( onrejected, ); } @@ -122,6 +155,74 @@ const initServer = (): ActorsServer => { }; }; +const urlFor = ( + serverUrl: string, + actor: ActorConstructor | string, + actorMethod: string, +): string => { + return `${serverUrl}/actors/${ + typeof actor === "string" ? actor : actor.name + }/invoke/${actorMethod}`; +}; + +const createHttpInvoker = < + TResponse, + TChannel extends DuplexChannel, +>( + actorId: string, + server: ActorsServer, +): ActorInvoker => { + return { + connect: (name, method, methodArgs) => { + const endpoint = urlFor(server.url, name, method); + const ws = new WebSocket( + `${endpoint}?args=${ + encodeURIComponent( + btoa(JSON.stringify({ args: methodArgs ?? [] })), + ) + }&${ACTOR_ID_QS_NAME}=${actorId}`, + ); + return makeWebSocket(ws) as Promise; + }, + invoke: async (name, method, methodArgs) => { + const endpoint = urlFor(server.url, name, method); + const abortCtrl = new AbortController(); + const resp = await fetch( + endpoint, + { + method: "POST", + signal: abortCtrl.signal, + headers: { + "Content-Type": "application/json", + [ACTOR_ID_HEADER_NAME]: actorId, + ...server.deploymentId + ? { ["x-deno-deployment-id"]: server.deploymentId } + : {}, + }, + body: JSON.stringify({ + args: methodArgs ?? [], + }), + }, + ); + if ( + resp.headers.get("content-type") === + EVENT_STREAM_RESPONSE_HEADER + ) { + const iterator = readFromStream(resp); + const retn = iterator.return; + iterator.return = function (val) { + abortCtrl.abort(); + return retn?.call(iterator, val) ?? val; + }; + return iterator; + } + if (resp.status === 204) { + return; + } + return resp.json(); + }, + }; +}; /** * utilities to create and manage actors. */ @@ -134,60 +235,19 @@ export const actors = { _server ??= initServer(); } const actorsServer = server ?? _server!; + const name = typeof actor === "string" ? actor : actor.name; return { id: (id: string): Promisify => { return new Proxy>({} as Promisify, { - get: (_, prop) => { - const endpoint = `${actorsServer.url}/actors/${ - typeof actor === "string" ? actor : actor.name - }/invoke/${String(prop)}`; - const fetcher = async (...args: unknown[]) => { - const abortCtrl = new AbortController(); - const resp = await fetch( - endpoint, - { - method: "POST", - signal: abortCtrl.signal, - headers: { - "Content-Type": "application/json", - [ACTOR_ID_HEADER_NAME]: id, - ...actorsServer.deploymentId - ? { ["x-deno-deployment-id"]: actorsServer.deploymentId } - : {}, - }, - body: JSON.stringify({ - args: args ?? [], - }), - }, - ); - if ( - resp.headers.get("content-type") === - EVENT_STREAM_RESPONSE_HEADER - ) { - const iterator = readFromStream(resp); - const retn = iterator.return; - iterator.return = function (val) { - abortCtrl.abort(); - return retn?.call(iterator, val) ?? val; - }; - return iterator; - } - if (resp.status === 204) { - return; - } - return resp.json(); - }; + get: (_, method) => { + const invoker = createHttpInvoker(id, actorsServer); return (...args: unknown[]) => { - const awaiter = new ActorAwaiter(() => fetcher(...args), () => { - const ws = new WebSocket( - `${endpoint}?args=${ - encodeURIComponent( - btoa(JSON.stringify({ args: args ?? [] })), - ) - }&${ACTOR_ID_QS_NAME}=${id}`, - ); - return makeWebSocket(ws); - }); + const awaiter = new ActorAwaiter( + name, + String(method), + args, + invoker, + ); return awaiter; }; }, diff --git a/src/actors/runtime.ts b/src/actors/runtime.ts index bf3f5e1..4d21f11 100644 --- a/src/actors/runtime.ts +++ b/src/actors/runtime.ts @@ -1,5 +1,10 @@ import { type ServerSentEventMessage, ServerSentEventStream } from "@std/http"; -import { ACTOR_ID_HEADER_NAME, ACTOR_ID_QS_NAME } from "./proxy.ts"; +import { ActorError } from "./errors.ts"; +import { + ACTOR_ID_HEADER_NAME, + ACTOR_ID_QS_NAME, + type ActorInvoker, +} from "./proxy.ts"; import { ActorState } from "./state.ts"; import type { ActorStorage } from "./storage.ts"; import { DenoKvActorStorage } from "./storage/denoKv.ts"; @@ -53,7 +58,7 @@ export type ActorConstructor = new ( /** * Represents an actor invoker. */ -export interface ActorInvoker { +export interface ActorInstance { /** * The actor instance. */ @@ -72,8 +77,9 @@ export interface ActorInvoker { * Represents the runtime for managing and invoking actors. */ export class ActorRuntime { - private actors: Map = new Map(); + private actors: Map = new Map(); private initilized = false; + private invoker: ActorInvoker; /** * Creates an instance of ActorRuntime. * @param actorsConstructors - An array of actor constructors. @@ -81,6 +87,34 @@ export class ActorRuntime { constructor( protected actorsConstructors: Array, ) { + const invoke: ActorInvoker["invoke"] = async (actorName, method, args) => { + const actorInvoker = actorName ? this.actors.get(actorName) : undefined; + if (!actorInvoker) { + throw new ActorError(`actor ${actorName} not found`, "NOT_FOUND"); + } + const { actor, initialization } = actorInvoker; + if (!(method in actor)) { + throw new ActorError( + `actor ${actorName} not found`, + "METHOD_NOT_FOUND", + ); + } + const methodImpl = actor[method as keyof typeof actor]; + if (!isInvocable(methodImpl)) { + throw new ActorError( + `actor ${actorName} not found`, + "METHOD_NOT_INVOCABLE", + ); + } + await initialization; + return await (methodImpl as Function).bind(actor)( + ...Array.isArray(args) ? args : [args], + ); + }; + this.invoker = { + invoke, + connect: invoke, + }; } getActorStorage(actorId: string, actorName: string): ActorStorage { @@ -147,16 +181,12 @@ export class ActorRuntime { } const groups = result?.pathname.groups ?? {}; const actorName = groups[ACTOR_NAME_PATH_PARAM]; - const actorInvoker = actorName ? this.actors.get(actorName) : undefined; - if (!actorInvoker) { - return new Response(`actor ${actorName} not found`, { - status: 404, - }); - } - const { actor, initialization } = actorInvoker; const method = groups[METHOD_NAME_PATH_PARAM]; - if (!method || !(method in actor)) { - return new Response(`method not found for the actor`, { status: 404 }); + if (!method || !actorName) { + return new Response( + `method ${method} not found for the actor ${actorName}`, + { status: 404 }, + ); } let args = []; if (req.headers.get("content-type")?.includes("application/json")) { @@ -169,55 +199,58 @@ export class ActorRuntime { : {}; args = parsedArgs.args; } - const methodImpl = actor[method as keyof typeof actor]; - if (!isInvocable(methodImpl)) { - return new Response( - `cannot invoke actor method for type ${typeof methodImpl}`, - { - status: 400, - }, - ); - } - await initialization; - const res = await (methodImpl as Function).bind(actor)( - ...Array.isArray(args) ? args : [args], - ); - if (req.headers.get("upgrade") === "websocket" && isUpgrade(res)) { - const { socket, response } = Deno.upgradeWebSocket(req); - makeWebSocket(socket).then((ch) => res(ch)).finally(() => socket.close()); - return response; - } - if (isEventStreamResponse(res)) { - req.signal.onabort = () => { - res?.return?.(); - }; + try { + const res = await this.invoker.invoke(actorName, method, args); + if (req.headers.get("upgrade") === "websocket" && isUpgrade(res)) { + const { socket, response } = Deno.upgradeWebSocket(req); + makeWebSocket(socket).then((ch) => res(ch)).finally(() => + socket.close() + ); + return response; + } + if (isEventStreamResponse(res)) { + req.signal.onabort = () => { + res?.return?.(); + }; - return new Response( - new ReadableStream({ - async pull(controller) { - for await (const content of res) { - controller.enqueue({ - data: encodeURIComponent(JSON.stringify(content)), - id: Date.now(), - event: "message", - }); - } - controller.close(); - }, - cancel() { - res?.return?.(); + return new Response( + new ReadableStream({ + async pull(controller) { + for await (const content of res) { + controller.enqueue({ + data: encodeURIComponent(JSON.stringify(content)), + id: Date.now(), + event: "message", + }); + } + controller.close(); + }, + cancel() { + res?.return?.(); + }, + }).pipeThrough(new ServerSentEventStream()), + { + headers: { + "Content-Type": EVENT_STREAM_RESPONSE_HEADER, + }, }, - }).pipeThrough(new ServerSentEventStream()), - { - headers: { - "Content-Type": EVENT_STREAM_RESPONSE_HEADER, - }, - }, - ); - } - if (typeof res === "undefined") { - return new Response(null, { status: 204 }); + ); + } + if (typeof res === "undefined") { + return new Response(null, { status: 204 }); + } + return Response.json(res); + } catch (err) { + if (err instanceof ActorError) { + return new Response(err.message, { + status: { + METHOD_NOT_FOUND: 404, + METHOD_NOT_INVOCABLE: 405, + NOT_FOUND: 404, + }[err.code] ?? 500, + }); + } + throw err; } - return Response.json(res); } }