integrate extra io stepping logic to the JS bindings

bindings/javascript/sync/src/js_protocol_io.rs

@@ -242,27 +242,47 @@ impl ProtocolIO for JsProtocolIo {
         }))
     }
 
-    fn register(&self, callback: Box<dyn FnMut() -> bool>) {
-        tracing::info!("register callback in the ProtocolIo");
+    fn add_work(&self, callback: Box<dyn FnMut() -> bool + Send + Sync>) {
+        let mut work = self.work.lock().unwrap();
+        work.push_back(callback);
+    }
+
+    fn step_work(&self) {
+        let mut items = {
+            let mut work = self.work.lock().unwrap();
+            work.drain(..).collect::<VecDeque<_>>()
+        };
+        let length = items.len();
+        for _ in 0..length {
+            let mut item = items.pop_front().unwrap();
+            if item() {
+                continue;
+            }
+            items.push_back(item);
+        }
+        {
+            let mut work = self.work.lock().unwrap();
+            work.extend(items);
+        }
     }
 }
 
 #[napi]
 pub struct JsProtocolIo {
     requests: Mutex<Vec<JsProtocolRequestBytes>>,
+    work: Mutex<VecDeque<Box<dyn FnMut() -> bool + Send + Sync>>>,
 }
 
 impl Default for JsProtocolIo {
     fn default() -> Self {
         Self {
             requests: Mutex::new(Vec::new()),
+            work: Mutex::new(VecDeque::new()),
         }
     }
 }
 
-#[napi]
 impl JsProtocolIo {
-    #[napi]
     pub fn take_request(&self) -> Option<JsProtocolRequestBytes> {
         self.requests.lock().unwrap().pop()
     }

bindings/javascript/sync/src/lib.rs

@@ -14,6 +14,7 @@ use napi_derive::napi;
 use turso_node::{DatabaseOpts, IoLoopTask};
 use turso_sync_engine::{
     database_sync_engine::{DatabaseSyncEngine, DatabaseSyncEngineOpts},
+    protocol_io::ProtocolIO,
     types::{Coro, DatabaseChangeType, DatabaseSyncEngineProtocolVersion},
 };
 
@@ -304,6 +305,11 @@ impl SyncEngine {
         Ok(self.protocol()?.take_request())
     }
 
+    #[napi]
+    pub fn protocol_io_step(&self) -> napi::Result<()> {
+        Ok(self.protocol()?.step_work())
+    }
+
     #[napi]
     pub fn push(&self) -> GeneratorHolder {
         self.run(async move |coro, guard| {