3pp-mirror/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2025-06-21 14:16:59 +02:00
Code Issues Projects Releases Packages Wiki Activity

✨ Improve events/sse internal API

Browse source 
For make code cleaner and more evident for a quick view
This commit is contained in:
Andrey Antukh 2025-05-15 09:07:56 +02:00
parent 6524e75770
commit cf274099c4
2 changed files with 29 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

29
backend/src/app/http/sse.clj
View file

@ -9,7 +9,6 @@
(:refer-clojure :exclude [tap])
(:require
[app.common.data :as d]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.common.transit :as t]
[app.http.errors :as errors]
@ -54,16 +53,20 @@
::yres/status 200
::yres/body (yres/stream-body
(fn [_ output]
(binding [events/*channel* (sp/chan :buf buf :xf (keep encode))]
(let [listener (events/start-listener
(partial write! output)
(partial pu/close! output))]
(try
(let [channel (sp/chan :buf buf :xf (keep encode))
listener (events/start-listener
channel
(partial write! output)
(partial pu/close! output))]
(try
(binding [events/*channel* channel]
(let [result (handler)]
(events/tap :end result))
(catch Throwable cause
(let [result (errors/handle' cause request)]
(events/tap :error result)))
(finally
(sp/close! events/*channel*)
(px/await! listener)))))))}))
(events/tap :end result)))
(catch Throwable cause
(let [result (errors/handle' cause request)]
(events/tap channel :error result)))
(finally
(sp/close! channel)
(px/await! listener))))))}))

30
backend/src/app/util/events.clj
View file

@ -10,7 +10,6 @@
to them. Mainly used in http.sse for progress reporting."
(:refer-clojure :exclude [tap run!])
(:require
[app.common.data.macros :as dm]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[promesa.exec :as px]
@ -18,33 +17,30 @@
(def ^:dynamic *channel* nil)
(defn channel
[]
(sp/chan :buf 32))
(defn tap
[type data]
(when-let [channel *channel*]
(sp/put! channel [type data])
nil))
([type data]
(when-let [channel *channel*]
(sp/put! channel [type data])
nil))
([channel type data]
(when channel
(sp/put! channel [type data])
nil)))
(defn start-listener
[on-event on-close]
(dm/assert!
"expected active events channel"
(sp/chan? *channel*))
[channel on-event on-close]
(assert (sp/chan? channel) "expected active events channel")
(px/thread
{:virtual true}
(try
(loop []
(when-let [event (sp/take! *channel*)]
(when-let [event (sp/take! channel)]
(let [result (ex/try! (on-event event))]
(if (ex/exception? result)
(do
(l/wrn :hint "unexpected exception" :cause result)
(sp/close! *channel*))
(sp/close! channel))
(recur)))))
(finally
(on-close)))))
@ -55,7 +51,7 @@
[f on-event]
(binding [*channel* (sp/chan :buf 32)]
(let [listener (start-listener on-event (constantly nil))]
(let [listener (start-listener *channel* on-event (constantly nil))]
(try
(f)
(finally