feat(json-crdt): 🎸 add autoflushing functionality

Author: streamich

src/json-crdt/file/File.ts

@@ -79,22 +79,17 @@ 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 autoflushUnsubscribe = api.autoFlush();
     const onPatchUnsubscribe = api.onPatch.listen((patch) => {
       log.push(patch);
     });
-    const onLocalChangesUnsubscribe = api.onLocalChanges.listen(drain);
-    const onBeforeTransactionUnsubscribe = api.onBeforeTransaction.listen(drain);
-    const onTransactionUnsubscribe = api.onTransaction.listen(drain);
+    const onFlushUnsubscribe = api.onFlush.listen((patch) => {
+      log.push(patch);
+    });
     return () => {
+      autoflushUnsubscribe();
       onPatchUnsubscribe();
-      onLocalChangesUnsubscribe();
-      onBeforeTransactionUnsubscribe();
-      onTransactionUnsubscribe();
+      onFlushUnsubscribe();
     };
   }
 

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

@@ -42,6 +42,10 @@ export class ModelApi<N extends JsonNode = JsonNode> implements SyncStore<JsonNo
   /** 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 before a transaction is started. */
+  public readonly onBeforeTransaction = new FanOut<void>();
+  /** Emitted after transaction completes. */
+  public readonly onTransaction = new FanOut<void>();
   /** 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. */
@@ -248,6 +252,12 @@ export class ModelApi<N extends JsonNode = JsonNode> implements SyncStore<JsonNo
     return this.model.view();
   }
 
+  public transaction(callback: () => void) {
+    this.onBeforeTransaction.emit();
+    callback();
+    this.onTransaction.emit();
+  }
+
   /**
    * Flushes the builder and returns a patch.
    *
@@ -261,15 +271,26 @@ 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 stopAutoFlush?: () => void = undefined;
 
-  public transaction(callback: () => void) {
-    this.onBeforeTransaction.emit();
-    callback();
-    this.onTransaction.emit();
+  /**
+   * Begins to automatically flush buffered operations into patches, grouping
+   * operations by microtasks or by transactions. To capture the patch, listen
+   * to the `.onFlush` event.
+   *
+   * @returns Callback to stop auto flushing.
+   */
+  public autoFlush(): (() => void) {
+    const drain = () => this.builder.patch.ops.length && this.flush();
+    const onLocalChangesUnsubscribe = this.onLocalChanges.listen(drain);
+    const onBeforeTransactionUnsubscribe = this.onBeforeTransaction.listen(drain);
+    const onTransactionUnsubscribe = this.onTransaction.listen(drain);
+    return this.stopAutoFlush = () => {
+      this.stopAutoFlush = undefined;
+      onLocalChangesUnsubscribe();
+      onBeforeTransactionUnsubscribe();
+      onTransactionUnsubscribe();
+    };
   }
 
   // ---------------------------------------------------------------- SyncStore