feat(infra): livadata

packages/common/infra/package.json

@@ -7,6 +7,7 @@
     "./command": "./src/command/index.ts",
     "./atom": "./src/atom/index.ts",
     "./app-config-storage": "./src/app-config-storage.ts",
+    "./livedata": "./src/livedata/index.ts",
     ".": "./src/index.ts"
   },
   "dependencies": {
@@ -16,9 +17,11 @@
     "@blocksuite/blocks": "0.12.0-nightly-202401120404-4219e86",
     "@blocksuite/global": "0.12.0-nightly-202401120404-4219e86",
     "@blocksuite/store": "0.12.0-nightly-202401120404-4219e86",
+    "foxact": "^0.2.20",
     "jotai": "^2.5.1",
     "jotai-effect": "^0.2.3",
     "nanoid": "^5.0.3",
+    "react": "18.2.0",
     "tinykeys": "^2.1.0",
     "yjs": "^13.6.10",
     "zod": "^3.22.4"

packages/common/infra/src/livedata/__tests__/livedata.spec.ts

@@ -0,0 +1,62 @@
+import { map, Observable, of } from 'rxjs';
+import { describe, expect, test, vitest } from 'vitest';
+import * as Y from 'yjs';
+
+import { LiveData } from '..';
+
+describe('livedata', () => {
+  test('LiveData', async () => {
+    const livedata = new LiveData(0);
+    expect(livedata.value).toBe(0);
+    livedata.next(1);
+    expect(livedata.value).toBe(1);
+    let subscribed = 0;
+    livedata.subscribe(v => {
+      subscribed = v;
+    });
+    livedata.next(2);
+    expect(livedata.value).toBe(2);
+    await vitest.waitFor(() => subscribed === 2);
+  });
+
+  test('from', async () => {
+    {
+      const livedata = LiveData.from(of(1, 2, 3, 4), 0);
+      expect(livedata.value).toBe(4);
+    }
+    {
+      const livedata1 = new LiveData(1);
+      const livedata2 = LiveData.from(livedata1.pipe(map(v => 'live' + v)), '');
+      expect(livedata2.value).toBe('live1');
+      livedata1.next(2);
+      expect(livedata2.value).toBe('live2');
+    }
+
+    {
+      let observableClosed = false;
+      const observable = new Observable(() => {
+        return () => {
+          observableClosed = true;
+        };
+      });
+      const livedata = LiveData.from(observable, 0);
+      livedata.complete();
+      expect(
+        observableClosed,
+        'should close parent observable, when livedata complete'
+      ).toBe(true);
+    }
+  });
+
+  test('from yjs', async () => {
+    const ydoc = new Y.Doc();
+    const ymap = ydoc.getMap('test');
+
+    const livedata = LiveData.fromY<any>(ymap);
+    expect(livedata.value).toEqual({});
+    ymap.set('a', 1);
+    expect(livedata.value).toEqual({ a: 1 });
+    ymap.set('b', 2);
+    expect(livedata.value).toEqual({ a: 1, b: 2 });
+  });
+});

packages/common/infra/src/livedata/index.ts

@@ -0,0 +1,115 @@
+import { DebugLogger } from '@affine/debug';
+import { Observable } from 'rxjs';
+import { BehaviorSubject } from 'rxjs';
+
+const logger = new DebugLogger('livedata');
+
+/**
+ * LiveData is a reactive data type.
+ *
+ * ## basic usage
+ *
+ * @example
+ * ```ts
+ * const livedata = new LiveData(0); // create livedata with initial value
+ *
+ * livedata.next(1); // update value
+ *
+ * console.log(livedata.value); // get current value
+ *
+ * livedata.subscribe(v => { // subscribe to value changes
+ *  console.log(v); // 1
+ * });
+ * ```
+ *
+ * ## observable
+ *
+ * LiveData is a rxjs observable, you can use rxjs operators.
+ *
+ * @example
+ * ```ts
+ * new LiveData(0).pipe(
+ *   map(v => v + 1),
+ *   filter(v => v > 1),
+ *   ...
+ * )
+ * ```
+ *
+ * NOTICE: different from normal observable, LiveData will always emit the latest value when you subscribe to it.
+ *
+ * ## from observable
+ *
+ * LiveData can be created from observable or from other livedata.
+ *
+ * @example
+ * ```ts
+ * const A = LiveData.from(
+ *   of(1, 2, 3, 4), // from observable
+ *   0 // initial value
+ * );
+ *
+ * const B = LiveData.from(
+ *   A.pipe(map(v => 'from a ' + v)), // from other livedata
+ *   '' // initial value
+ * );
+ * ```
+ *
+ * NOTICE: LiveData.from will not complete when the observable completes, you can use `spreadComplete` option to change
+ * this behavior.
+ *
+ * @see {@link https://rxjs.dev/api/index/class/BehaviorSubject}
+ */
+export class LiveData<T = unknown> extends BehaviorSubject<T> {
+  static from<T>(
+    observable: Observable<T>,
+    initialValue: T,
+    { spreadComplete = false }: { spreadComplete?: boolean } = {}
+  ): LiveData<T> {
+    const data = new LiveData(initialValue);
+
+    const subscription = observable.subscribe({
+      next(value) {
+        data.next(value);
+      },
+      error(err) {
+        if (spreadComplete) {
+          data.error(err);
+        } else {
+          logger.error('uncatched error in livedata', err);
+        }
+      },
+      complete() {
+        if (spreadComplete) {
+          data.complete();
+        }
+      },
+    });
+    data.subscribe({
+      complete() {
+        subscription.unsubscribe();
+      },
+      error() {
+        subscription.unsubscribe();
+      },
+    });
+
+    return data;
+  }
+
+  static fromY<T>(ydata: any): LiveData<T> {
+    if (typeof ydata.toJSON !== 'function') {
+      throw new Error('unsupported yjs type');
+    }
+
+    return LiveData.from<T>(
+      new Observable(subscriber => {
+        ydata.observeDeep(() => {
+          subscriber.next(ydata.toJSON());
+        });
+      }),
+      ydata.toJSON()
+    );
+  }
+}
+
+export * from './react';

packages/common/infra/src/livedata/react.ts

@@ -0,0 +1,41 @@
+import { use } from 'foxact/use';
+import { useEffect, useState } from 'react';
+
+import type { LiveData } from '.';
+
+/**
+ * subscribe LiveData and return the value.
+ */
+export function useLiveData<T>(liveData: LiveData<T>): T {
+  const [data, setData] = useState(liveData.value);
+
+  useEffect(() => {
+    const subscription = liveData.subscribe(value => {
+      setData(value);
+    });
+    return () => subscription.unsubscribe();
+  }, [liveData]);
+
+  return data;
+}
+
+/**
+ * subscribe LiveData and return the value. If the value is nullish, will suspends until the value is not nullish.
+ */
+export function useEnsureLiveData<T>(liveData: LiveData<T>): NonNullable<T> {
+  const data = useLiveData(liveData);
+
+  if (data === null || data === undefined) {
+    return use(
+      new Promise(resolve => {
+        liveData.subscribe(value => {
+          if (value === null || value === undefined) {
+            resolve(value);
+          }
+        });
+      })
+    );
+  }
+
+  return data;
+}

yarn.lock

@@ -12856,6 +12856,7 @@ __metadata:
     "@blocksuite/presets": "npm:0.12.0-nightly-202401120404-4219e86"
     "@blocksuite/store": "npm:0.12.0-nightly-202401120404-4219e86"
     async-call-rpc: "npm:^6.3.1"
+    foxact: "npm:^0.2.20"
     jotai: "npm:^2.5.1"
     jotai-effect: "npm:^0.2.3"
     nanoid: "npm:^5.0.3"