Async audit

src/status_im/chat/handlers.cljs

@@ -1,7 +1,5 @@
 (ns status-im.chat.handlers 
-  (:require-macros [cljs.core.async.macros :as am])
   (:require [re-frame.core :refer [enrich after debug dispatch reg-fx]] 
-            [cljs.core.async :as a]
             [clojure.string :as string]
             [status-im.ui.components.styles :refer [default-chat-color]]
             [status-im.chat.constants :as chat-consts]

src/status_im/native_module/impl/module.cljs

@@ -7,7 +7,6 @@
             [cljs.core.async :as async :refer [<! timeout]]
             [status-im.utils.js-resources :as js-res]
             [status-im.utils.platform :as p] 
-            [status-im.utils.scheduler :as scheduler]
             [status-im.utils.types :as types]
             [status-im.utils.transducers :as transducers]
             [status-im.utils.async :as async-util]

src/status_im/protocol/web3/delivery.cljs

@@ -1,6 +1,6 @@
 (ns status-im.protocol.web3.delivery
-  (:require-macros [cljs.core.async.macros :refer [go-loop go]])
+  (:require-macros [cljs.core.async.macros :as async])
-  (:require [cljs.core.async :refer [<! timeout]]
+  (:require [cljs.core.async :as async]
             [status-im.protocol.web3.transport :as t]
             [status-im.protocol.web3.utils :as u]
             [status-im.protocol.encryption :as e]
@@ -53,16 +53,11 @@
 (defonce pending-message-callback (atom nil))
 (defonce recipient->pending-message (atom {}))
 
-(defn set-pending-mesage-callback!
+(def ^:private pending-message-queue (async/chan 100))
-  [callback]
-  (reset! pending-message-callback callback))
 
-(defn add-pending-message!
+(async/go-loop [[web3 {:keys [type message-id requires-ack? to ack?] :as message}]
-  [web3 {:keys [type message-id requires-ack? to ack?] :as message}]
+                (async/<! pending-message-queue)]
-  {:pre [(valid? :protocol/message message)]}
+  (when message
-  (go
-    (debug :add-pending-message! message)
-    ;; encryption can take some time, better to run asynchronously
     (prepare-message
       web3 message
       (fn [message']
@@ -82,7 +77,19 @@
             (swap! messages assoc-in [web3 message-id to] pending-message)
             (when to
               (swap! recipient->pending-message
-                     update to set/union #{[web3 message-id to]}))))))))
+                     update to set/union #{[web3 message-id to]}))))))
+    (recur (async/<! pending-message-queue))))
+
+(defn set-pending-mesage-callback!
+  [callback]
+  (reset! pending-message-callback callback))
+
+(defn add-pending-message!
+  [web3 message]
+  {:pre [(valid? :protocol/message message)]} 
+  (debug :add-pending-message! message)
+    ;; encryption can take some time, better to run asynchronously
+    (async/put! pending-message-queue [web3 message]))
 
 (s/def :delivery/pending-message
   (s/keys :req-un [:message/sig :message/to :shh/payload :payload/ack? ::id
@@ -228,11 +235,10 @@
     (reset! loop-state stop?)
     ;; go go!!!
     (debug :init-loop)
-    (go-loop [_ nil]
+    (async/go-loop [_ nil]
       (doseq [[_ messages] (@messages web3)]
         (doseq [[_ {:keys [id message to type] :as data}] messages]
           ;; check each message asynchronously 
-          (go
           (when (should-be-retransmitted? options data)
             (try
               (let [message' (check-ttl message type ttl-config default-ttl)
@@ -241,13 +247,13 @@
               (catch :default err
                 (log/error :post-message-error err))
               (finally
-                  (attempt-was-made! web3 id to)))))))
+                (attempt-was-made! web3 id to))))))
       (when-not @stop?
-        (recur (<! (timeout delivery-loop-ms-interval)))))
+        (recur (async/<! (async/timeout delivery-loop-ms-interval)))))
-    (go-loop [_ nil]
+    (async/go-loop [_ nil]
       (when-not @stop?
         (online-message)
-        (recur (<! (timeout (* 1000 send-online-s-interval))))))))
+        (recur (async/<! (async/timeout (* 1000 send-online-s-interval))))))))
 
 (defn reset-pending-messages! [to]
   (doseq [key (@recipient->pending-message to)]

src/status_im/utils/event.cljs

@@ -1,9 +0,0 @@
-(ns status-im.utils.event
-  (:require [cljs.core.async :refer [<!]])
-  (:require-macros [cljs.core.async.macros :refer [go]]))
-
-(defn handle-channel-events [chan handler]
-  (go (loop [[message args] (<! chan)]
-        (when message
-          (handler message args)
-          (recur (<! chan))))))

src/status_im/utils/pre_receiver.cljs

@@ -8,10 +8,6 @@
 (defn- add-message-mock [{:keys [id clock-value] :as msg}]
   (log/debug "add-message-mock:" id clock-value))
 
-(defn- delay-message [msg out ms]
-  (async/go (async/<! (async/timeout ms))
-            (async/put! out msg)))
-
 (defn- earliest-clock-value-seen? [seen id clock-value]
   (->> seen
        (filter (fn [[_ x]] (= x id)))
@@ -30,7 +26,8 @@
     (async/go-loop []
       (let [{:keys [id clock-value] :as msg} (async/<! in-ch)]
         (swap! seen conj [clock-value id])
-        (delay-message msg mature-ch delay-ms))
+        (async/<! (async/timeout delay-ms))
+        (async/put! mature-ch msg))
       (recur))
     (async/go-loop []
       (let [{:keys [id clock-value] :as msg} (async/<! mature-ch)]
@@ -39,6 +36,6 @@
             (do (swap! seen disj [clock-value id])
                 (add-fn msg))
             (async/put! mature-ch msg))
-          (add-fn msg))
+          (add-fn msg)))
-        (recur)))
+      (recur))
     in-ch))

src/status_im/utils/scheduler.cljs

@@ -1,10 +0,0 @@
-(ns status-im.utils.scheduler
-  (:require-macros [cljs.core.async.macros :refer [go]])
-  (:require [cljs.core.async :refer [<! timeout]]))
-
-(defn s->ms [s] (* 1000 s))
-
-(defn execute-later
-  [function timeout-ms]
-  (go (<! (timeout timeout-ms))
-      (function)))