Penpot/penpot
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2025-04-03 10:31:38 -05:00
Code Issues Activity

✨ Add the ability to stream events on rpc methods

Browse source
This commit is contained in:
Andrey Antukh 2023-12-01 16:18:39 +01:00 • committed by Andrés Moya
parent f3e9efa6fe
commit 03518a8da1
21 changed files with 408 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
backend/resources/app/templates/api-doc-entry.tmpl
View file

@ -25,6 +25,12 @@
<span>SCHEMA</span>
</span>
{% endif %}
{% if item.sse %}
<span class="tag">
<span>SSE</span>
</span>
{% endif %}
</div>
</div>
<div class="rpc-row-detail hidden">

11
backend/resources/app/templates/api-doc.tmpl
View file

@ -56,6 +56,17 @@
desired content-type on the <b>`Accept`</b> header, the transit encoding is used
by default.</p>
<h3>SSE (Server-Sent Events)</h3>
<p>The methods marked with <b>SSE</b> returns
a <a href="https://html.spec.whatwg.org/multipage/server-sent-events.html"> SSE
formatted</a> stream on the response body, always with status 200. The events are
always encoded using `application/transit+json` encoding (for now no content
negotiation is possible on methods that return SSE streams). </p>
<p>On the javascript side you can use
the <a href="https://github.com/rexxars/eventsource-parser">eventsoure-parser</a>
library for propertly parsing the response body using the
standard <a href="https://developer.mozilla.org/en-US/docs/Web/API/Fetch_API">Fetch
API</a></p>
<h3>Limits</h3>
<p>The rate limit work per user basis (this means that different api keys share

12
backend/src/app/features/components_v2.clj
View file

@ -32,6 +32,7 @@
[app.common.types.shape-tree :as ctst]
[app.common.uuid :as uuid]
[app.db :as db]
[app.http.sse :as sse]
[app.media :as media]
[app.rpc.commands.files :as files]
[app.rpc.commands.files-snapshot :as fsnap]
@ -362,6 +363,8 @@
shapes from library components. Mark the file with
the :components-v2 option."
[file-data libraries]
(sse/tap {:type :migration-progress
:section :components})
(let [components (ctkl/components-seq file-data)]
(if (empty? components)
(assoc-in file-data [:options :components-v2] true)
@ -435,6 +438,9 @@
add-instance-grid
(fn [fdata frame-id grid assets]
(reduce (fn [result [component position]]
(sse/tap {:type :migration-progress
:section :components
:name (:name component)})
(add-main-instance result component frame-id (gpt/add position
(gpt/point grid-gap grid-gap))))
fdata
@ -701,6 +707,9 @@
(->> (d/zip media-group grid)
(map (fn [[mobj position]]
(l/trc :hint "submit graphic processing" :file-id (str (:id fdata)) :id (str (:id mobj)))
(sse/tap {:type :migration-progress
:section :graphics
:name (:name mobj)})
(px/submit! executor (partial process mobj position))))
(reduce (fn [fdata promise]
(if-let [changes (deref promise)]
@ -713,6 +722,8 @@
(defn- migrate-graphics
[fdata]
(sse/tap {:type :migration-progress
:section :graphics})
(if (empty? (:media fdata))
fdata
(let [[fdata page-id start-pos]
@ -812,7 +823,6 @@
(defn migrate-file!
[system file-id & {:keys [validate? throw-on-validate?]}]
(let [tpoint (dt/tpoint)
file-id (if (string? file-id)
(parse-uuid file-id)

4
backend/src/app/http/errors.clj
View file

@ -232,3 +232,7 @@
(if (ex/error? cause)
(handle-error cause request nil)
(handle-exception cause request nil)))
(defn handle'
[cause request]
(::rres/body (handle cause request)))

86
backend/src/app/http/sse.clj Normal file
View file

@ -0,0 +1,86 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC
(ns app.http.sse
"SSE (server sent events) helpers"
(: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]
[promesa.core :as p]
[promesa.exec :as px]
[promesa.exec.csp :as sp]
[promesa.util :as pu]
[ring.response :as rres])
(:import
java.io.OutputStream))
(def ^:dynamic *channel* nil)
(defn- write!
[^OutputStream output ^bytes data]
(l/trc :hint "writting data" :data data :length (alength data))
(.write output data)
(.flush output))
(defn- create-writer-loop
[^OutputStream output]
(try
(loop []
(when-let [event (sp/take! *channel*)]
(let [result (ex/try! (write! output event))]
(if (ex/exception? result)
(l/wrn :hint "unexpected exception on sse writer" :cause result)
(recur)))))
(finally
(pu/close! output))))
(defn- encode
[[name data]]
(try
(let [data (with-out-str
(println "event:" (d/name name))
(println "data:" (t/encode-str data {:type :json-verbose}))
(println))]
(.getBytes data "UTF-8"))
(catch Throwable cause
(l/err :hint "unexpected error on encoding value on sse stream"
:cause cause)
nil)))
;; ---- PUBLIC API
(def default-headers
{"Content-Type" "text/event-stream;charset=UTF-8"
"Cache-Control" "no-cache, no-store, max-age=0, must-revalidate"
"Pragma" "no-cache"})
(defn tap
([data] (tap "event" data))
([name data]
(when-let [channel *channel*]
(sp/put! channel [name data])
nil)))
(defn response
[handler & {:keys [buf] :or {buf 32} :as opts}]
(fn [request]
{::rres/headers default-headers
::rres/status 200
::rres/body (reify rres/StreamableResponseBody
(-write-body-to-stream [_ _ output]
(binding [*channel* (sp/chan :buf buf :xf (keep encode))]
(let [writer (px/run! :virtual (partial create-writer-loop output))]
(try
(tap "end" (handler))
(catch Throwable cause
(tap "error" (errors/handle' cause request)))
(finally
(sp/close! *channel*)
(p/await! writer)))))))}))

18
backend/src/app/rpc.clj
View file

@ -57,14 +57,16 @@
(defn- handle-response
[request result]
(if (fn? result)
(result request)
(let [mdata (meta result)]
(-> {::rres/status (::http/status mdata 200)
::rres/headers (::http/headers mdata {})
::rres/body (rph/unwrap result)}
(handle-response-transformation request mdata)
(handle-before-comple-hook mdata)))))
(let [mdata (meta result)
response (if (fn? result)
(result request)
(let [result (rph/unwrap result)]
{::rres/status (::http/status mdata 200)
::rres/headers (::http/headers mdata {})
::rres/body result}))]
(-> response
(handle-response-transformation request mdata)
(handle-before-comple-hook mdata))))
(defn- rpc-handler
"Ring handler that dispatches cmd requests and convert between

101
backend/src/app/rpc/commands/binfile.clj
View file

@ -15,6 +15,7 @@
[app.common.files.validate :as fval]
[app.common.fressian :as fres]
[app.common.logging :as l]
[app.common.schema :as sm]
[app.common.spec :as us]
[app.common.types.file :as ctf]
[app.common.uuid :as uuid]
@ -22,6 +23,7 @@
[app.db :as db]
[app.features.components-v2 :as features.components-v2]
[app.features.fdata :as features.fdata]
[app.http.sse :as sse]
[app.loggers.audit :as-alias audit]
[app.loggers.webhooks :as-alias webhooks]
[app.media :as media]
@ -30,7 +32,6 @@
[app.rpc.commands.projects :as projects]
[app.rpc.commands.teams :as teams]
[app.rpc.doc :as-alias doc]
[app.rpc.helpers :as rph]
[app.storage :as sto]
[app.storage.tmp :as tmp]
[app.tasks.file-gc]
@ -38,11 +39,13 @@
[app.util.pointer-map :as pmap]
[app.util.services :as sv]
[app.util.time :as dt]
[app.worker :as-alias wrk]
[clojure.set :as set]
[clojure.spec.alpha :as s]
[clojure.walk :as walk]
[cuerdas.core :as str]
[datoteka.io :as io]
[promesa.core :as p]
[promesa.util :as pu]
[ring.response :as rres]
[yetti.adapter :as yt])
@ -642,6 +645,9 @@
validate? (contains? cf/flags :file-validation)
features (cfeat/get-team-enabled-features cf/flags team)]
(sse/tap {:type :import-progress
:section :read-import})
;; Process all sections
(run! (fn [section]
(l/dbg :hint "reading section" :section section ::l/sync? true)
@ -651,6 +657,8 @@
(assoc ::section section)
(assoc ::input input))]
(binding [*options* options]
(sse/tap {:type :import-progress
:section section})
(read-section options))))
[:v1/metadata :v1/files :v1/rels :v1/sobjects])
@ -1056,54 +1064,71 @@
;; --- Command: export-binfile
(s/def ::file-id ::us/uuid)
(s/def ::include-libraries? ::us/boolean)
(s/def ::embed-assets? ::us/boolean)
(s/def ::export-binfile
(s/keys :req [::rpc/profile-id]
:req-un [::file-id ::include-libraries? ::embed-assets?]))
(def ^:private
schema:export-binfile
(sm/define
[:map {:title "export-binfile"}
[:file-id ::sm/uuid]
[:include-libraries? :boolean]
[:embed-assets? :boolean]]))
(sv/defmethod ::export-binfile
"Export a penpot file in a binary format."
{::doc/added "1.15"
::webhooks/event? true}
::webhooks/event? true
::sm/result schema:export-binfile}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id file-id include-libraries? embed-assets?] :as params}]
(files/check-read-permissions! pool profile-id file-id)
(let [body (reify rres/StreamableResponseBody
(-write-body-to-stream [_ _ output-stream]
(-> cfg
(assoc ::file-ids [file-id])
(assoc ::embed-assets? embed-assets?)
(assoc ::include-libraries? include-libraries?)
(export! output-stream))))]
(fn [_]
{::rres/status 200
::rres/headers {"content-type" "application/octet-stream"}
::rres/body (reify rres/StreamableResponseBody
(-write-body-to-stream [_ _ output-stream]
(-> cfg
(assoc ::file-ids [file-id])
(assoc ::embed-assets? embed-assets?)
(assoc ::include-libraries? include-libraries?)
(export! output-stream))))}))
(fn [_]
{::rres/status 200
::rres/body body
::rres/headers {"content-type" "application/octet-stream"}})))
(s/def ::file ::media/upload)
(s/def ::import-binfile
(s/keys :req [::rpc/profile-id]
:req-un [::project-id ::file]))
;; --- Command: import-binfile
(def ^:private
schema:import-binfile
(sm/define
[:map {:title "import-binfile"}
[:project-id ::sm/uuid]
[:file ::media/upload]]))
(declare ^:private import-binfile)
(sv/defmethod ::import-binfile
"Import a penpot file in a binary format."
{::doc/added "1.15"
::webhooks/event? true}
::webhooks/event? true
::sse/stream? true
::sm/params schema:import-binfile}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id project-id file] :as params}]
(db/with-atomic [conn pool]
(projects/check-read-permissions! conn profile-id project-id)
(let [ids (import! (assoc cfg
::input (:path file)
::project-id project-id
::profile-id profile-id
::ignore-index-errors? true))]
(projects/check-read-permissions! pool profile-id project-id)
(let [params (-> cfg
(assoc ::input (:path file))
(assoc ::project-id project-id)
(assoc ::profile-id profile-id)
(assoc ::ignore-index-errors? true))]
(with-meta
(sse/response #(import-binfile params))
{::audit/props {:file nil}})))
(db/update! conn :project
{:modified-at (dt/now)}
{:id project-id})
(rph/with-meta ids
{::audit/props {:file nil :file-ids ids}}))))
(defn- import-binfile
[{:keys [::wrk/executor ::project-id] :as params}]
(db/tx-run! params
(fn [{:keys [::db/conn] :as params}]
;; NOTE: the importation process performs some operations that
;; are not very friendly with virtual threads, and for avoid
;; unexpected blocking of other concurrent operations we
;; dispatch that operation to a dedicated executor.
(let [result (p/thread-call executor (partial import! params))]
(db/update! conn :project
{:modified-at (dt/now)}
{:id project-id})
(deref result)))))

65
backend/src/app/rpc/commands/management.clj
View file

@ -14,6 +14,7 @@
[app.common.schema :as sm]
[app.common.uuid :as uuid]
[app.db :as db]
[app.http.sse :as sse]
[app.loggers.webhooks :as-alias webhooks]
[app.rpc :as-alias rpc]
[app.rpc.commands.binfile :as binfile]
@ -27,7 +28,9 @@
[app.util.pointer-map :as pmap]
[app.util.services :as sv]
[app.util.time :as dt]
[app.worker :as-alias wrk]
[clojure.walk :as walk]
[promesa.core :as p]
[promesa.exec :as px]))
;; --- COMMAND: Duplicate File
@ -405,29 +408,6 @@
;; --- COMMAND: Clone Template
(defn- clone-template!
[{:keys [::db/conn] :as cfg} {:keys [profile-id template-id project-id]}]
(let [template (tmpl/get-template-stream cfg template-id)
project (db/get-by-id conn :project project-id {:columns [:id :team-id]})]
(when-not template
(ex/raise :type :not-found
:code :template-not-found
:hint "template not found"))
(teams/check-edition-permissions! conn profile-id (:team-id project))
(-> cfg
;; FIXME: maybe reuse the conn instead of creating more
;; connections in the import process?
(dissoc ::db/conn)
(assoc ::binfile/input template)
(assoc ::binfile/project-id (:id project))
(assoc ::binfile/profile-id profile-id)
(assoc ::binfile/ignore-index-errors? true)
(assoc ::binfile/migrate? true)
(binfile/import!))))
(def ^:private
schema:clone-template
(sm/define
@ -435,15 +415,46 @@
[:project-id ::sm/uuid]
[:template-id ::sm/word-string]]))
(declare ^:private clone-template)
(sv/defmethod ::clone-template
"Clone into the specified project the template by its id."
{::doc/added "1.16"
::sse/stream? true
::webhooks/event? true
::sm/params schema:clone-template}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id] :as params}]
(db/with-atomic [conn pool]
(-> (assoc cfg ::db/conn conn)
(clone-template! (assoc params :profile-id profile-id)))))
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id project-id template-id] :as params}]
(let [project (db/get-by-id pool :project project-id {:columns [:id :team-id]})
_ (teams/check-edition-permissions! pool profile-id (:team-id project))
template (tmpl/get-template-stream cfg template-id)
params (-> cfg
(assoc ::binfile/input template)
(assoc ::binfile/project-id (:id project))
(assoc ::binfile/profile-id profile-id)
(assoc ::binfile/ignore-index-errors? true)
(assoc ::binfile/migrate? true))]
(when-not template
(ex/raise :type :not-found
:code :template-not-found
:hint "template not found"))
(sse/response #(clone-template params))))
(defn- clone-template
[{:keys [::wrk/executor ::binfile/project-id] :as params}]
(db/tx-run! params
(fn [{:keys [::db/conn] :as params}]
;; NOTE: the importation process performs some operations that
;; are not very friendly with virtual threads, and for avoid
;; unexpected blocking of other concurrent operations we
;; dispatch that operation to a dedicated executor.
(let [result (p/thread-call executor (partial binfile/import! params))]
(db/update! conn :project
{:modified-at (dt/now)}
{:id project-id})
(deref result)))))
;; --- COMMAND: Get list of builtin templates

2
backend/src/app/rpc/doc.clj
View file

@ -16,6 +16,7 @@
[app.common.schema.openapi :as oapi]
[app.common.schema.registry :as sr]
[app.config :as cf]
[app.http.sse :as-alias sse]
[app.loggers.webhooks :as-alias webhooks]
[app.rpc :as-alias rpc]
[app.util.json :as json]
@ -55,6 +56,7 @@
:module (or (some-> (::module mdata) d/name)
(-> (:ns mdata) (str/split ".") last))
:auth (::rpc/auth mdata true)
:sse (::sse/stream? mdata false)
:webhook (::webhooks/event? mdata false)
:docs (::sv/docstring mdata)
:deprecated (::deprecated mdata)

32
backend/test/backend_tests/helpers.clj
View file

@ -15,12 +15,13 @@
[app.common.pprint :as pp]
[app.common.schema :as sm]
[app.common.spec :as us]
[app.common.transit :as tr]
[app.common.uuid :as uuid]
[app.config :as cf]
[app.db :as db]
[app.main :as main]
[app.media]
[app.media :as-alias mtx]
[app.media]
[app.migrations]
[app.msgbus :as-alias mbus]
[app.rpc :as-alias rpc]
@ -43,8 +44,12 @@
[integrant.core :as ig]
[mockery.core :as mk]
[promesa.core :as p]
[promesa.exec :as px]
[ring.response :as rres]
[yetti.request :as yrq])
(:import
java.io.PipedInputStream
java.io.PipedOutputStream
java.util.UUID
org.postgresql.ds.PGSimpleDataSource))
@ -553,3 +558,28 @@
(assoc :return-list [])
(assoc :call-args nil)
(assoc :call-args-list [])))))
(defn- slurp'
[input & opts]
(let [sw (java.io.StringWriter.)]
(with-open [^java.io.Reader r (java.io.InputStreamReader. input "UTF-8")]
(io/copy r sw)
(.toString sw))))
(defn consume-sse
[callback]
(let [{:keys [::rres/status ::rres/body ::rres/headers] :as response} (callback {})
output (PipedOutputStream.)
input (PipedInputStream. output)]
(try
(px/exec! :virtual #(rres/-write-body-to-stream body nil output))
(into []
(map (fn [event]
(let [[item1 item2] (re-seq #"(.*): (.*)\n?" event)]
[(keyword (nth item1 2))
(tr/decode-str (nth item2 2))])))
(-> (slurp' input)
(str/split "\n\n")))
(finally
(.close input)))))

9
backend/test/backend_tests/rpc_management_test.clj
View file

@ -6,6 +6,7 @@
(ns backend-tests.rpc-management-test
(:require
[app.common.pprint :as pp]
[app.common.uuid :as uuid]
[app.db :as db]
[app.http :as http]
@ -604,9 +605,11 @@
(t/is (nil? (:error out)))
(let [result (:result out)]
(t/is (set? result))
(t/is (uuid? (first result)))
(t/is (= 1 (count result))))))
(t/is (fn? result))
(let [events (th/consume-sse result)]
(t/is (= 8 (count events)))
(t/is (= :end (first (last events))))))))
(t/deftest get-list-of-buitin-templates
(let [prof (th/create-profile* 1 {:is-active true})

2
common/deps.edn
View file

@ -32,7 +32,7 @@
funcool/tubax {:mvn/version "2021.05.20-0"}
funcool/cuerdas {:mvn/version "2022.06.16-403"}
funcool/promesa {:git/sha "658c429c56c11c33da7594fa2ef53f4e6afedac4"
funcool/promesa {:git/sha "484b7f5c0d08d817746caa685ed9ac5583eb37fa"
:git/url "https://github.com/funcool/promesa"}
funcool/datoteka {:mvn/version "3.0.66"

2
docker/devenv/files/nginx.conf
View file

@ -105,6 +105,8 @@ http {
location /api {
proxy_pass http://127.0.0.1:6060/api;
proxy_buffering off;
proxy_http_version 1.1;
}
location /admin {

1
frontend/package.json
View file

@ -78,6 +78,7 @@
"dependencies": {
"date-fns": "^2.30.0",
"draft-js": "^0.11.7",
"eventsource-parser": "^1.1.1",
"highlight.js": "^11.8.0",
"js-beautify": "^1.14.9",
"jszip": "^3.10.1",

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

@ -10,6 +10,7 @@
[app.common.data.macros :as dm]
[app.common.features :as cfeat]
[app.common.files.helpers :as cfh]
[app.common.logging :as log]
[app.common.schema :as sm]
[app.common.uri :as u]
[app.common.uuid :as uuid]
@ -25,6 +26,7 @@
[app.util.dom :as dom]
[app.util.i18n :as i18n :refer [tr]]
[app.util.router :as rt]
[app.util.sse :as sse]
[app.util.time :as dt]
[app.util.timers :as tm]
[app.util.webapi :as wapi]
@ -32,6 +34,8 @@
[clojure.set :as set]
[potok.core :as ptk]))
(log/set-level! :warn)
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Initialization
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@ -949,7 +953,17 @@
(let [{:keys [on-success on-error]
:or {on-success identity
on-error rx/throw}} (meta params)]
(->> (rp/cmd! :clone-template {:project-id project-id :template-id template-id})
(->> (rp/cmd! ::sse/clone-template {:project-id project-id
:template-id template-id})
(rx/tap (fn [event]
(let [payload (sse/get-payload event)
type (sse/get-type event)]
(if (= type "event")
(log/dbg :hint "clone-template: progress" :section (:section payload) :name (:name payload))
(log/dbg :hint "clone-template: end")))))
(rx/filter sse/end-of-stream?)
(rx/map sse/get-payload)
(rx/tap on-success)
(rx/catch on-error))))))

32
frontend/src/app/main/repo.cljs
View file

@ -7,9 +7,11 @@
(ns app.main.repo
(:require
[app.common.data :as d]
[app.common.transit :as t]
[app.common.uri :as u]
[app.config :as cf]
[app.util.http :as http]
[app.util.sse :as sse]
[beicon.core :as rx]
[cuerdas.core :as str]))
@ -56,8 +58,14 @@
{:query-params [:file-id :revn]
:form-data? true}
::sse/clone-template
{:response-type ::sse/stream}
::sse/import-binfile
{:response-type ::sse/stream
:form-data? true}
:export-binfile {:response-type :blob}
:import-binfile {:form-data? true}
:retrieve-list-of-builtin-templates {:query-params :all}
})
@ -85,9 +93,9 @@
:else :post)
request {:method method
:uri (u/join cf/public-uri "api/rpc/command/" (name id))
:uri (u/join cf/public-uri "api/rpc/command/" nid)
:credentials "include"
:headers {"accept" "application/transit+json"}
:headers {"accept" "application/transit+json,text/event-stream,*/*"}
:body (when (= method :post)
(if form-data?
(http/form-data params)
@ -97,11 +105,21 @@
(if query-params
(select-keys params query-params)
nil))
:response-type (or response-type :text)}]
(->> (http/send! request)
(rx/map decode-fn)
(rx/mapcat handle-response))))
:response-type
(if (= response-type ::sse/stream)
:stream
(or response-type :text))}
result (->> (http/send! request)
(rx/map decode-fn)
(rx/mapcat handle-response))]
(cond->> result
(= ::sse/stream response-type)
(rx/mapcat (fn [body]
(-> (sse/create-stream body)
(sse/read-stream t/decode-str)))))))
(defmulti cmd! (fn [id _] id))

6
frontend/src/app/main/ui/dashboard/projects.cljs
View file

@ -147,8 +147,10 @@
(mf/use-fn
(mf/deps template default-project-id)
(fn []
(let [mdata {:on-success on-template-cloned-success :on-error on-template-cloned-error}
params {:project-id default-project-id :template-id (:id template)}]
(let [mdata {:on-success on-template-cloned-success
:on-error on-template-cloned-error}
params {:project-id default-project-id
:template-id (:id template)}]
(swap! state #(assoc % :status :importing))
(st/emit! (with-meta (dd/clone-template (with-meta params mdata))
{::ev/origin "get-started-hero-block"})))))]

27
frontend/src/app/util/http.cljs
View file

@ -105,17 +105,22 @@
(defn send!
[{:keys [response-type] :or {response-type :text} :as params}]
(letfn [(on-response [response]
(let [body (case response-type
:json (.json ^js response)
:text (.text ^js response)
:blob (.blob ^js response))]
(->> (rx/from body)
(rx/map (fn [body]
{::response response
:status (.-status ^js response)
:headers (parse-headers (.-headers ^js response))
:body body})))))]
(letfn [(on-response [^js response]
(if (= :stream response-type)
(rx/of {:status (.-status response)
:headers (parse-headers (.-headers response))
:body (.-body response)
::response response})
(let [body (case response-type
:json (.json ^js response)
:text (.text ^js response)
:blob (.blob ^js response))]
(->> (rx/from body)
(rx/map (fn [body]
{::response response
:status (.-status ^js response)
:headers (parse-headers (.-headers ^js response))
:body body}))))))]
(->> (fetch params)
(rx/mapcat on-response))))

54
frontend/src/app/util/sse.cljs Normal file
View file

@ -0,0 +1,54 @@
;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS INC
(ns app.util.sse
(:require
["eventsource-parser/stream" :as sse]
[beicon.core :as rx]))
(defn create-stream
[^js/ReadableStream stream]
(.. stream
(pipeThrough (js/TextDecoderStream.))
(pipeThrough (sse/EventSourceParserStream.))))
(defn read-stream
[^js/ReadableStream stream decode-fn]
(letfn [(read-items [^js reader]
(->> (rx/from (.read reader))
(rx/mapcat (fn [result]
(if (.-done result)
(rx/empty)
(rx/concat
(rx/of (.-value result))
(read-items reader)))))))]
(->> (read-items (.getReader stream))
(rx/mapcat (fn [^js event]
(let [type (.-event event)
data (.-data event)
data (decode-fn data)]
(if (= "error" type)
(rx/throw (ex-info "stream exception" data))
(rx/of #js {:type type :data data}))))))))
(defn get-type
[event]
(unchecked-get event "type"))
(defn get-payload
[event]
(unchecked-get event "data"))
(defn end-of-stream?
[event]
(= "end" (get-type event)))
(defn event?
[event]
(= "event" (get-type event)))

20
frontend/src/app/worker/import.cljs
View file

@ -22,6 +22,7 @@
[app.util.i18n :as i18n :refer [tr]]
[app.util.import.parser :as cip]
[app.util.json :as json]
[app.util.sse :as sse]
[app.util.webapi :as wapi]
[app.util.zip :as uz]
[app.worker.impl :as impl]
@ -329,7 +330,7 @@
(map #(assoc % :type :fill)))
stroke-images-data (->> (cip/get-stroke-images-data node)
(map #(assoc % :type :stroke)))
images-data (concat
fill-images-data
stroke-images-data
@ -709,15 +710,22 @@
:response-type :blob
:method :get})
(rx/map :body)
(rx/mapcat #(rp/cmd! :import-binfile {:file % :project-id project-id}))
(rx/map (fn [_]
{:status :import-finish
:file-id (:file-id data)}))
(rx/mapcat (fn [file]
(->> (rp/cmd! ::sse/import-binfile {:file file :project-id project-id})
(rx/tap (fn [event]
(let [payload (sse/get-payload event)
type (sse/get-type event)]
(if (= type "event")
(log/dbg :hint "import-binfile: progress" :section (:section payload) :name (:name payload))
(log/dbg :hint "import-binfile: end")))))
(rx/filter sse/end-of-stream?)
(rx/map (fn [_]
{:status :import-finish
:file-id (:file-id data)})))))
(rx/catch (fn [cause]
(log/error :hint "unexpected error on import process"
:project-id project-id
::log/sync? true)
;; TODO: consider do thi son logging directly ?
(when (map? cause)
(println "Error data:")

8
frontend/yarn.lock
View file

@ -7028,6 +7028,13 @@ __metadata:
languageName: node
linkType: hard
"eventsource-parser@npm:^1.1.1":
version: 1.1.1
resolution: "eventsource-parser@npm:1.1.1"
checksum: 6eae5e8300dc5d4dcd29d09d037a43b9954077fbcc936904065d1abf5ec503f1fa56a942ad4f796b7dcc3bf64224440917ea635ef4a75f5522e9951df4b9aadf
languageName: node
linkType: hard
"evp_bytestokey@npm:^1.0.0, evp_bytestokey@npm:^1.0.3":
version: 1.0.3
resolution: "evp_bytestokey@npm:1.0.3"
@ -7578,6 +7585,7 @@ __metadata:
concurrently: "npm:^8.2.2"
date-fns: "npm:^2.30.0"
draft-js: "npm:^0.11.7"
eventsource-parser: "npm:^1.1.1"
gettext-parser: "npm:^7.0.1"
gulp: "npm:4.0.2"
gulp-cached: "npm:^1.1.1"