3pp-mirror/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2025-05-16 07:46:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

✨ Improve how topic is managed on ws notifications

Browse source
This commit is contained in:
Andrey Antukh 2024-10-30 12:32:03 +01:00 • committed by Alonso Torres
parent 9409078069
commit 97a1bf15ef
6 changed files with 16 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

9
backend/src/app/http/websocket.clj
View file

@ -112,7 +112,6 @@
fsub (::file-subscription @state) fsub (::file-subscription @state)
tsub (::team-subscription @state) tsub (::team-subscription @state)
msg {:type :disconnect msg {:type :disconnect
:subs-id profile-id
:profile-id profile-id :profile-id profile-id
:session-id session-id}] :session-id session-id}]
@ -137,9 +136,7 @@
(l/trace :fn "handle-message" :event "subscribe-team" :team-id team-id :conn-id id) (l/trace :fn "handle-message" :event "subscribe-team" :team-id team-id :conn-id id)
(let [prev-subs (get @state ::team-subscription) (let [prev-subs (get @state ::team-subscription)
channel (sp/chan :buf (sp/dropping-buffer 64) channel (sp/chan :buf (sp/dropping-buffer 64)
:xf (comp :xf (remove #(= (:session-id %) session-id)))]
(remove #(= (:session-id %) session-id))
(map #(assoc % :subs-id team-id))))]
(sp/pipe channel output-ch false) (sp/pipe channel output-ch false)
(mbus/sub! msgbus :topic team-id :chan channel) (mbus/sub! msgbus :topic team-id :chan channel)
@ -158,8 +155,7 @@
(l/trace :fn "handle-message" :event "subscribe-file" :file-id file-id :conn-id id) (l/trace :fn "handle-message" :event "subscribe-file" :file-id file-id :conn-id id)
(let [psub (::file-subscription @state) (let [psub (::file-subscription @state)
fch (sp/chan :buf (sp/dropping-buffer 64) fch (sp/chan :buf (sp/dropping-buffer 64)
:xf (comp (remove #(= (:session-id %) session-id)) :xf (remove #(= (:session-id %) session-id)))]
(map #(assoc % :subs-id file-id))))]
(let [subs {:file-id file-id :channel fch :topic file-id}] (let [subs {:file-id file-id :channel fch :topic file-id}]
(swap! state assoc ::file-subscription subs)) (swap! state assoc ::file-subscription subs))
@ -190,7 +186,6 @@
;; Notifify the rest of participants of the new connection. ;; Notifify the rest of participants of the new connection.
(let [message {:type :join-file (let [message {:type :join-file
:file-id file-id :file-id file-id
:subs-id file-id
:session-id session-id :session-id session-id
:profile-id profile-id}] :profile-id profile-id}]
(mbus/pub! msgbus :topic file-id :message message)))) (mbus/pub! msgbus :topic file-id :message message))))

6
backend/src/app/msgbus.clj
View file

@ -99,8 +99,9 @@
nil)) nil))
(defn pub! (defn pub!
[{::keys [pub-ch]} & {:as params}] [{::keys [pub-ch]} & {:keys [topic] :as params}]
(sp/put! pub-ch params)) (let [params (update params :message assoc :topic topic)]
(sp/put! pub-ch params)))
(defn purge! (defn purge!
[{:keys [::state ::wrk/executor] :as msgbus} chans] [{:keys [::state ::wrk/executor] :as msgbus} chans]
@ -230,7 +231,6 @@
(l/debug :hint "io-loop thread terminated"))))) (l/debug :hint "io-loop thread terminated")))))
(defn- redis-pub! (defn- redis-pub!
"Publish a message to the redis server. Asynchronous operation, "Publish a message to the redis server. Asynchronous operation,
intended to be used in core.async go blocks." intended to be used in core.async go blocks."

3
backend/src/app/rpc/commands/teams.clj
View file

@ -653,7 +653,7 @@
(mbus/pub! msgbus (mbus/pub! msgbus
:topic member-id :topic member-id
:message {:type :team-role-change :message {:type :team-role-change
:subs-id member-id :topic member-id
:team-id team-id :team-id team-id
:role role}) :role role})
@ -713,7 +713,6 @@
:topic member-id :topic member-id
:message {:type :team-membership-change :message {:type :team-membership-change
:change :removed :change :removed
:subs-id member-id
:team-id team-id :team-id team-id
:team-name (:name team)}) :team-name (:name team)})

2
frontend/playwright/data/workspace/ws-notifications.js
View file

@ -3,5 +3,5 @@ export const presenceFixture = {
"~:file-id": "~uc7ce0794-0992-8105-8004-38f280443849", "~:file-id": "~uc7ce0794-0992-8105-8004-38f280443849",
"~:session-id": "~u37730924-d520-80f1-8004-4ae6e5c3942d", "~:session-id": "~u37730924-d520-80f1-8004-4ae6e5c3942d",
"~:profile-id": "~uc7ce0794-0992-8105-8004-38e630f29a9b", "~:profile-id": "~uc7ce0794-0992-8105-8004-38e630f29a9b",
"~:subs-id": "~uc7ce0794-0992-8105-8004-38f280443849", "~:topic": "~uc7ce0794-0992-8105-8004-38f280443849",
}; };

6
frontend/src/app/main/data/dashboard.cljs
View file

@ -80,9 +80,9 @@
(->> stream (->> stream
(rx/filter (ptk/type? ::dws/message)) (rx/filter (ptk/type? ::dws/message))
(rx/map deref) (rx/map deref)
(rx/filter (fn [{:keys [subs-id] :as msg}] (rx/filter (fn [{:keys [topic] :as msg}]
(or (= subs-id uuid/zero) (or (= topic uuid/zero)
(= subs-id profile-id)))) (= topic profile-id))))
(rx/map process-message)) (rx/map process-message))
;; Once the teams are fecthed, initialize features related ;; Once the teams are fecthed, initialize features related

11
frontend/src/app/main/data/workspace/notifications.cljs
View file

@ -65,15 +65,16 @@
(->> (rx/from initmsg) (->> (rx/from initmsg)
(rx/map dws/send)) (rx/map dws/send))
;; Subscribe to notifications of the subscription ;; Subscribe to notifications of the subscription
(->> stream (->> stream
(rx/filter (ptk/type? ::dws/message)) (rx/filter (ptk/type? ::dws/message))
(rx/map deref) (rx/map deref)
(rx/filter (fn [{:keys [subs-id] :as msg}] (rx/filter (fn [{:keys [topic] :as msg}]
(or (= subs-id uuid/zero) (or (= topic uuid/zero)
(= subs-id profile-id) (= topic profile-id)
(= subs-id team-id) (= topic team-id)
(= subs-id file-id)))) (= topic file-id))))
(rx/map process-message)) (rx/map process-message))
;; On reconnect, send again the subscription messages ;; On reconnect, send again the subscription messages