parent
00d665b2d5
commit
87dc4ead85
|
@ -1,5 +1,7 @@
|
|||
(ns status-im.chat.handlers
|
||||
(:require [re-frame.core :refer [enrich after debug dispatch reg-fx]]
|
||||
(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]
|
||||
|
|
|
@ -6,7 +6,8 @@
|
|||
[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.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]
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
(ns status-im.protocol.web3.delivery
|
||||
(:require-macros [cljs.core.async.macros :as async])
|
||||
(:require [cljs.core.async :as async]
|
||||
(:require-macros [cljs.core.async.macros :refer [go-loop go]])
|
||||
(:require [cljs.core.async :refer [<! timeout]]
|
||||
[status-im.protocol.web3.transport :as t]
|
||||
[status-im.protocol.web3.utils :as u]
|
||||
[status-im.protocol.encryption :as e]
|
||||
|
@ -53,11 +53,16 @@
|
|||
(defonce pending-message-callback (atom nil))
|
||||
(defonce recipient->pending-message (atom {}))
|
||||
|
||||
(def ^:private pending-message-queue (async/chan 100))
|
||||
(defn set-pending-mesage-callback!
|
||||
[callback]
|
||||
(reset! pending-message-callback callback))
|
||||
|
||||
(async/go-loop [[web3 {:keys [type message-id requires-ack? to ack?] :as message}]
|
||||
(async/<! pending-message-queue)]
|
||||
(when message
|
||||
(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
|
||||
(prepare-message
|
||||
web3 message
|
||||
(fn [message']
|
||||
|
@ -77,19 +82,7 @@
|
|||
(swap! messages assoc-in [web3 message-id to] pending-message)
|
||||
(when to
|
||||
(swap! recipient->pending-message
|
||||
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]))
|
||||
update to set/union #{[web3 message-id to]}))))))))
|
||||
|
||||
(s/def :delivery/pending-message
|
||||
(s/keys :req-un [:message/sig :message/to :shh/payload :payload/ack? ::id
|
||||
|
@ -235,25 +228,26 @@
|
|||
(reset! loop-state stop?)
|
||||
;; go go!!!
|
||||
(debug :init-loop)
|
||||
(async/go-loop [_ nil]
|
||||
(go-loop [_ nil]
|
||||
(doseq [[_ messages] (@messages web3)]
|
||||
(doseq [[_ {:keys [id message to type] :as data}] messages]
|
||||
;; 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))))))
|
||||
;; 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)))))))
|
||||
(when-not @stop?
|
||||
(recur (async/<! (async/timeout delivery-loop-ms-interval)))))
|
||||
(async/go-loop [_ nil]
|
||||
(recur (<! (timeout delivery-loop-ms-interval)))))
|
||||
(go-loop [_ nil]
|
||||
(when-not @stop?
|
||||
(online-message)
|
||||
(recur (async/<! (async/timeout (* 1000 send-online-s-interval))))))))
|
||||
(recur (<! (timeout (* 1000 send-online-s-interval))))))))
|
||||
|
||||
(defn reset-pending-messages! [to]
|
||||
(doseq [key (@recipient->pending-message to)]
|
||||
|
|
|
@ -0,0 +1,9 @@
|
|||
(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))))))
|
|
@ -8,6 +8,10 @@
|
|||
(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)))
|
||||
|
@ -26,8 +30,7 @@
|
|||
(async/go-loop []
|
||||
(let [{:keys [id clock-value] :as msg} (async/<! in-ch)]
|
||||
(swap! seen conj [clock-value id])
|
||||
(async/<! (async/timeout ms))
|
||||
(async/put! mature-ch msg))
|
||||
(delay-message msg mature-ch delay-ms))
|
||||
(recur))
|
||||
(async/go-loop []
|
||||
(let [{:keys [id clock-value] :as msg} (async/<! mature-ch)]
|
||||
|
|
|
@ -0,0 +1,10 @@
|
|||
(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)))
|
Loading…
Reference in New Issue