diff --git a/backend/src/app/http/content_negotiation.clj b/backend/src/app/http/content_negotiation.clj new file mode 100644 index 0000000000..6bf76a7824 --- /dev/null +++ b/backend/src/app/http/content_negotiation.clj @@ -0,0 +1,58 @@ +;; 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 SUBSIDIARY SL + +(ns app.http.content-negotiation + "Content format negotiation helpers shared between the regular + response formatting middleware and the SSE streaming endpoints. + + The negotiated format affects only the encoding of the response + payload; the transport (regular http body or `text/event-stream`) + is not affected." + (:require + [app.common.json :as json] + [app.util.pointer-map :as pmap] + [cuerdas.core :as str] + [yetti.request :as yreq])) + +(defn- format-from-params + [{:keys [query-params]}] + (and (= "json" (get query-params :_fmt)) + :json)) + +(defn negotiate-format + "Determine the response payload format for the request. Returns + `:json` or `:transit`. + + The `:_fmt=json` query parameter takes precedence over the + `Accept` header; when no explicit signal is present, transit is + the default." + [request] + (or (format-from-params request) + (let [accept (yreq/get-header request "accept")] + (cond + (or (= accept "application/transit+json") + (str/includes? accept "application/transit+json")) + :transit + + (or (= accept "application/json") + (str/includes? accept "application/json")) + :json + + :else + :transit)))) + +(defn write-json-value + [_ val] + (if (pmap/pointer-map? val) + [(pmap/get-id val) (meta val)] + val)) + +(defn json-encode-str + "Encode value as a JSON string using the same conventions as the + regular JSON API responses: camelCase keys, keywords/UUIDs and + instants serialized as plain JSON scalars." + [v] + (json/encode v :key-fn json/write-camel-key :value-fn write-json-value)) diff --git a/backend/src/app/http/middleware.clj b/backend/src/app/http/middleware.clj index 22c1cc29d9..110c81349c 100644 --- a/backend/src/app/http/middleware.clj +++ b/backend/src/app/http/middleware.clj @@ -13,9 +13,9 @@ [app.common.transit :as t] [app.config :as cf] [app.http :as-alias http] + [app.http.content-negotiation :as cnegot] [app.http.errors :as errors] [app.tokens :as tokens] - [app.util.pointer-map :as pmap] [cuerdas.core :as str] [yetti.adapter :as yt] [yetti.middleware :as ymw] @@ -126,12 +126,6 @@ (def ^:const buffer-size (:xnio/buffer-size yt/defaults)) -(defn- write-json-value - [_ val] - (if (pmap/pointer-map? val) - [(pmap/get-id val) (meta val)] - val)) - (defn wrap-format-response [handler] (letfn [(transit-streamable-body [data opts _ output-stream] @@ -153,7 +147,7 @@ data (encode data)] (with-open [^OutputStream bos (buffered-output-stream output-stream buffer-size)] (with-open [^java.io.OutputStreamWriter writer (java.io.OutputStreamWriter. bos)] - (json/write writer data :key-fn json/write-camel-key :value-fn write-json-value)))) + (json/write writer data :key-fn json/write-camel-key :value-fn cnegot/write-json-value)))) (catch java.io.IOException _) (catch Throwable cause (binding [l/*context* {:value data}] @@ -183,24 +177,10 @@ (assoc ::yres/body (yres/stream-body (partial transit-streamable-body body opts))))) response))) - (format-from-params [{:keys [query-params] :as request}] - (and (= "json" (get query-params :_fmt)) - "application/json")) - (format-response [response request] - (let [accept (or (format-from-params request) - (yreq/get-header request "accept"))] - (cond - (or (= accept "application/transit+json") - (str/includes? accept "application/transit+json")) - (format-response-with-transit response request) - - (or (= accept "application/json") - (str/includes? accept "application/json")) - (format-response-with-json response request) - - :else - (format-response-with-transit response request)))) + (case (cnegot/negotiate-format request) + :transit (format-response-with-transit response request) + :json (format-response-with-json response request))) (process-response [response request] (cond-> response diff --git a/backend/src/app/http/sse.clj b/backend/src/app/http/sse.clj index 0fc7c385bd..4d70bd8bf9 100644 --- a/backend/src/app/http/sse.clj +++ b/backend/src/app/http/sse.clj @@ -10,6 +10,7 @@ [app.common.data :as d] [app.common.logging :as l] [app.common.transit :as t] + [app.http.content-negotiation :as cnegot] [app.http.errors :as errors] [app.util.events :as events] [promesa.exec :as px] @@ -26,11 +27,11 @@ (.flush output)) (defn- encode - [[name data]] + [encode-fn [name data]] (try (let [data (with-out-str (println "event:" (d/name name)) - (println "data:" (t/encode-str data {:type :json-verbose})) + (println "data:" (encode-fn data)) (println))] (.getBytes ^String data "UTF-8")) (catch Throwable cause @@ -38,6 +39,12 @@ :cause cause) nil))) +(defn- resolve-encoder + [format] + (case format + :transit #(t/encode-str % {:type :json-verbose}) + :json cnegot/json-encode-str)) + ;; ---- PUBLIC API (def default-headers @@ -47,27 +54,32 @@ "X-Accel-Buffering" "no"}) (defn response + "Create a streaming SSE response. The payload of each event is + encoded in transit or plain JSON depending on the request content + negotiation (see `app.http.content-negotiation/negotiate-format`), + transit being the default." [handler & {:keys [buf] :or {buf 32} :as opts}] (fn [request] - {::yres/headers default-headers - ::yres/status 200 - ::yres/body (yres/stream-body - (fn [_ output] + (let [encode-fn (resolve-encoder (cnegot/negotiate-format request))] + {::yres/headers default-headers + ::yres/status 200 + ::yres/body (yres/stream-body + (fn [_ output] - (let [channel (sp/chan :buf buf :xf (keep encode)) - listener (events/spawn-listener - channel - (partial write! output) - (partial pu/close! output))] - (try - (binding [events/*channel* channel] - (let [result (handler)] - (events/tap :end result))) + (let [channel (sp/chan :buf buf :xf (keep (partial encode encode-fn))) + listener (events/spawn-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 channel :error result))) + (catch Throwable cause + (let [result (errors/handle' cause request)] + (events/tap channel :error result))) - (finally - (sp/close! channel) - (px/await! listener))))))})) + (finally + (sp/close! channel) + (px/await! listener))))))}))) diff --git a/backend/test/backend_tests/http_sse_test.clj b/backend/test/backend_tests/http_sse_test.clj new file mode 100644 index 0000000000..3215a9fcf8 --- /dev/null +++ b/backend/test/backend_tests/http_sse_test.clj @@ -0,0 +1,108 @@ +;; 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 SUBSIDIARY SL + +(ns backend-tests.http-sse-test + (:require + [app.common.transit :as tr] + [app.http.sse :as sse] + [app.util.events :as events] + [clojure.string :as str] + [clojure.test :as t] + [yetti.response :as yres]) + (:import + java.io.ByteArrayOutputStream)) + +(def ^:private test-file-id #uuid "00000000-0000-0000-0000-000000000001") + +(def ^:private progress-json + (str "{\"fileId\":\"" test-file-id "\",\"index\":1,\"total\":2}")) + +(defn- make-handler + [] + (fn [] + (events/tap :progress {:file-id test-file-id :index 1 :total 2}) + {:status "ok"})) + +(defn- run-sse + "Runs the sse response with the provided fake request and returns + the raw stream content as a string." + [request] + (let [response ((sse/response (make-handler)) request) + output (ByteArrayOutputStream.)] + (yres/write-body-to-stream (::yres/body response) response output) + (.toString output "UTF-8"))) + +(defn- data-events + [stream] + (->> (str/split stream #"\n\n") + (keep (fn [block] + (let [[_ event] (re-find #"event: (.+)" block) + [_ data] (re-find #"data: (.+)" block)] + (when (and event data) + {:event (str/trim event) :data (str/trim data)})))) + (vec))) + +(t/deftest default-format-is-transit + (let [[progress end] (data-events (run-sse {}))] + (t/is (= "progress" (:event progress))) + (t/is (= {:file-id test-file-id :index 1 :total 2} + (tr/decode-str (:data progress)))) + (t/is (= "end" (:event end))) + (t/is (= {:status "ok"} (tr/decode-str (:data end)))))) + +(t/deftest accept-transit-header-uses-transit + (let [[progress] (data-events (run-sse {:headers {"accept" "application/transit+json"}}))] + (t/is (= {:file-id test-file-id :index 1 :total 2} + (tr/decode-str (:data progress)))))) + +(t/deftest accept-frontend-combo-header-uses-transit + (let [[progress] (data-events (run-sse {:headers {"accept" "application/transit+json,text/event-stream,*/*"}}))] + (t/is (= {:file-id test-file-id :index 1 :total 2} + (tr/decode-str (:data progress)))))) + +(t/deftest accept-json-header-uses-json + (let [stream (run-sse {:headers {"accept" "application/json"}}) + [progress end] (data-events stream)] + (t/is (= "progress" (:event progress))) + (t/is (= progress-json (:data progress))) + (t/is (= "end" (:event end))) + (t/is (= "{\"status\":\"ok\"}" (:data end))))) + +(t/deftest fmt-json-query-param-uses-json + (let [[progress] (data-events (run-sse {:query-params {:_fmt "json"}}))] + (t/is (= progress-json (:data progress))))) + +(t/deftest fmt-json-query-param-precedence-over-accept + (let [[progress] (data-events (run-sse {:query-params {:_fmt "json"} + :headers {"accept" "application/transit+json"}}))] + (t/is (= progress-json (:data progress))))) + +(t/deftest wildcard-accept-defaults-to-transit + (let [[progress] (data-events (run-sse {:headers {"accept" "*/*"}}))] + (t/is (= {:file-id test-file-id :index 1 :total 2} + (tr/decode-str (:data progress)))))) + +(t/deftest error-events-are-negotiated + (let [response ((sse/response (fn [] + (throw (ex-info "boom" + {:type :validation + :code :generic + :hint "boom"})))) + {:headers {"accept" "application/json"}}) + output (ByteArrayOutputStream.)] + (yres/write-body-to-stream (::yres/body response) response output) + (let [[error] (data-events (.toString output "UTF-8"))] + (t/is (= "error" (:event error))) + (t/is (= "{\"type\":\"validation\",\"code\":\"generic\",\"hint\":\"boom\"}" + (:data error)))))) + +(t/deftest content-type-is-event-stream-on-both-formats + (let [check (fn [request] + (let [response ((sse/response (make-handler)) request)] + (get (::yres/headers response) "Content-Type")))] + (t/is (= "text/event-stream;charset=UTF-8" (check {}))) + (t/is (= "text/event-stream;charset=UTF-8" + (check {:headers {"accept" "application/json"}}))))) diff --git a/frontend/src/app/main/repo.cljs b/frontend/src/app/main/repo.cljs index 722c9a882a..cb9516c471 100644 --- a/frontend/src/app/main/repo.cljs +++ b/frontend/src/app/main/repo.cljs @@ -8,6 +8,7 @@ (:require [app.common.data :as d] [app.common.exceptions :as ex] + [app.common.json :as json] [app.common.logging :as log] [app.common.time :as ct] [app.common.transit :as t] @@ -158,6 +159,7 @@ [id params options] (let [{:keys [response-type stream? + response-format form-data? raw-transit? query-params @@ -180,11 +182,16 @@ response-type (d/nilv response-type :text) + accept + (if (and stream? (= response-format :json)) + "application/json,text/event-stream,*/*" + "application/transit+json,text/event-stream,*/*") + request {:method method :uri (u/join cf/public-uri "api/main/methods/" nid) :credentials "include" - :headers {"accept" "application/transit+json,text/event-stream,*/*" + :headers {"accept" accept "x-external-session-id" (cf/external-session-id) "x-session-id" (str cf/session-id) "x-event-origin" (::ev/origin (meta params))} @@ -225,7 +232,9 @@ (if response-stream? (-> (sse/create-stream body) - (sse/read-stream t/decode-str)) + (sse/read-stream (if (= response-format :json) + json/decode + t/decode-str))) (->> response (http/process-response-type response-type)