Support actor discriminator

Signed-off-by: Marcos Candeia <[email protected]>
deco-cx · Dec 1, 2024 · 047453b · 047453b
1 parent 40d8443
commit 047453b
Show file tree

Hide file tree

Showing 6 changed files with 63 additions and 14 deletions.
diff --git a/src/actors/proxy.ts b/src/actors/proxy.ts
@@ -1,5 +1,10 @@
 // deno-lint-ignore-file no-explicit-any
-import { type ActorProxy, create, createHttpInvoker } from "./proxyutil.ts";
+import {
+  type ActorProxy,
+  create,
+  createHttpInvoker,
+  type ProxyFactory,
+} from "./proxyutil.ts";
 import type { Actor, ActorConstructor } from "./runtime.ts";
 export type { ActorProxy };
 export interface ActorsServer {
@@ -20,8 +25,9 @@ export const actors = {
   proxy: <TInstance extends Actor>(
     actor: ActorConstructor<TInstance> | string,
     options?: ActorsOptions | undefined,
-  ): { id: (id: string) => ActorProxy<TInstance> } => {
-    const factory = (id: string) => createHttpInvoker(id, options);
+  ): { id: ProxyFactory<TInstance> } => {
+    const factory = (id: string, discriminator?: string) =>
+      createHttpInvoker(id, discriminator, options);
     return create(actor, factory);
   },
 };
diff --git a/src/actors/proxyutil.ts b/src/actors/proxyutil.ts
@@ -17,6 +17,13 @@ import { retry } from "./util/retry.ts";
 export const ACTOR_ID_HEADER_NAME = "x-deno-isolate-instance-id";
 export const ACTOR_ID_QS_NAME = "deno_isolate_instance_id";
 export const ACTOR_CONSTRUCTOR_NAME_HEADER = "x-error-constructor-name";
+export const ACTOR_DISCRIMINATOR_HEADER_NAME = "x-actor-discriminator";
+export const ACTOR_DISCRIMINATOR_QS_NAME = "actor_discriminator";
+
+export type ProxyFactory<TInstance> = (
+  id: string,
+  discriminator?: string,
+) => ActorProxy<TInstance>;
 /**
  * Promise.prototype.then onfufilled callback type.
  */
@@ -325,6 +332,7 @@ export const createHttpInvoker = <
   TChannel extends DuplexChannel<unknown, unknown>,
 >(
   actorId: string,
+  discriminator?: string,
   options?: ActorsOptions,
 ): ActorInvoker<TResponse, TChannel> => {
   const server = options?.server;
@@ -345,7 +353,11 @@ export const createHttpInvoker = <
               }),
             ),
           )
-        }&${ACTOR_ID_QS_NAME}=${actorId}`);
+        }&${ACTOR_ID_QS_NAME}=${actorId}${
+          discriminator
+            ? `&${ACTOR_DISCRIMINATOR_QS_NAME}=${discriminator}`
+            : ""
+        }`);
         url.protocol = url.protocol === "http:" ? "ws:" : "wss:";
         const ws = new WebSocket(
           url,
@@ -362,6 +374,9 @@ export const createHttpInvoker = <
           headers: {
             "Content-Type": "application/json",
             [options?.actorIdHeaderName ?? ACTOR_ID_HEADER_NAME]: actorId,
+            ...discriminator
+              ? { [ACTOR_DISCRIMINATOR_HEADER_NAME]: discriminator }
+              : {},
             ...actorsServer.deploymentId
               ? { ["x-deno-deployment-id"]: actorsServer.deploymentId }
               : {},
@@ -425,22 +440,23 @@ export const createHttpInvoker = <
 export const WELL_KNOWN_RPC_MEHTOD = "_rpc";
 export const create = <TInstance extends Actor>(
   actor: ActorConstructor<TInstance> | string,
-  invokerFactory: (id: string) => ActorInvoker,
+  invokerFactory: (id: string, discriminator?: string) => ActorInvoker,
   metadata?: unknown,
   disposer?: () => void,
-): { id: (id: string) => ActorProxy<TInstance> } => {
+): { id: ProxyFactory<TInstance> } => {
   const name = typeof actor === "string" ? actor : actor.name;
   return {
-    id: (id: string): ActorProxy<TInstance> => {
+    id: (id: string, discriminator?: string): ActorProxy<TInstance> => {
       return new Proxy<ActorProxy<TInstance>>({} as ActorProxy<TInstance>, {
         get: (_, method) => {
           if (method === "withMetadata") {
-            return (m: unknown) => create(actor, invokerFactory, m).id(id);
+            return (m: unknown) =>
+              create(actor, invokerFactory, m).id(id, discriminator);
           }
           if (method === Symbol.dispose && disposer) {
             return disposer;
           }
-          const invoker = invokerFactory(id);
+          const invoker = invokerFactory(id, discriminator);
           if (method === "rpc") {
             return () => {
               const awaiter = new ActorAwaiter(
@@ -458,6 +474,7 @@ export const create = <TInstance extends Actor>(
                 () => awaiter.close(),
               ).id(
                 id,
+                discriminator,
               );
             };
           }

diff --git a/src/actors/runtime.test.ts b/src/actors/runtime.test.ts
@@ -103,7 +103,7 @@ Deno.test("counter tests", async () => {
   const counterProxy = actors.proxy(Counter);
 
   const counterActor = counterProxy.id(actorId);
-  using rpcActor = counterProxy.id("12345").rpc();
+  using rpcActor = counterProxy.id("12345", "1234").rpc();
 
   const name = `Counter Actor`;
   const ch = counterActor.chan(name);

diff --git a/src/actors/runtime.ts b/src/actors/runtime.ts
@@ -2,6 +2,8 @@ import process from "node:process";
 import { ActorError } from "./errors.ts";
 import {
   ACTOR_CONSTRUCTOR_NAME_HEADER,
+  ACTOR_DISCRIMINATOR_HEADER_NAME,
+  ACTOR_DISCRIMINATOR_QS_NAME,
   ACTOR_ID_HEADER_NAME,
   ACTOR_ID_QS_NAME,
 } from "./proxyutil.ts";
@@ -10,11 +12,11 @@ import type { ActorState } from "./state.ts";
 import type { ActorStorage } from "./storage.ts";
 import { DenoKvActorStorage } from "./storage/denoKv.ts";
 import { S3ActorStorage } from "./storage/s3.ts";
-import { serializeUint8Array } from "./util/buffers.ts";
 import {
   EVENT_STREAM_RESPONSE_HEADER,
   isEventStreamResponse,
 } from "./stream.ts";
+import { serializeUint8Array } from "./util/buffers.ts";
 import { isUpgrade, makeWebSocket } from "./util/channels/channel.ts";
 import {
   type ServerSentEventMessage,
@@ -88,13 +90,16 @@ export class ActorRuntime {
     });
   }
 
-  private getOrCreateSilo(actorId: string): ActorSilo {
-    let silo = this.silos.get(actorId);
+  private getOrCreateSilo(actorId: string, discriminator?: string): ActorSilo {
+    let silo = this.silos.get(
+      discriminator ? `${actorId}#${discriminator}` : actorId,
+    );
     if (!silo) {
       silo = new ActorSilo(
         this.actorsConstructors,
         actorId,
         this.getActorStorage.bind(this),
+        discriminator,
       );
       this.silos.set(actorId, silo);
     }
@@ -119,6 +124,19 @@ export class ActorRuntime {
     return null;
   }
 
+  actorDiscriminator(req: Request): string | null;
+  actorDiscriminator(url: URL, req: Request): string | null;
+  actorDiscriminator(reqOrUrl: URL | Request, req?: Request): string | null {
+    if (reqOrUrl instanceof Request) {
+      return this.actorDiscriminator(new URL(reqOrUrl.url), reqOrUrl);
+    }
+    if (reqOrUrl instanceof URL && req instanceof Request) {
+      return req.headers.get(ACTOR_DISCRIMINATOR_HEADER_NAME) ??
+        reqOrUrl.searchParams.get(ACTOR_DISCRIMINATOR_QS_NAME);
+    }
+    return null;
+  }
+
   /**
    * Handles an incoming request and invokes the corresponding actor method.
    * @param req - The incoming request.
@@ -133,7 +151,10 @@ export class ActorRuntime {
       });
     }
 
-    const silo = this.getOrCreateSilo(actorId);
+    const silo = this.getOrCreateSilo(
+      actorId,
+      this.actorDiscriminator(url, req) || undefined,
+    );
 
     const result = parseActorInvokeApi(url.pathname);
     if (!result) {

diff --git a/src/actors/silo.ts b/src/actors/silo.ts
@@ -35,6 +35,7 @@ export class ActorSilo {
       actorId: string,
       actorName: string,
     ) => ActorStorage,
+    private discriminator?: string,
   ) {
     this.invoker = {
       invoke: this.invoke.bind(this),
@@ -47,6 +48,7 @@ export class ActorSilo {
       const storage = this.getActorStorage(this.actorId, Actor.name);
       const state = new ActorState({
         id: this.actorId,
+        discriminator: this.discriminator,
         storage,
         proxy: (actor) => {
           const invoker = (id: string) => {

diff --git a/src/actors/state.ts b/src/actors/state.ts
@@ -4,6 +4,7 @@ import type { ActorStorage } from "./storage.ts";
 
 export interface ActorStateOptions {
   id: string;
+  discriminator?: string;
   storage: ActorStorage;
   proxy: <TInstance extends Actor>(
     actor: ActorConstructor<TInstance>,
@@ -14,6 +15,7 @@ export interface ActorStateOptions {
  */
 export class ActorState {
   public id: string;
+  public discriminator?: string;
   public storage: ActorStorage;
   public proxy: <TInstance extends Actor>(
     actor: ActorConstructor<TInstance>,
@@ -23,6 +25,7 @@ export class ActorState {
     this.storage = options.storage;
     this.proxy = options.proxy;
     this.id = options.id;
+    this.discriminator = options.discriminator;
   }
 
   blockConcurrencyWhile<T>(callback: () => Promise<T>): Promise<T> {