From 24d0b68c1f6c40cca99e030a3937786ee5aca98e Mon Sep 17 00:00:00 2001
From: janherich <jan.herich@gmail.com>
Date: Thu, 11 Jan 2018 15:55:55 +0100
Subject: [PATCH] Async audit

---
 src/status_im/chat/handlers.cljs             |  6 +-
 src/status_im/native_module/impl/module.cljs |  3 +-
 src/status_im/protocol/web3/delivery.cljs    | 60 +++++++++++---------
 src/status_im/utils/event.cljs               |  9 ---
 src/status_im/utils/pre_receiver.cljs        | 11 ++--
 src/status_im/utils/scheduler.cljs           | 10 ----
 6 files changed, 40 insertions(+), 59 deletions(-)
 delete mode 100644 src/status_im/utils/event.cljs
 delete mode 100644 src/status_im/utils/scheduler.cljs

diff --git a/src/status_im/chat/handlers.cljs b/src/status_im/chat/handlers.cljs
index 66af0acf7c..7ea23394ae 100644
--- a/src/status_im/chat/handlers.cljs
+++ b/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]
+(ns status-im.chat.handlers 
+  (:require [re-frame.core :refer [enrich after debug dispatch reg-fx]] 
             [clojure.string :as string]
             [status-im.ui.components.styles :refer [default-chat-color]]
             [status-im.chat.constants :as chat-consts]
diff --git a/src/status_im/native_module/impl/module.cljs b/src/status_im/native_module/impl/module.cljs
index 6e645190ef..58fe3b5cd6 100644
--- a/src/status_im/native_module/impl/module.cljs
+++ b/src/status_im/native_module/impl/module.cljs
@@ -6,8 +6,7 @@
             [taoensso.timbre :as log]
             [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.platform :as p] 
             [status-im.utils.types :as types]
             [status-im.utils.transducers :as transducers]
             [status-im.utils.async :as async-util]
diff --git a/src/status_im/protocol/web3/delivery.cljs b/src/status_im/protocol/web3/delivery.cljs
index cb5228ced7..13a5ab1519 100644
--- a/src/status_im/protocol/web3/delivery.cljs
+++ b/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 [cljs.core.async :refer [<! timeout]]
+  (:require-macros [cljs.core.async.macros :as async])
+  (: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!
-  [callback]
-  (reset! pending-message-callback callback))
+(def ^:private pending-message-queue (async/chan 100))
 
-(defn add-pending-message!
-  [web3 {:keys [type message-id requires-ack? to ack?] :as message}]
-  {:pre [(valid? :protocol/message message)]}
-  (go
-    (debug :add-pending-message! message)
-    ;; encryption can take some time, better to run asynchronously
+(async/go-loop [[web3 {:keys [type message-id requires-ack? to ack?] :as message}]
+                (async/<! pending-message-queue)]
+  (when message
     (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,26 +235,25 @@
     (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)
-                      callback (delivery-callback web3 post-error-callback data message')]
-                  (t/post-message! web3 message' callback))
-                (catch :default err
-                  (log/error :post-message-error err))
-                (finally
-                  (attempt-was-made! web3 id to)))))))
+          ;; check each message asynchronously 
+          (when (should-be-retransmitted? options data)
+            (try
+              (let [message' (check-ttl message type ttl-config default-ttl)
+                    callback (delivery-callback web3 post-error-callback data message')]
+                (t/post-message! web3 message' callback))
+              (catch :default err
+                (log/error :post-message-error err))
+              (finally
+                (attempt-was-made! web3 id to))))))
       (when-not @stop?
-        (recur (<! (timeout delivery-loop-ms-interval)))))
-    (go-loop [_ nil]
+        (recur (async/<! (async/timeout delivery-loop-ms-interval)))))
+    (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)]
diff --git a/src/status_im/utils/event.cljs b/src/status_im/utils/event.cljs
deleted file mode 100644
index 25e1775b99..0000000000
--- a/src/status_im/utils/event.cljs
+++ /dev/null
@@ -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))))))
\ No newline at end of file
diff --git a/src/status_im/utils/pre_receiver.cljs b/src/status_im/utils/pre_receiver.cljs
index 17c4e93859..86f235387e 100644
--- a/src/status_im/utils/pre_receiver.cljs
+++ b/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))
-        (recur)))
+          (add-fn msg)))
+      (recur))
     in-ch))
diff --git a/src/status_im/utils/scheduler.cljs b/src/status_im/utils/scheduler.cljs
deleted file mode 100644
index c67841c227..0000000000
--- a/src/status_im/utils/scheduler.cljs
+++ /dev/null
@@ -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)))