enhance(util): add ratelimit

src/main/frontend/components/file_sync.cljs

@@ -40,7 +40,7 @@
                    (state/set-state! :file-sync/toggling? true)
                    (if-not graph-txid-exists?
                      (as/go
-                       (notifications/show! "Not a remote graph, ready to init remote graph!" :warn)
+                       (notifications/show! "Going to init a remote graph!" :warn)
                        (let [repo (state/get-current-repo)
                              GraphName (util/node-path.basename repo)]
                          (when-let [GraphUUID (get (as/<! (file-sync-handler/create-graph GraphName)) 2)]

src/main/frontend/handler/events.cljs

@@ -121,7 +121,7 @@
      (graph-switch graph))))
 
 (defmethod handle :graph/switch [[_ graph opts]]
-  (if (outliner-file/writes-finished?)
+  (if @outliner-file/*writes-finished?
     (graph-switch-on-persisted graph opts)
     (notification/show!
      "Please wait seconds until all changes are saved for the current graph."

src/main/frontend/modules/outliner/file.cljs

@@ -13,14 +13,8 @@
             [frontend.state :as state]))
 
 (defonce write-chan (async/chan 100))
-(defonce write-chan-batch-buf (atom []))
-
 (def batch-write-interval 1000)
 
-(defn writes-finished?
-  []
-  (empty? @write-chan-batch-buf))
-
 (defn do-write-file!
   [repo page-db-id]
   (let [page-block (db/pull repo '[*] page-db-id)
@@ -37,6 +31,7 @@
 (defn write-files!
   [pages]
   (when (seq pages)
+    (println :write-files pages)
     (when-not config/publishing?
       (doseq [[repo page-id] (set pages)]
         (try (do-write-file! repo page-id)
@@ -57,7 +52,16 @@
     (when-let [repo (state/get-current-repo)]
       (async/put! write-chan [repo page-db-id]))))
 
-(util/batch write-chan
-            batch-write-interval
-            write-files!
-            write-chan-batch-buf)
+(def *writes-finished? (atom true))
+
+(let [ch (util/ratelimit write-chan batch-write-interval
+                         :filter-fn
+                         #(do (reset! *writes-finished? false) true)
+                         :flush-fn
+                         #(reset! *writes-finished? true))]
+
+  (async/go-loop []
+    (let [item (async/<! ch)
+          items (cons item (util/drain-chan ch))]
+      (write-files! items)
+      (recur))))

src/main/frontend/util.cljc

@@ -1086,6 +1086,47 @@
         (do (swap! buf conj v)
             (recur buf t))))))
 
+(defn ratelimit
+  "return a channel CH,
+  ratelimit flush items in in-ch every max-duration(ms),
+  opts:
+  - :filter-fn filter item before putting items into returned CH, (filter-fn item)
+  - :flush-fn exec flush-fn when time to flush, (flush-fn item-coll)
+  - :stop-ch stop go-loop when stop-ch closed
+  - :distinct-coll? distinct coll when put into CH
+  - :chan-buffer buffer of return CH, default use (async/chan 1000)"
+  [in-ch max-duration & {:keys [filter-fn flush-fn stop-ch distinct-coll? chan-buffer]}]
+  (let [ch (if chan-buffer (async/chan chan-buffer) (async/chan 1000))
+        stop-ch* (or stop-ch (async/chan))]
+    (async/go-loop [timeout-ch (async/timeout max-duration) coll []]
+      (let [{:keys [timeout e stop]}
+            (async/alt! timeout-ch {:timeout true}
+                        in-ch ([e] {:e e})
+                        stop-ch* {:stop true})]
+        (cond
+          timeout
+          (do (async/onto-chan! ch coll false)
+              (flush-fn coll)
+              (recur (async/timeout max-duration) []))
+
+          (some? e)
+          (if (filter-fn e)
+            (recur timeout-ch (cond-> (conj coll e) distinct-coll? distinct))
+            (recur timeout-ch coll))
+
+          (or stop
+              ;; got nil from in-ch, means in-ch is closed
+              ;; so we stop the whole go-loop
+              (nil? e))
+          (async/close! ch))))
+    ch))
+
+(defn drain-chan
+  "drop all stuffs in CH, and return all of them"
+  [ch]
+  (->> (repeatedly #(async/poll! ch))
+       (take-while identity)))
+
 #?(:cljs
    (defn trace!
      []