feat(json-crdt): 🎸 buffer local transactions and allow to create transactions

Author: streamich

src/json-crdt/file/File.ts

@@ -79,16 +79,22 @@ export class File implements Printable {
   public sync(): (() => void) {
     const {model, log} = this;
     const api = model.api;
+    const drain = () => {
+      if (!api.builder.patch.ops.length) return;
+      const patch = api.flush();
+      this.log.push(patch);
+    };
     const onPatchUnsubscribe = api.onPatch.listen((patch) => {
       log.push(patch);
     });
-    const onLocalChangeUnsubscribe = api.onLocalChange.listen(() => {
-      const patch = api.flush();
-      if (patch.ops.length) this.log.push(patch);
-    });
+    const onLocalChangesUnsubscribe = api.onLocalChanges.listen(drain);
+    const onBeforeTransactionUnsubscribe = api.onBeforeTransaction.listen(drain);
+    const onTransactionUnsubscribe = api.onTransaction.listen(drain);
     return () => {
       onPatchUnsubscribe();
-      onLocalChangeUnsubscribe();
+      onLocalChangesUnsubscribe();
+      onBeforeTransactionUnsubscribe();
+      onTransactionUnsubscribe();
     };
   }
 

src/json-crdt/file/__tests__/File.spec.ts

@@ -130,6 +130,17 @@ describe('.sync()', () => {
     expect(file.log.replayToEnd().view()).toEqual({foo: 'bar', x: 1});
   });
 
+  test('processes local transactions', async () => {
+    const {file, model} = setup({foo: 'bar'});
+    file.sync();
+    const logLength = file.log.patches.size();
+    model.api.transaction(() => {
+      model.api.obj([]).set({x: 1});
+      model.api.obj([]).set({y: 2});
+    });
+    expect(file.log.patches.size()).toBe(logLength + 1);
+  });
+
   test('keeps track of remote changes', async () => {
     const {file, model} = setup({foo: 'bar'});
     const clone = model.clone();

src/json-crdt/model/api/ModelApi.ts

@@ -39,6 +39,9 @@ export class ModelApi<N extends JsonNode = JsonNode> implements SyncStore<JsonNo
   public readonly onBeforeLocalChange = new FanOut<number>();
   /** Emitted after local changes through `model.api` are applied. */
   public readonly onLocalChange = new FanOut<number>();
+  /** Emitted after local changes through `model.api` are applied. Same as
+   * `.onLocalChange`, but this event buffered withing a microtask. */
+  public readonly onLocalChanges = new MicrotaskBufferFanOut<number>(this.onLocalChange);
   /** Emitted when the model changes. Combines `onReset`, `onPatch` and `onLocalChange`. */
   public readonly onChange = new MergeFanOut<number | Patch | void>([this.onReset, this.onPatch, this.onLocalChange]);
   /** Emitted when the model changes. Same as `.onChange`, but this event is emitted once per microtask. */
@@ -249,6 +252,7 @@ export class ModelApi<N extends JsonNode = JsonNode> implements SyncStore<JsonNo
    * Flushes the builder and returns a patch.
    *
    * @returns A JSON CRDT patch.
+   * @todo Make this return undefined if there are no operations in the builder.
    */
   public flush(): Patch {
     const patch = this.builder.flush();
@@ -257,6 +261,17 @@ export class ModelApi<N extends JsonNode = JsonNode> implements SyncStore<JsonNo
     return patch;
   }
 
+  /** Emitted before a transaction is started. */
+  public readonly onBeforeTransaction = new FanOut<void>();
+  /** Emitted after transaction completes. */
+  public readonly onTransaction = new FanOut<void>();
+
+  public transaction(callback: () => void) {
+    this.onBeforeTransaction.emit();
+    callback();
+    this.onTransaction.emit();
+  }
+
   // ---------------------------------------------------------------- SyncStore
 
   public readonly subscribe = (callback: () => void) => this.onChanges.listen(() => callback());