feat: support transport negotiation between client and server

Author: NathanFlurry

packages/actor-core/src/client/client.ts

@@ -19,7 +19,7 @@ import { importEventSource } from "@/common/eventsource";
  */
 export interface ClientOptions {
 	encoding?: Encoding;
-	transport?: Transport;
+	supportedTransports?: Transport[];
 }
 
 /**
@@ -135,7 +135,7 @@ export class Client {
 	#managerEndpointPromise: Promise<string>;
 	//#regionPromise: Promise<Region | undefined>;
 	#encodingKind: Encoding;
-	#transportKind: Transport;
+	#supportedTransports: Transport[];
 
 	// External imports
 	#dynamicImportsPromise: Promise<DynamicImports>;
@@ -164,7 +164,7 @@ export class Client {
 		//this.#regionPromise = this.#fetchRegion();
 
 		this.#encodingKind = opts?.encoding ?? "cbor";
-		this.#transportKind = opts?.transport ?? "websocket";
+		this.#supportedTransports = opts?.supportedTransports ?? ["websocket", "sse"];
 
 		// Import dynamic dependencies
 		this.#dynamicImportsPromise = (async () => {
@@ -198,10 +198,14 @@ export class Client {
 				getForId: {
 					actorId,
 				},
-			},
+			}
 		});
 
-		const handle = await this.#createHandle(resJson.endpoint, opts?.parameters);
+		const handle = await this.#createHandle(
+			resJson.endpoint,
+			opts?.parameters,
+			resJson.supportedTransports
+		);
 		return this.#createProxy(handle) as ActorHandle<A>;
 	}
 
@@ -259,10 +263,14 @@ export class Client {
 					tags,
 					create,
 				},
-			},
+			}
 		});
 
-		const handle = await this.#createHandle(resJson.endpoint, opts?.parameters);
+		const handle = await this.#createHandle(
+			resJson.endpoint,
+			opts?.parameters,
+			resJson.supportedTransports
+		);
 		return this.#createProxy(handle) as ActorHandle<A>;
 	}
 
@@ -307,24 +315,30 @@ export class Client {
 		>("POST", "/manager/actors", {
 			query: {
 				create,
-			},
+			}
 		});
 
-		const handle = await this.#createHandle(resJson.endpoint, opts?.parameters);
+		const handle = await this.#createHandle(
+			resJson.endpoint,
+			opts?.parameters,
+			resJson.supportedTransports
+		);
 		return this.#createProxy(handle) as ActorHandle<A>;
 	}
 
 	async #createHandle(
 		endpoint: string,
 		parameters: unknown,
+		serverTransports: Transport[],
 	): Promise<ActorHandleRaw> {
 		const imports = await this.#dynamicImportsPromise;
 
 		const handle = new ActorHandleRaw(
 			endpoint,
 			parameters,
 			this.#encodingKind,
-			this.#transportKind,
+			this.#supportedTransports,
+			serverTransports,
 			imports,
 		);
 		handle.__connect();

packages/actor-core/src/client/errors.ts

@@ -24,6 +24,12 @@ export class MalformedResponseMessage extends ActorClientError {
 	}
 }
 
+export class NoSupportedTransport extends ActorClientError {
+	constructor() {
+		super("No supported transport available between client and server");
+	}
+}
+
 export class RpcError extends ActorClientError {
 	constructor(
 		public readonly code: string,

packages/actor-core/src/client/handle.ts

@@ -93,7 +93,8 @@ export class ActorHandleRaw {
 		private readonly endpoint: string,
 		private readonly parameters: unknown,
 		private readonly encodingKind: Encoding,
-		private readonly transportKind: Transport,
+		private readonly supportedTransports: Transport[],
+		private readonly serverTransports: Transport[],
 		private readonly dynamicImports: DynamicImports,
 	) {
 		this.#keepNodeAliveInterval = setInterval(() => 60_000);
@@ -205,12 +206,13 @@ enc
 			this.#onOpenPromise = Promise.withResolvers();
 
 			// Connect transport
-			if (this.transportKind === "websocket") {
+			const transport = this.#pickTransport();
+			if (transport === "websocket") {
 				this.#connectWebSocket();
-			} else if (this.transportKind === "sse") {
+			} else if (transport === "sse") {
 				this.#connectSse();
 			} else {
-				assertUnreachable(this.transportKind);
+				assertUnreachable(transport);
 			}
 
 			// Wait for result
@@ -220,10 +222,23 @@ enc
 		}
 	}
 
+	#pickTransport(): Transport {
+		// Choose first supported transport from server's list that client also supports
+		const transport = this.serverTransports.find(t => 
+			this.supportedTransports.includes(t)
+		);
+		
+		if (!transport) {
+			throw new errors.NoSupportedTransport();
+		}
+
+		return transport;
+	}
+
 	#connectWebSocket() {
 		const { WebSocket } = this.dynamicImports;
 
-		const url = this.#buildConnectionUrl();
+		const url = this.#buildConnectionUrl("websocket");
 
 		const ws = new WebSocket(url);
 		if (this.encodingKind === "cbor") {
@@ -255,7 +270,7 @@ enc
 	#connectSse() {
 		const { EventSource } = this.dynamicImports;
 
-		const url = this.#buildConnectionUrl();
+		const url = this.#buildConnectionUrl("sse");
 
 		const eventSource = new EventSource(url);
 		this.#transport = { sse: eventSource };
@@ -391,8 +406,8 @@ enc
 		logger().warn("socket error", { event });
 	}
 
-	#buildConnectionUrl(): string {
-		let url = `${this.endpoint}/connect/${this.transportKind}?encoding=${this.encodingKind}`;
+	#buildConnectionUrl(transport: Transport): string {
+		let url = `${this.endpoint}/connect/${transport}?encoding=${this.encodingKind}`;
 
 		if (this.parameters !== undefined) {
 			const paramsStr = JSON.stringify(this.parameters);
@@ -593,7 +608,10 @@ enc
 			}
 			return JSON.parse(data);
 		} else if (this.encodingKind === "cbor") {
-			if (this.transportKind === "sse") {
+			if (!this.#transport) {
+				// Do thing
+				throw new Error("Cannot parse message when no transport defined");
+			} else if ("sse" in this.#transport) {
 				// Decode base64 since SSE sends raw strings
 				if (typeof data === "string") {
 					const binaryString = atob(data);
@@ -605,10 +623,10 @@ enc
 						`Expected data to be a string for SSE, got ${data}.`,
 					);
 				}
-			} else if (this.transportKind === "websocket") {
+			} else if ("websocket" in this.#transport) {
 				// Do nothing
 			} else {
-				assertUnreachable(this.transportKind);
+				assertUnreachable(this.#transport);
 			}
 
 			// Decode data

packages/actor-core/src/manager/protocol/mod.ts

@@ -1,5 +1,6 @@
 import { z } from "zod";
 import { ActorQuerySchema } from "./query";
+import { TransportSchema } from "@/actor/protocol/message/mod";
 export * from "./query";
 
 export const ActorsRequestSchema = z.object({
@@ -8,6 +9,7 @@ export const ActorsRequestSchema = z.object({
 
 export const ActorsResponseSchema = z.object({
 	endpoint: z.string(),
+	supportedTransports: z.array(TransportSchema),
 });
 
 //export const RivetConfigResponseSchema = z.object({

packages/platforms/cloudflare-workers/src/manager.ts

@@ -36,7 +36,10 @@ export function buildManager(env: Env): ManagerDriver {
 				//
 				//return res.actor;
 
-				return { endpoint: buildActorEndpoint(origin, query.getForId.actorId) };
+				return { 
+					endpoint: buildActorEndpoint(origin, query.getForId.actorId),
+					supportedTransports: ["websocket", "sse"]
+				};
 			}
 			if ("getOrCreateForTags" in query) {
 				const tags = query.getOrCreateForTags.tags;
@@ -78,7 +81,10 @@ async function getWithTags(
 		`actor:tags:${JSON.stringify(tags)}:id`,
 	);
 	if (actorId) {
-		return { endpoint: buildActorEndpoint(origin, actorId) };
+		return { 
+			endpoint: buildActorEndpoint(origin, actorId),
+			supportedTransports: ["websocket", "sse"]
+		};
 	}
 	return undefined;
 
@@ -131,5 +137,8 @@ async function createActor(
 		actorId.toString(),
 	);
 
-	return { endpoint: buildActorEndpoint(origin, actorId.toString()) };
+	return { 
+		endpoint: buildActorEndpoint(origin, actorId.toString()),
+		supportedTransports: ["websocket", "sse"]
+	};
 }

packages/platforms/redis/src/manager.ts

@@ -37,7 +37,10 @@ export function buildManager(redis: Redis): ManagerDriver {
 				//
 				//return res.actor;
 
-				return { endpoint: buildActorEndpoint(origin, query.getForId.actorId) };
+				return { 
+					endpoint: buildActorEndpoint(origin, query.getForId.actorId),
+					supportedTransports: ["websocket", "sse"]
+				};
 			}
 			if ("getOrCreateForTags" in query) {
 				const tags = query.getOrCreateForTags.tags;
@@ -81,7 +84,10 @@ async function getWithTags(
 
 	const actorId = await redis.get(`actor_tags:${JSON.stringify(tags)}:id`);
 	if (actorId) {
-		return { endpoint: buildActorEndpoint(origin, actorId) };
+		return { 
+			endpoint: buildActorEndpoint(origin, actorId),
+			supportedTransports: ["websocket", "sse"]
+		};
 	}
 	return undefined;
 
@@ -126,5 +132,8 @@ async function createActor(
 		[`actor_tags:${JSON.stringify(createRequest.tags)}:id`]: actorId,
 	});
 
-	return { endpoint: buildActorEndpoint(origin, actorId.toString()) };
+	return { 
+		endpoint: buildActorEndpoint(origin, actorId.toString()),
+		supportedTransports: ["websocket", "sse"]
+	};
 }

packages/platforms/rivet/src/manager.ts

@@ -68,7 +68,10 @@ export function buildManager(clientConfig: RivetClientConfig): ManagerDriver {
 					);
 				}
 
-				return { endpoint: buildActorEndpoint(res.actor) };
+				return { 
+					endpoint: buildActorEndpoint(res.actor),
+					supportedTransports: ["websocket", "sse"]
+				};
 			}
 			if ("getOrCreateForTags" in query) {
 				const tags = query.getOrCreateForTags.tags;
@@ -137,7 +140,10 @@ async function getWithTags(
 		actors.sort((a: RivetActor, b: RivetActor) => a.id.localeCompare(b.id));
 	}
 
-	return { endpoint: buildActorEndpoint(actors[0]) };
+	return { 
+		endpoint: buildActorEndpoint(actors[0]),
+		supportedTransports: ["websocket", "sse"]
+	};
 }
 
 async function createActor(
@@ -177,7 +183,10 @@ async function createActor(
 		req,
 	);
 
-	return { endpoint: buildActorEndpoint(actor) };
+	return { 
+		endpoint: buildActorEndpoint(actor),
+		supportedTransports: ["websocket", "sse"]
+	};
 }
 
 async function getBuildWithTags(