Tablebox ドキュメントサイト English

Aggregation

Aggregation は、大勢の書き手が一つの名前の下に書いたものを、どの読み手にも同じ一つの文書にまとめ、各書き手に変更ごとに適用されたかを伝え、全書き手の今の値を一覧します。

パス

パスは参加者のルートからの相対です。{name} は空を含む任意のパスで、{writer} は最初の .aggregation の段より後ろのすべてです。

書く人 パス トラック 中身
書き手 {name}/.aggregation/{writer} patch.json 書き手の変更: 各グループに一つのフレーム、RFC 7396 の merge patch として当てる JSON オブジェクト。グループの番号が変更の番号
書き手 同じ value.json 書き手の今の値。JSON スナップショットのトラック
Aggregation .aggregation/{name} state.json まとめた文書。JSON スナップショットのトラック
Aggregation 同じ writers.json {"{writer}": value} の JSON スナップショットのトラック。今 value.json を出している書き手ごとに一項目
Aggregation 同じ applied/{writer} その書き手の変更ごとに一フレーム: {"applied": n} か {"refused": n, "code": name}

alice として書けるのは、publish が {name}/.aggregation/alice を覆うセッションだけです。そのため publish が board/.aggregation/{id} のリンクなら、各自は自分としてボードに書けます。.aggregation/{name} を読むと書き手が書いたものが分かるので、書き手のパスと同じように許してください。

TypeScript（ブラウザー） · aggregation.ts
// A shared board through Aggregation: changes go to board/.aggregation/{me}, the merged board and
// every writer's current value come from .aggregation/board.
import { type Connection, json, moq } from 'tablebox.io';

export async function board(connection: Connection, me: string, show: (board: unknown, writers: unknown) => void) {
  const mine = connection.publish(`board/.aggregation/${me}`);
  const changes = mine.createTrack('patch.json');
  const value = new json.Snapshot.Producer({ track: mine.createTrack('value.json') });
  // A writer never uses a change number twice, so each broadcast numbers its changes from the time.
  let next = Date.now();
  const merged = await connection.read('.aggregation/board');
  const state = new json.Snapshot.Consumer({ track: merged.track('state.json').subscribe() });
  const writers = new json.Snapshot.Consumer({ track: merged.track('writers.json').subscribe() });
  let board: unknown;
  let everyone: unknown;
  void (async () => { for await (board of state) show(board, everyone); })();
  void (async () => { for await (everyone of writers) show(board, everyone); })();
  return {
    // One RFC 7396 merge patch per change: { "note-1": { "text": "Hi" } } sets a key, null deletes it.
    change: (patch: Record<string, unknown>) => {
      const group = new moq.Group.Producer(next++);
      changes.writeGroup(group);
      group.writeJson(patch);
      group.close();
    },
    // This writer's own entry in writers.json, such as its name and cursor.
    set: (mine: unknown) => value.update(mine),
  };
}

まとめ方と確認

.aggregation/{name} のどれかのトラックを読むと、その名前の Aggregation が動き始め、{name}/.aggregation/ の下の書き手をすべて読みます。各書き手の変更はその書き手の順に、異なる書き手の変更は Aggregation に届いた順に当て、どのキーでも最後に当てた変更が勝ちます。state.json の読み手は、どの Relay にいても、同じ文書の並びを読みます。

Aggregation が動いている間に書き手が出した変更には、applied/{writer} に一フレームずつ返ります。state.json に入ったら applied、そうでなければ refused と、bad-request（フレームが JSON オブジェクトでない）、too-large（文書が 16 MiB を超える）、missing-variable のどれかです。n はその変更のグループ番号です。書き手は変更に 1 ずつ増える番号を振り、同じ番号を二度使いません。各ブロードキャストは Unix エポックからのミリ秒の時刻から始め、それより前の番号の方が大きければその次から始めます。そのため applied/{writer} の一フレームは、書き手のどのブロードキャストから来た変更でも、一つの変更を指します。書き手は書く前に自分の applied/{writer} を読み、その読みが始まり直すたびに（再接続の後や、引き継ぎで終わった後）、確認の返っていない変更を出し直します。merge patch は二度当てても同じ文書になります。

全書き手の値

writers.json は、名前やカーソルなど、各書き手の最新の value.json の値を持ちます。書き手の項目は、そのブロードキャストか value.json が終わると消えるので、一覧は今いる人に合わせて変わります。全員が全員のブロードキャストを一覧するには大きすぎる範囲では、代わりに人を writers.json に置き、要る人だけが読めます。

保存

state.json は、保存してある文書から始まり、変わるたびに Persistence を通して保存されます。そのため Persistence のストレージの変数が要ります。Aggregation は、Persistence が保存してある文書を渡してから state.json に出し、何も保存されていなければ {} から始めます。文書は、ストレージの STORAGE_PREFIX の下の .aggregation/{project path of name}/state.json にあります。

動き方と失敗

Aggregation は、出力が読まれているか書き手がいる間動き、どちらもいなくなってから 20 秒で止まります。state.json を保存する Persistence は、最後の書き手が去ってから 20 秒までそれを読むので、ほかに読む人のいない名前は、最後の書き手が去ってから 40 秒で止まります。一つのプロジェクトのすべての名前は、一度に一つのインスタンスが受け持ちます。