mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
5f312dac4c
Chord's shared watch: listeners take a tagged call, { kind: "value", value, ops, retired } or
{ kind: "failed", error }; a recoverable failure keeps the watch open and the next value arrives as
[["r", value]]; WatchEnd is tagged by kind, and an aborted context ends a watch "aborted".
Durable: observers take the watch's shapes (advance(call, ctx), end(end)); onReport(report) with
report.kind and report.error, session_failed is storage_failed; subscribeCommits is observeCommits;
SessionEnd and ReportFrom are tagged by kind; FileChange is tagged and file watchers stop with
stop(ctx); CompactionFailure and CompactionStatus name their trigger (pi.live version 2 with a
migrate); the task_outdated report uses the scheduler's reasons; TaskInspection's ready is pending;
DocumentCommitChange is tagged. Stored formats are unchanged.
58 lines
2.3 KiB
TypeScript
58 lines
2.3 KiB
TypeScript
// Serialize asynchronous document work with a watch.
|
|
// Run from packages/durable:
|
|
// node --conditions=source --experimental-strip-types test/examples/04-watches.ts
|
|
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
|
|
import { createSession, defineDoc, MemoryStorage } from "../../src/index.ts";
|
|
|
|
const context = BACKGROUND_CONTEXT;
|
|
const session = createSession(new MemoryStorage());
|
|
|
|
const Notes = defineDoc<{ text: string }>({
|
|
checkpointWhen: () => false,
|
|
kind: "example.notes",
|
|
version: 1,
|
|
scope: "conversation",
|
|
history: "rewindable",
|
|
fork: "asOf",
|
|
initial: () => ({ text: "" }),
|
|
});
|
|
const chat = await session.commit(async (tx) => {
|
|
const conversation = await tx.createConversation({ ownership: { kind: "ownerless" } });
|
|
(await tx.doc(Notes, conversation.id)).text = "first";
|
|
return conversation;
|
|
}, context);
|
|
|
|
// A watch starts from one stable acquisition revision. Callbacks never overlap:
|
|
// commits made while one runs arrive folded into the next call, which gets the
|
|
// newest value and ops that turn the previously delivered value into it (one
|
|
// full-value replacement `[["r", value]]` once more than 10,000 ops pile up).
|
|
const notesWatch = await session.watchDoc(Notes, chat.id, context);
|
|
if (notesWatch === undefined) throw new Error("notes are absent");
|
|
console.log("watch baseline:", notesWatch.value);
|
|
const busy = Promise.withResolvers<void>();
|
|
const done = Promise.withResolvers<void>();
|
|
let calls = 0;
|
|
notesWatch.start(async (call, _deliveryContext) => {
|
|
// Calls are tagged: a `failed` call is a failure a source recovers from, which Durable never sends (its failures
|
|
// end the watch, see `closed`).
|
|
if (call.kind !== "value") return;
|
|
const { value, ops } = call;
|
|
console.log("watch update:", value, JSON.stringify(ops));
|
|
// The first call stays busy, as a slow consumer would, while three more commits land.
|
|
if (++calls === 1) await busy.promise;
|
|
else done.resolve();
|
|
});
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(Notes, chat.id)).text = "observed";
|
|
}, context);
|
|
for (const more of [" asynchronously", ", folded", "!"]) {
|
|
await session.commit(async (tx) => {
|
|
(await tx.doc(Notes, chat.id)).text += more;
|
|
}, context);
|
|
}
|
|
busy.resolve();
|
|
await done.promise;
|
|
console.log("listener calls:", calls);
|
|
await notesWatch.stop();
|
|
|
|
await session.close(context);
|