Connect

// Open a connection by actor name. The selector query parameters are the same
// ones used by the HTTP endpoints; the token and encoding travel as
// WebSocket subprotocols because browsers cannot set headers on an upgrade.
const url = new URL("wss://api.rivet.dev/gateway/counter/connect");
url.searchParams.set("rvt-namespace", process.env.RIVET_NAMESPACE!);
url.searchParams.set("rvt-method", "getOrCreate");
url.searchParams.set("rvt-key", "my-counter");
url.searchParams.set("rvt-pool", "default");

const ws = new WebSocket(url, [
	"rivet",
	"rivet_encoding.json",
	`rivet_token.${process.env.RIVET_TOKEN}`,
]);

// The server sends Init as the first message once the actor is ready.
ws.onmessage = (event) => {
	const { body } = JSON.parse(String(event.data));
	if (body.tag === "Init") {
		console.log(
			"connected to actor",
			body.val.actorId,
			"as",
			body.val.connectionId,
		);
	}
};

export {};

import { createClient } from "rivetkit/client";
import type { registry } from "./index";

const client = createClient<typeof registry>(process.env.RIVET_ENDPOINT);

// connect() opens a WebSocket to /gateway/{actor}/connect and completes the
// Init handshake. The connection reconnects automatically if it drops.
const counter = client.counter.getOrCreate(["my-counter"]);
const conn = counter.connect();

// Close the socket when you are done with it.
await conn.dispose();

const ws = new WebSocket(
	`wss://api.rivet.dev/gateway/${process.env.ACTOR_ID}/connect`,
	["rivet", "rivet_encoding.json", `rivet_token.${process.env.RIVET_TOKEN}`],
);

// Pick a unique id per request. The ActionResponse (or Error) that answers
// it carries the same id, so several actions can be in flight at once.
let nextId = 1;
const pending = new Map<number, (output: unknown) => void>();

function callAction(name: string, args: unknown[]): Promise<unknown> {
	const id = nextId++;
	ws.send(
		JSON.stringify({
			body: { tag: "ActionRequest", val: { id, name, args } },
		}),
	);
	return new Promise((resolve) => pending.set(id, resolve));
}

ws.onmessage = (event) => {
	const { body } = JSON.parse(String(event.data));
	if (body.tag === "ActionResponse") {
		pending.get(body.val.id)?.(body.val.output);
		pending.delete(body.val.id);
	}
};

ws.onopen = async () => {
	const count = await callAction("increment", [1]);
	console.log(count); // 1
};

export {};

import { createClient } from "rivetkit/client";
import type { registry } from "./index";

const client = createClient<typeof registry>(process.env.RIVET_ENDPOINT);
const conn = client.counter.getOrCreate(["my-counter"]).connect();

// Actions called on a connection are sent as ActionRequest messages over the
// WebSocket instead of separate HTTP requests. The promise resolves with the
// ActionResponse output.
const count = await conn.increment(1);
console.log(count); // 1

const ws = new WebSocket(
	`wss://api.rivet.dev/gateway/${process.env.ACTOR_ID}/connect`,
	["rivet", "rivet_encoding.json", `rivet_token.${process.env.RIVET_TOKEN}`],
);

ws.onopen = () => {
	// Subscribe once. The server pushes an Event message every time the actor
	// broadcasts or sends countChanged to this connection.
	ws.send(
		JSON.stringify({
			body: {
				tag: "SubscriptionRequest",
				val: { eventName: "countChanged", subscribe: true },
			},
		}),
	);
};

ws.onmessage = (event) => {
	const { body } = JSON.parse(String(event.data));
	if (body.tag === "Event" && body.val.name === "countChanged") {
		const [count] = body.val.args;
		console.log("count is now", count);
	}
};

export {};

import { createClient } from "rivetkit/client";
import type { registry } from "./index";

const client = createClient<typeof registry>(process.env.RIVET_ENDPOINT);
const conn = client.counter.getOrCreate(["my-counter"]).connect();

// on() sends a SubscriptionRequest and routes matching Event messages to the
// callback. The returned function unsubscribes.
const unsubscribe = conn.on("countChanged", (count: number) => {
	console.log("count is now", count);
});

await conn.increment(1); // logs "count is now 1"
unsubscribe();

const ws = new WebSocket(
	`wss://api.rivet.dev/gateway/${process.env.ACTOR_ID}/connect`,
	["rivet", "rivet_encoding.json", `rivet_token.${process.env.RIVET_TOKEN}`],
);

ws.onmessage = (event) => {
	const { body } = JSON.parse(String(event.data));
	if (body.tag !== "Error") return;

	const { group, code, message, actionId } = body.val;
	if (actionId !== null) {
		// The error answers the ActionRequest with this id.
		console.error(
			`action ${actionId} failed: ${group}.${code}: ${message}`,
		);
	} else {
		// Connection-level error, for example a rejected message.
		console.error(`connection error: ${group}.${code}: ${message}`);
	}
};

// Errors that reject the upgrade itself arrive as a close frame instead,
// with a reason of the form "{group}.{code}#{ray_id}".
ws.onclose = (event) => {
	if (event.code !== 1000) console.error("closed:", event.code, event.reason);
};

export {};

import { ActorError, createClient } from "rivetkit/client";
import type { registry } from "./index";

const client = createClient<typeof registry>(process.env.RIVET_ENDPOINT);
const conn = client.counter.getOrCreate(["my-counter"]).connect();

// Error messages that answer an action reject that action's promise.
try {
	await conn.increment(1);
} catch (error) {
	if (error instanceof ActorError) {
		console.error(`${error.group}.${error.code}: ${error.message}`);
	}
}

// Errors that are not tied to an action go to onError.
conn.onError((error) => {
	console.error(`connection error: ${error.group}.${error.code}`);
});