feat: add subscribe to containers in wasm

crates/loro-wasm/src/lib.rs

@@ -225,27 +225,7 @@ impl Loro {
     pub fn subscribe(&self, f: js_sys::Function) -> u32 {
         let observer = observer::Observer::new(f);
         self.0.borrow_mut().subscribe_deep(Box::new(move |e| {
-            let promise = Promise::resolve(&JsValue::NULL);
+            call_after_micro_task(observer.clone(), e);
-            let ob = observer.clone();
-            type C = Closure<dyn FnMut(JsValue)>;
-            let drop_handler: Rc<RefCell<Option<C>>> = Rc::new(RefCell::new(None));
-            let copy = drop_handler.clone();
-            let closure = Closure::once(move |_: JsValue| {
-                ob.call1(
-                    &Event {
-                        local: e.local,
-                        origin: e.origin.clone(),
-                        target: e.target.clone(),
-                        diff: Either::A(e.diff.to_vec()),
-                        path: Either::A(e.absolute_path.clone()),
-                    }
-                    .into(),
-                );
-
-                drop(copy);
-            });
-            let _ = promise.then(&closure);
-            drop_handler.borrow_mut().replace(closure);
         }))
     }
 
@@ -264,6 +244,29 @@ impl Loro {
     }
 }
 
+fn call_after_micro_task(ob: observer::Observer, e: Arc<loro_internal::event::Event>) {
+    let promise = Promise::resolve(&JsValue::NULL);
+    type C = Closure<dyn FnMut(JsValue)>;
+    let drop_handler: Rc<RefCell<Option<C>>> = Rc::new(RefCell::new(None));
+    let copy = drop_handler.clone();
+    let closure = Closure::once(move |_: JsValue| {
+        ob.call1(
+            &Event {
+                local: e.local,
+                origin: e.origin.clone(),
+                target: e.target.clone(),
+                diff: Either::A(e.diff.to_vec()),
+                path: Either::A(e.absolute_path.clone()),
+            }
+            .into(),
+        );
+
+        drop(copy);
+    });
+    let _ = promise.then(&closure);
+    drop_handler.borrow_mut().replace(closure);
+}
+
 impl Default for Loro {
     fn default() -> Self {
         Self::new()
@@ -399,6 +402,37 @@ impl LoroText {
     pub fn length(&self) -> usize {
         self.0.len()
     }
+
+    pub fn subscribe(&self, txn: &JsTransaction, f: js_sys::Function) -> JsResult<u32> {
+        let observer = observer::Observer::new(f);
+        let txn = get_transaction_mut(txn);
+        let ans = self.0.subscribe(
+            &txn,
+            Box::new(move |e| {
+                call_after_micro_task(observer.clone(), e);
+            }),
+        )?;
+        Ok(ans)
+    }
+
+    #[wasm_bindgen(js_name = "subscribeOnce")]
+    pub fn subscribe_once(&self, txn: &JsTransaction, f: js_sys::Function) -> JsResult<u32> {
+        let observer = observer::Observer::new(f);
+        let txn = get_transaction_mut(txn);
+        let ans = self.0.subscribe_once(
+            &txn,
+            Box::new(move |e| {
+                call_after_micro_task(observer.clone(), e);
+            }),
+        )?;
+        Ok(ans)
+    }
+
+    pub fn unsubscribe(&self, txn: &JsTransaction, subscription: u32) -> JsResult<()> {
+        let txn = get_transaction_mut(txn);
+        self.0.unsubscribe(&txn, subscription)?;
+        Ok(())
+    }
 }
 
 #[wasm_bindgen]
@@ -477,6 +511,50 @@ impl LoroMap {
         };
         Ok(container)
     }
+
+    pub fn subscribe(&self, txn: &JsTransaction, f: js_sys::Function) -> JsResult<u32> {
+        let observer = observer::Observer::new(f);
+        let txn = get_transaction_mut(txn);
+        let ans = self.0.subscribe(
+            &txn,
+            Box::new(move |e| {
+                call_after_micro_task(observer.clone(), e);
+            }),
+        )?;
+        Ok(ans)
+    }
+
+    #[wasm_bindgen(js_name = "subscribeOnce")]
+    pub fn subscribe_once(&self, txn: &JsTransaction, f: js_sys::Function) -> JsResult<u32> {
+        let observer = observer::Observer::new(f);
+        let txn = get_transaction_mut(txn);
+        let ans = self.0.subscribe_once(
+            &txn,
+            Box::new(move |e| {
+                call_after_micro_task(observer.clone(), e);
+            }),
+        )?;
+        Ok(ans)
+    }
+
+    #[wasm_bindgen(js_name = "subscribeDeep")]
+    pub fn subscribe_deep(&self, txn: &JsTransaction, f: js_sys::Function) -> JsResult<u32> {
+        let observer = observer::Observer::new(f);
+        let txn = get_transaction_mut(txn);
+        let ans = self.0.subscribe_deep(
+            &txn,
+            Box::new(move |e| {
+                call_after_micro_task(observer.clone(), e);
+            }),
+        )?;
+        Ok(ans)
+    }
+
+    pub fn unsubscribe(&self, txn: &JsTransaction, subscription: u32) -> JsResult<()> {
+        let txn = get_transaction_mut(txn);
+        self.0.unsubscribe(&txn, subscription)?;
+        Ok(())
+    }
 }
 
 #[wasm_bindgen]
@@ -554,6 +632,50 @@ impl LoroList {
         };
         Ok(container)
     }
+
+    pub fn subscribe(&self, txn: &JsTransaction, f: js_sys::Function) -> JsResult<u32> {
+        let observer = observer::Observer::new(f);
+        let txn = get_transaction_mut(txn);
+        let ans = self.0.subscribe(
+            &txn,
+            Box::new(move |e| {
+                call_after_micro_task(observer.clone(), e);
+            }),
+        )?;
+        Ok(ans)
+    }
+
+    #[wasm_bindgen(js_name = "subscribeOnce")]
+    pub fn subscribe_once(&self, txn: &JsTransaction, f: js_sys::Function) -> JsResult<u32> {
+        let observer = observer::Observer::new(f);
+        let txn = get_transaction_mut(txn);
+        let ans = self.0.subscribe_once(
+            &txn,
+            Box::new(move |e| {
+                call_after_micro_task(observer.clone(), e);
+            }),
+        )?;
+        Ok(ans)
+    }
+
+    #[wasm_bindgen(js_name = "subscribeDeep")]
+    pub fn subscribe_deep(&self, txn: &JsTransaction, f: js_sys::Function) -> JsResult<u32> {
+        let observer = observer::Observer::new(f);
+        let txn = get_transaction_mut(txn);
+        let ans = self.0.subscribe_deep(
+            &txn,
+            Box::new(move |e| {
+                call_after_micro_task(observer.clone(), e);
+            }),
+        )?;
+        Ok(ans)
+    }
+
+    pub fn unsubscribe(&self, txn: &JsTransaction, subscription: u32) -> JsResult<()> {
+        let txn = get_transaction_mut(txn);
+        self.0.unsubscribe(&txn, subscription)?;
+        Ok(())
+    }
 }
 
 #[wasm_bindgen(typescript_custom_section)]

loro-js/src/index.ts

@@ -136,6 +136,9 @@ declare module "loro-wasm" {
     ): never;
 
     get(index: number): Value;
+    subscribe(txn: Transaction | Loro, listener: Listener): number;
+    subscribeDeep(txn: Transaction | Loro, listener: Listener): number;
+    subscribeOnce(txn: Transaction | Loro, listener: Listener): number;
   }
 
   interface LoroMap {
@@ -161,5 +164,14 @@ declare module "loro-wasm" {
     ): never;
 
     get(key: string): Value;
+    subscribe(txn: Transaction | Loro, listener: Listener): number;
+    subscribeDeep(txn: Transaction | Loro, listener: Listener): number;
+    subscribeOnce(txn: Transaction | Loro, listener: Listener): number;
+  }
+
+  interface LoroText {
+    subscribe(txn: Transaction | Loro, listener: Listener): number;
+    subscribeDeep(txn: Transaction | Loro, listener: Listener): number;
+    subscribeOnce(txn: Transaction | Loro, listener: Listener): number;
   }
 }

loro-js/tests/event.test.ts

@@ -1,10 +1,9 @@
 import { describe, expect, it } from "vitest";
 import {
-  Diff,
+  Delta,
   ListDiff,
   Loro,
   LoroEvent,
-  LoroMap,
   MapDIff as MapDiff,
   TextDiff,
 } from "../src";
@@ -131,6 +130,87 @@ describe("event", () => {
       } as MapDiff],
     );
   });
+
+  describe("subscribe container events", () => {
+    it("text", async () => {
+      const loro = new Loro();
+      const text = loro.getText("text");
+      let ran = 0;
+      let oneTimeRan = 0;
+      text.subscribeOnce(loro, (_) => {
+        oneTimeRan += 1;
+      });
+      const sub = text.subscribe(loro, (event) => {
+        if (!ran) {
+          expect(event.diff[0].diff).toStrictEqual(
+            [{ type: "insert", "value": "123" }] as Delta<string>[],
+          );
+        }
+        ran += 1;
+        expect(event.target).toBe(text.id);
+      });
+      text.insert(loro, 0, "123");
+      text.insert(loro, 1, "456");
+      await zeroMs();
+      expect(ran).toBeTruthy();
+      // subscribeOnce test
+      expect(oneTimeRan).toBe(1);
+      expect(text.toString()).toEqual("145623");
+
+      // unsubscribe
+      const oldRan = ran;
+      text.unsubscribe(loro, sub);
+      text.insert(loro, 0, "789");
+      expect(ran).toBe(oldRan);
+    });
+
+    it("map subscribe deep", async () => {
+      const loro = new Loro();
+      const map = loro.getMap("map");
+      let times = 0;
+      const sub = map.subscribeDeep(loro, (event) => {
+        times += 1;
+      });
+
+      const subMap = map.insertContainer(loro, "sub", "Map");
+      await zeroMs();
+      expect(times).toBe(1);
+      const text = subMap.insertContainer(loro, "k", "Text");
+      await zeroMs();
+      expect(times).toBe(2);
+      text.insert(loro, 0, "123");
+      await zeroMs();
+      expect(times).toBe(3);
+
+      // unsubscribe
+      map.unsubscribe(loro, sub);
+      text.insert(loro, 0, "123");
+      await zeroMs();
+      expect(times).toBe(3);
+    });
+
+    it("list subscribe deep", async () => {
+      const loro = new Loro();
+      const list = loro.getList("list");
+      let times = 0;
+      const sub = list.subscribeDeep(loro, (_) => {
+        times += 1;
+      });
+
+      const text = list.insertContainer(loro, 0, "Text");
+      await zeroMs();
+      expect(times).toBe(1);
+      text.insert(loro, 0, "123");
+      await zeroMs();
+      expect(times).toBe(2);
+
+      // unsubscribe
+      list.unsubscribe(loro, sub);
+      text.insert(loro, 0, "123");
+      await zeroMs();
+      expect(times).toBe(2);
+    });
+  });
 });
 
 function zeroMs(): Promise<void> {