agents/state 中的 State 会将一个 JSON 值存储在 Durable Object 的 SQLite 数据库中。首次读取时会加载该值，并在内存中缓存；每次变更都会经过验证，变更后还会调用回调。将其与 WebSockets 搭配使用，可将值同步给已连接的客户端。
import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { State } from "agents/state";
export class SettingsAgent extends DurableObject {
settings = new State({
initialState: { theme: "light", notifications: true },
});
lifecycle = Lifecycle.install(this).use(this.settings);
async onRequest(request) {
if (request.method === "PUT") {
this.settings.set(await request.json());
}
return Response.json(this.settings.get());
}
}import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { State } from "agents/state";
type Settings = { theme: "light" | "dark"; notifications: boolean };
export class SettingsAgent extends DurableObject<Env> {
readonly settings = new State<Settings>({
initialState: { theme: "light", notifications: true },
});
readonly lifecycle = Lifecycle.install(this).use(this.settings);
async onRequest(request: Request): Promise<Response> {
if (request.method === "PUT") {
this.settings.set(await request.json<Settings>());
}
return Response.json(this.settings.get());
}
}| 方法 | 说明 |
|---|---|
get() |
返回当前值；若未存储任何值，则返回 undefined。 |
set(next, source?) |
验证并存储 next，然后调用 onChanged。source 默认为 "server"。 |
set() 会同步写入；若验证器拒绝变更，则抛出异常。此能力的 ID 是 "state"，因此每个 Durable Object 只能安装一个 State。如需存储多个值，请将它们放入同一个对象，或使用自己的表。
| 选项 | 说明 |
|---|---|
initialState |
若不存在值，则在首次读取时存储。 |
validateStateChange |
(next, source) => void。若要在存储前拒绝变更，请抛出异常。此函数必须同步执行。 |
onChanged |
(state, source) => void 或一个 Promise。在变更存储后运行。即使失败，也不会撤销写入。 |
两个钩子都会收到 StateChangeSource。对于你自己的代码所做的变更，其值为 "server"；对于从客户端发送变更的连接，则为对应的 Connection：
import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { State } from "agents/state";
export class DocAgent extends DurableObject {
doc = new State({
initialState: { title: "Untitled" },
validateStateChange: (next, source) => {
if (source !== "server" && next.lockedBy !== undefined) {
throw new Error("Only the server can lock a document");
}
},
onChanged: (state, source) => {
const who = source === "server" ? "server" : source.id;
console.log(`${who} changed the title to ${state.title}`);
},
});
lifecycle = Lifecycle.install(this).use(this.doc);
}import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { State, type StateChangeSource } from "agents/state";
type Doc = { title: string; lockedBy?: string };
export class DocAgent extends DurableObject<Env> {
readonly doc = new State<Doc>({
initialState: { title: "Untitled" },
validateStateChange: (next: Doc, source: StateChangeSource) => {
if (source !== "server" && next.lockedBy !== undefined) {
throw new Error("Only the server can lock a document");
}
},
onChanged: (state: Doc, source: StateChangeSource) => {
const who = source === "server" ? "server" : source.id;
console.log(`${who} changed the title to ${state.title}`);
},
});
readonly lifecycle = Lifecycle.install(this).use(this.doc);
}将同一个 State 实例传递给 WebSockets 和 Lifecycle。在 onChanged 中调用 broadcastState(source)，将每次变更告知其他客户端：
import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { routeAgentRequest } from "agents/routing";
import { State } from "agents/state";
import { WebSockets } from "agents/websockets";
export class CounterAgent extends DurableObject {
state = new State({
initialState: { count: 0 },
validateStateChange: (next) => {
if (!Number.isSafeInteger(next.count) || next.count < 0) {
throw new Error("count must be a non-negative integer");
}
},
onChanged: (_state, source) => this.webSockets.broadcastState(source),
});
webSockets = new WebSockets({ state: this.state });
lifecycle = Lifecycle.install(this).use(this.state).use(this.webSockets);
}
export default {
async fetch(request, env) {
return (
(await routeAgentRequest(request, env)) ??
new Response("Not found", { status: 404 })
);
},
};import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";
import { routeAgentRequest } from "agents/routing";
import { State } from "agents/state";
import { WebSockets } from "agents/websockets";
type Counter = { count: number };
export class CounterAgent extends DurableObject<Env> {
readonly state: State<Counter> = new State<Counter>({
initialState: { count: 0 },
validateStateChange: (next) => {
if (!Number.isSafeInteger(next.count) || next.count < 0) {
throw new Error("count must be a non-negative integer");
}
},
onChanged: (_state, source) => this.webSockets.broadcastState(source),
});
readonly webSockets: WebSockets = new WebSockets({ state: this.state });
readonly lifecycle = Lifecycle.install(this)
.use(this.state)
.use(this.webSockets);
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
return (
(await routeAgentRequest(request, env)) ??
new Response("Not found", { status: 404 })
);
},
} satisfies ExportedHandler<Env>;state 和 webSockets 字段会相互引用，因此请为它们指定显式类型。
连接建立时，WebSockets 会先向每个客户端发送身份帧，再发送当前值。客户端调用 setState() 时，变更会连同该连接作为 source 一起交由 validateStateChange 处理。服务器会记录被拒绝的变更，并向客户端返回通用错误。broadcastState(source) 会将新值发送给除 source 之外的所有连接，因为来源连接已经拥有该值。
使用 useAgent 在 React 中读取和更新该值：
import { useAgent } from "agents/react";
export function Counter() {
const agent = useAgent<{ count: number }>({
agent: "CounterAgent",
name: "demo",
});
const count = agent.state?.count ?? 0;
return (
<button onClick={() => agent.setState({ count: count + 1 })}>
{count}
</button>
);
}若要拒绝某些连接的状态变更，请使用只读连接。只读连接仍会接收状态。
connection.setState() 是另一项功能。它会在单个连接上存储一个值，且不会与其他客户端同步。
State 会将值存储在 cf_agents_state 表中。每次变更都会重写整个值，因此大型或关系型数据应存放在你自己的表中。
Agent 类会安装相互连接的 State 和 WebSockets。它的 initialState、this.state、this.setState()、validateStateChange() 和 onStateChanged() 会调用这两项功能，并自动广播每次变更。更多信息请参阅存储和同步状态。