3pp-mirror/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2025-06-05 22:01:38 +02:00
Code Issues Projects Releases Packages Wiki Activity

✨ Improve websocket notifications metrics.

Browse source
This commit is contained in:
Andrey Antukh 2021-01-25 20:15:28 +01:00 • committed by Hirunatan
parent f8b349814c
commit 8236d84dfa
1 changed files with 65 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

79
backend/src/app/notifications.clj
View file

@ -41,19 +41,32 @@
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ {:keys [session metrics] :as cfg}] [_ {:keys [session metrics] :as cfg}]
(let [wrap-session (:middleware session) (let [wrap-session (:middleware session)
mtx-active-conn (mtx/create
{:name "http_ws_notifications_active_connections" mtx-active-connections
(mtx/create
{:name "websocket_notifications_active_connections"
:registry (:registry metrics) :registry (:registry metrics)
:type :gauge :type :gauge
:help "Active websocket connections on notifications service."}) :help "Active websocket connections on notifications service."})
mtx-msg-counter (mtx/create
{:name "http_ws_notifications_message_counter" mtx-message-recv
(mtx/create
{:name "websocket_notifications_message_recv_timing"
:registry (:registry metrics) :registry (:registry metrics)
:type :counter :type :summary
:help "Counter of total messages processed on websocket conenction on notifications service."}) :help "Message receive summary timing (ms)."})
mtx-message-send
(mtx/create
{:name "websocket_notifications_message_send_timing"
:registry (:registry metrics)
:type :summary
:help "Message receive summary timing (ms)."})
cfg (assoc cfg cfg (assoc cfg
:mtx-active-conn mtx-active-conn :mtx-active-connections mtx-active-connections
:mtx-msg-counter mtx-msg-counter)] :mtx-message-recv mtx-message-recv
:mtx-message-send mtx-message-send)]
(-> #(handler cfg %) (-> #(handler cfg %)
(wrap-session) (wrap-session)
(wrap-keyword-params) (wrap-keyword-params)
@ -100,18 +113,10 @@
;; WebSocket Http Handler ;; WebSocket Http Handler
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(declare on-connect) (declare handle-connect)
(defrecord WebSocket [conn in out sub]) (defrecord WebSocket [conn in out sub])
;; (defonce metrics-active-connections
;; (mtx/gauge {:id "notificatons__active_connections"
;; :help "Active connections to the notifications service."}))
;; (defonce metrics-message-counter
;; (mtx/counter {:id "notificatons__messages_counter"
;; :help "A total number of messages handled by the notifications service."}))
(defn- ws-send (defn- ws-send
[conn data] [conn data]
(try (try
@ -122,12 +127,17 @@
false))) false)))
(defn websocket (defn websocket
[{:keys [file-id team-id redis mtx-active-conn mtx-msg-counter] :as cfg}] [{:keys [file-id team-id redis] :as cfg}]
(let [in (a/chan 32) (let [in (a/chan 32)
out (a/chan 32)] out (a/chan 32)
{:on-connect mtx-active-connections (:mtx-active-connections cfg)
(fn [conn] mtx-message-send (:mtx-message-send cfg)
(mtx-active-conn :inc) mtx-message-recv (:mtx-message-recv cfg)
ws-send (mtx/wrap-summary ws-send mtx-message-send)]
(letfn [(on-connect [conn]
(mtx-active-connections :inc)
(let [sub (rd/subscribe redis {:xform (map t/decode-str) (let [sub (rd/subscribe redis {:xform (map t/decode-str)
:topics [file-id team-id]}) :topics [file-id team-id]})
ws (WebSocket. conn in out sub nil cfg)] ws (WebSocket. conn in out sub nil cfg)]
@ -140,33 +150,32 @@
(recur))))) (recur)))))
(a/go (a/go
(a/<! (on-connect ws)) (a/<! (handle-connect ws))
(a/close! sub)))) (a/close! sub))))
:on-error (on-error [_conn _e]
(fn [_conn _e]
(a/close! out) (a/close! out)
(a/close! in)) (a/close! in))
:on-close (on-close [_conn _status _reason]
(fn [_conn _status _reason] (mtx-active-connections :dec)
(mtx-active-conn :dec)
(a/close! out) (a/close! out)
(a/close! in)) (a/close! in))
:on-text (on-message [_ws message]
(fn [_ws message]
(mtx-msg-counter :inc)
(let [message (t/decode-str message)] (let [message (t/decode-str message)]
(a/>!! in message))) (a/>!! in message)))]
:on-bytes {:on-connect on-connect
(constantly nil)})) :on-error on-error
:on-close on-close
:on-text (mtx/wrap-summary on-message mtx-message-recv)
:on-bytes (constantly nil)})))
(declare handle-message) (declare handle-message)
(declare start-loop!) (declare start-loop!)
(defn- on-connect (defn- handle-connect
[{:keys [conn] :as ws}] [{:keys [conn] :as ws}]
(a/go (a/go
(try (try