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 ad80bfe9c4..052e8fa92e 100644 --- a/backend/src/app/http/middleware.clj +++ b/backend/src/app/http/middleware.clj @@ -13,11 +13,11 @@ [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.metrics :as mtx] [app.tokens :as tokens] [app.util.inet :as inet] - [app.util.pointer-map :as pmap] [cuerdas.core :as str] [yetti.adapter :as yt] [yetti.middleware :as ymw] @@ -186,12 +186,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] @@ -213,7 +207,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}] @@ -243,23 +237,11 @@ (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 + (let [format (cnegot/negotiate-format request)] + (case format + :transit (format-response-with-transit response request) + :json (format-response-with-json response request) (format-response-with-transit response request)))) (process-response [response request] diff --git a/backend/src/app/http/sse.clj b/backend/src/app/http/sse.clj index 4c1420876e..296c753d9a 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,13 @@ :cause cause) nil))) +(defn- resolve-encoder + [format] + (case format + :transit #(t/encode-str % {:type :json-verbose}) + :json cnegot/json-encode-str + #(t/encode-str % {:type :json-verbose}))) + ;; ---- PUBLIC API (def default-headers @@ -47,26 +55,30 @@ "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 [channel (sp/chan :buf buf :xf (keep encode)) - listener (events/spawn-listener - channel - (partial write! output) - (partial pu/close! output))] - (try - (binding [events/*sink* channel] - (try - (let [result (handler)] - (events/tap :end result)) - (catch Throwable cause - (let [result (errors/handle' cause request)] - (events/tap :error result))))) - (finally - (sp/close! channel) - (px/await! listener))))))})) + (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 (partial encode encode-fn))) + listener (events/spawn-listener + channel + (partial write! output) + (partial pu/close! output))] + (try + (binding [events/*sink* channel] + (try + (let [result (handler)] + (events/tap :end result)) + (catch Throwable cause + (let [result (errors/handle' cause request)] + (events/tap :error result))))) + (finally + (sp/close! channel) + (px/await! listener))))))}))) diff --git a/backend/src/app/http/websocket.clj b/backend/src/app/http/websocket.clj index 50bd64f115..8724ad4778 100644 --- a/backend/src/app/http/websocket.clj +++ b/backend/src/app/http/websocket.clj @@ -10,12 +10,15 @@ [app.binfile.common :as bfc] [app.common.data.macros :as dm] [app.common.exceptions :as ex] + [app.common.json :as json] [app.common.logging :as l] [app.common.pprint :as pp] [app.common.schema :as sm] [app.common.time :as ct] + [app.common.transit :as t] [app.common.uuid :as uuid] [app.db :as db] + [app.http.content-negotiation :as cnegot] [app.http.session :as session] [app.metrics :as mtx] [app.msgbus :as mbus] @@ -88,6 +91,39 @@ ;; WEBSOCKET HANDLER ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +(defn resolve-encoder + "Resolve the outbound message encoder for the negotiated format. + The transit encoding is the default for unexpected format values." + [format] + (case format + :transit #(t/encode-str % {:type :json-verbose}) + :json cnegot/json-encode-str + #(t/encode-str % {:type :json-verbose}))) + +(defn resolve-decoder + "Resolve the inbound message decoder for the negotiated format. + The transit decoding is the default for unexpected format values." + [format] + (case format + :transit t/decode-str + :json #(json/decode % {:key-fn json/read-kebab-key}) + t/decode-str)) + +(defn normalize-message + "Normalize the inbound message into the same shape produced by the + transit decoding: the `:type` value is a keyword and the entity ids + are UUID objects. This is a no-op for transit decoded messages." + [message] + (cond-> message + (string? (:type message)) + (assoc :type (keyword (:type message))) + + (string? (:team-id message)) + (assoc :team-id (uuid/parse* (:team-id message))) + + (string? (:file-id message)) + (assoc :file-id (uuid/parse* (:file-id message))))) + (defmulti handle-message (fn [_ _ message] (:type message))) @@ -293,7 +329,8 @@ :id :websocket-messages-total :labels recv-labels :inc 1) - (assoc message :profile-id profile-id :session-id session-id)) + (let [message (normalize-message message)] + (assoc message :profile-id profile-id :session-id session-id))) (defn- on-snd-message [{:keys [::mtx/metrics]} message] @@ -303,9 +340,13 @@ :inc 1) message) -(defn- http-handler +(defn http-handler + "The websocket upgrade request handler. Returns the upgrade + response containing the websocket listener built with the + codecs of the negotiated payload format." [cfg {:keys [params ::session/profile-id] :as request}] - (let [session-id (some-> params :session-id uuid/parse*)] + (let [session-id (some-> params :session-id uuid/parse*) + format (cnegot/negotiate-format request)] (when-not (uuid? session-id) (ex/raise :type :validation :code :missing-session-id @@ -327,12 +368,17 @@ :else (do - (l/trace :hint "websocket request" :profile-id profile-id :session-id session-id) + (l/trace :hint "websocket request" + :profile-id profile-id + :session-id session-id + :format format) {::yws/listener (ws/listener request ::ws/on-rcv-message (partial on-rcv-message cfg) ::ws/on-snd-message (partial on-snd-message cfg) ::ws/on-connect (partial on-connect cfg) ::ws/handler (partial handle-message cfg) + ::ws/encode-fn (resolve-encoder format) + ::ws/decode-fn (resolve-decoder format) ::profile-id profile-id ::session-id session-id)})))) diff --git a/backend/src/app/util/websocket.clj b/backend/src/app/util/websocket.clj index 514de1178c..ee5af1329c 100644 --- a/backend/src/app/util/websocket.clj +++ b/backend/src/app/util/websocket.clj @@ -26,6 +26,9 @@ (def max-missed-heartbeats 3) (def heartbeat-interval 5000) +(def default-encode-fn #(t/encode-str % {:type :json-verbose})) +(def default-decode-fn t/decode-str) + (defn- encode-beat [n] (doto (ByteBuffer/allocate 8) @@ -60,10 +63,16 @@ It also accepts some options that allows you parametrize the protocol behavior. The options map will be used as-as for the - initial data of the `ws` data structure" + initial data of the `ws` data structure. + + The returned map has the resolved options attached in its + metadata under the `::options` key; it is intended for testing + and introspection purposes only." [request & {:keys [::on-rcv-message ::on-snd-message ::on-connect + ::encode-fn + ::decode-fn ::input-buff-size ::output-buff-size ::idle-timeout] @@ -72,7 +81,9 @@ idle-timeout 60000 on-connect identity on-snd-message identity-3 - on-rcv-message identity-3} + on-rcv-message identity-3 + encode-fn default-encode-fn + decode-fn default-decode-fn} :as options}] (assert (fn? on-rcv-message) "'on-rcv-message' should be a function") @@ -99,40 +110,44 @@ (assoc ::output-ch output-ch) (assoc ::close-ch close-ch) (assoc ::remote-addr ip-addr) - (assoc ::user-agent uagent))] + (assoc ::user-agent uagent) + (assoc ::encode-fn encode-fn) + (assoc ::decode-fn decode-fn))] - {:on-open - (fn on-open [channel] - (l/dbg :fn "on-open" :conn-id (str id)) - (let [options (-> options - (assoc ::channel channel) - (on-connect)) - timeout (ct/duration idle-timeout)] + (with-meta + {:on-open + (fn on-open [channel] + (l/dbg :fn "on-open" :conn-id (str id)) + (let [options (-> options + (assoc ::channel channel) + (on-connect)) + timeout (ct/duration idle-timeout)] - (yws/set-idle-timeout! channel timeout) - (px/submit! :vthread (partial start-io-loop! options)))) + (yws/set-idle-timeout! channel timeout) + (px/submit! :vthread (partial start-io-loop! options)))) - :on-close - (fn on-close [_channel code reason] - (l/dbg :fn "on-close" - :conn-id (str id) - :code code - :reason reason) - (sp/close! close-ch)) + :on-close + (fn on-close [_channel code reason] + (l/dbg :fn "on-close" + :conn-id (str id) + :code code + :reason reason) + (sp/close! close-ch)) - :on-error - (fn on-error [_channel cause] - (sp/close! close-ch cause)) + :on-error + (fn on-error [_channel cause] + (sp/close! close-ch cause)) - :on-message - (fn on-message [_channel message] - (when (string? message) - (sp/offer! input-ch message) - (swap! state assoc ::last-activity-at (ct/now)))) + :on-message + (fn on-message [_channel message] + (when (string? message) + (sp/offer! input-ch message) + (swap! state assoc ::last-activity-at (ct/now)))) - :on-pong - (fn on-pong [_channel data] - (sp/put! hbeat-ch data))})) + :on-pong + (fn on-pong [_channel data] + (sp/put! hbeat-ch data))} + {::options options}))) (defn- handle-ping! [{:keys [::id ::beats ::channel] :as wsp} beat-id] @@ -143,7 +158,8 @@ (defn- start-io-loop! [{:keys [::id ::close-ch ::input-ch ::output-ch ::heartbeat-ch - ::channel ::handler ::beats ::on-rcv-message ::on-snd-message] + ::channel ::handler ::beats ::on-rcv-message ::on-snd-message + ::encode-fn ::decode-fn] :as wsp}] (try (handler wsp {:type :open}) @@ -169,7 +185,7 @@ (recur i)) (identical? p input-ch) - (let [message (t/decode-str msg) + (let [message (decode-fn msg) message (on-rcv-message message) {:keys [request-id] :as response} (handler wsp message)] (when (map? response) @@ -181,7 +197,7 @@ (identical? p output-ch) (let [message (on-snd-message msg) - message (t/encode-str message {:type :json-verbose})] + message (encode-fn message)] (yws/send channel message) (recur i)))))) diff --git a/backend/test/backend_tests/http_content_negotiation_test.clj b/backend/test/backend_tests/http_content_negotiation_test.clj new file mode 100644 index 0000000000..1cff5641f1 --- /dev/null +++ b/backend/test/backend_tests/http_content_negotiation_test.clj @@ -0,0 +1,40 @@ +;; 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-content-negotiation-test + (:require + [app.http.content-negotiation :as cnegot] + [clojure.test :as t])) + +(t/deftest negotiate-format-defaults-to-transit + (t/is (= :transit (cnegot/negotiate-format {}))) + (t/is (= :transit (cnegot/negotiate-format {:headers {"accept" "*/*"}}))) + (t/is (= :transit (cnegot/negotiate-format {:headers {"accept" "text/event-stream"}})))) + +(t/deftest negotiate-format-from-accept-header + (t/is (= :transit (cnegot/negotiate-format + {:headers {"accept" "application/transit+json"}}))) + (t/is (= :transit (cnegot/negotiate-format + {:headers {"accept" "application/transit+json,text/event-stream,*/*"}}))) + (t/is (= :json (cnegot/negotiate-format + {:headers {"accept" "application/json"}}))) + (t/is (= :json (cnegot/negotiate-format + {:headers {"accept" "application/json, text/event-stream"}})))) + +(t/deftest negotiate-format-from-query-param + (t/is (= :json (cnegot/negotiate-format {:query-params {:_fmt "json"}}))) + (t/is (= :json (cnegot/negotiate-format + {:query-params {:_fmt "json"} + :headers {"accept" "application/transit+json"}})))) + +(t/deftest negotiate-format-only-json-param-is-recognized + (t/is (= :json (cnegot/negotiate-format + {:query-params {:_fmt "transit"} + :headers {"accept" "application/json"}}))) + (t/is (= :transit (cnegot/negotiate-format {:query-params {:_fmt "transit"}})))) + +(t/deftest negotiate-format-without-accept-header + (t/is (= :transit (cnegot/negotiate-format {:headers {}})))) 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/backend/test/backend_tests/http_websocket_test.clj b/backend/test/backend_tests/http_websocket_test.clj index 6ffc10e744..5bc3b4b323 100644 --- a/backend/test/backend_tests/http_websocket_test.clj +++ b/backend/test/backend_tests/http_websocket_test.clj @@ -6,29 +6,33 @@ (ns backend-tests.http-websocket-test (:require + [app.common.json :as json] + [app.common.transit :as tr] [app.common.uuid :as uuid] [app.db :as db] - [app.http.websocket :as ws] + [app.http.session :as-alias session] + [app.http.websocket :as http.ws] [app.msgbus :as mbus] [app.nitrate :as nitrate] [app.rpc :as-alias rpc] [app.rpc.commands.files :as files] [app.rpc.commands.teams :as teams] - [app.util.websocket :as util-ws] + [app.util.websocket :as ws] [backend-tests.helpers :as th] [clojure.test :as t] - [promesa.exec.csp :as sp])) + [promesa.exec.csp :as sp] + [yetti.websocket :as yws])) (t/use-fixtures :once th/state-init) (t/use-fixtures :each th/database-reset) (defn make-wsp [profile-id state output-ch] - {::util-ws/id (uuid/next) - ::util-ws/state state - ::util-ws/output-ch output-ch - ::ws/profile-id profile-id - ::ws/session-id (uuid/next)}) + {::ws/id (uuid/next) + ::ws/state state + ::ws/output-ch output-ch + ::http.ws/profile-id profile-id + ::http.ws/session-id (uuid/next)}) (t/deftest subscribe-file-permission-check (let [profile1 (th/create-profile* 1 {:is-active true}) @@ -44,7 +48,7 @@ (t/is (thrown-with-msg? clojure.lang.ExceptionInfo #"not found" - ((get-method ws/handle-message :subscribe-file) + ((get-method http.ws/handle-message :subscribe-file) cfg wsp {:file-id (:id file)}))))) (t/testing "permission check passes for authorized user" @@ -63,7 +67,7 @@ (t/is (thrown-with-msg? clojure.lang.ExceptionInfo #"not found" - ((get-method ws/handle-message :subscribe-team) + ((get-method http.ws/handle-message :subscribe-team) cfg wsp {:team-id (:id team)}))))) (t/testing "permission check passes for authorized user" @@ -80,11 +84,11 @@ (with-redefs [nitrate/call nitrate-call mbus/sub! (fn [_ & {:keys [topics]}] (swap! calls conj topics))] - ((get-method ws/handle-message :subscribe-team) + ((get-method http.ws/handle-message :subscribe-team) th/*system* wsp {:team-id team-id})) - (some-> @state ::ws/team-subscription :channel sp/close!) + (some-> @state ::http.ws/team-subscription :channel sp/close!) {:topics @calls - :subscription (::ws/team-subscription @state)})) + :subscription (::http.ws/team-subscription @state)})) (t/deftest subscribe-team-subscribes-to-team-organization (let [profile (th/create-profile* 1 {:is-active true}) @@ -120,9 +124,9 @@ cfg th/*system* file-id (:id file) sub-ch (sp/chan :buf (sp/dropping-buffer 64)) - state (atom {::ws/file-subscription {:file-id file-id - :channel sub-ch - :topic file-id}}) + state (atom {::http.ws/file-subscription {:file-id file-id + :channel sub-ch + :topic file-id}}) output-ch (sp/chan :buf (sp/dropping-buffer 64)) wsp (make-wsp (:id profile) state output-ch)] @@ -132,7 +136,7 @@ :position {:x 10 :y 20} :zoom 1.0}] (t/is (nil? - ((get-method ws/handle-message :pointer-update) + ((get-method http.ws/handle-message :pointer-update) cfg wsp wrong-msg))))) (t/testing "does nothing when no file subscription exists" @@ -143,5 +147,127 @@ :position {:x 10 :y 20} :zoom 1.0}] (t/is (nil? - ((get-method ws/handle-message :pointer-update) + ((get-method http.ws/handle-message :pointer-update) cfg empty-wsp msg))))))) + +(t/deftest resolve-encoder-defaults-to-transit + (let [encode (http.ws/resolve-encoder :transit) + msg {:type :join-file + :file-id (uuid/next) + :profile-id (uuid/next)}] + (t/is (= msg (tr/decode-str (encode msg)))))) + +(t/deftest resolve-encoder-unexpected-format-falls-back-to-transit + (let [encode (http.ws/resolve-encoder :foo) + msg {:type :join-file :file-id (uuid/next)}] + (t/is (= msg (tr/decode-str (encode msg)))))) + +(t/deftest resolve-encoder-json + (let [encode (http.ws/resolve-encoder :json) + file-id (uuid/next) + encoded (encode {:type :join-file :file-id file-id})] + (t/is (string? encoded)) + (t/is (= "join-file" (get (json/decode encoded) "type"))) + (t/is (= (str file-id) (get (json/decode encoded) "fileId"))))) + +(t/deftest resolve-decoder-defaults-to-transit + (let [decode (http.ws/resolve-decoder :transit) + msg {:type :subscribe-file :file-id (uuid/next)}] + (t/is (= msg (decode (tr/encode-str msg {:type :json-verbose})))))) + +(t/deftest resolve-decoder-unexpected-format-falls-back-to-transit + (let [decode (http.ws/resolve-decoder :foo) + msg {:type :subscribe-file :file-id (uuid/next)}] + (t/is (= msg (decode (tr/encode-str msg {:type :json-verbose})))))) + +(t/deftest resolve-decoder-json + (let [decode (http.ws/resolve-decoder :json) + file-id (uuid/next) + decoded (decode (str "{\"type\":\"subscribe-file\"," + "\"fileId\":\"" file-id "\"}"))] + (t/is (= "subscribe-file" (:type decoded))) + (t/is (= (str file-id) (:file-id decoded))))) + +(t/deftest normalize-message-from-json-decoding + (let [file-id (uuid/next) + decoded {:type "subscribe-file" + :file-id (str file-id)} + message (http.ws/normalize-message decoded)] + (t/is (= :subscribe-file (:type message))) + (t/is (uuid? (:file-id message))) + (t/is (= file-id (:file-id message))))) + +(t/deftest normalize-message-noop-for-transit-decoding + (let [file-id (uuid/next) + decoded {:type :subscribe-file :file-id file-id}] + (t/is (= decoded (http.ws/normalize-message decoded))))) + +(t/deftest normalize-message-with-invalid-uuid + (let [message (http.ws/normalize-message + {:type "subscribe-file" :file-id "invalid"})] + (t/is (= :subscribe-file (:type message))) + (t/is (nil? (:file-id message))))) + +(t/deftest normalize-message-team-id + (let [team-id (uuid/next) + message (http.ws/normalize-message + {:type "subscribe-team" :team-id (str team-id)})] + (t/is (= :subscribe-team (:type message))) + (t/is (= team-id (:team-id message))))) + +(defn- ws-request + [{:keys [headers params]}] + {:params (merge {:session-id (str (uuid/next))} params) + ::session/profile-id (uuid/next) + :headers (merge {"upgrade" "websocket" + "connection" "Upgrade"} + headers)}) + +(defn- listener-options + [request] + (let [response (http.ws/http-handler {} request) + listener (::yws/listener response)] + (t/is (map? listener)) + (t/is (fn? (:on-open listener))) + (t/is (fn? (:on-close listener))) + (t/is (fn? (:on-error listener))) + (t/is (fn? (:on-message listener))) + (t/is (fn? (:on-pong listener))) + (::ws/options (meta listener)))) + +(t/deftest http-handler-negotiates-json-codecs + (let [request (ws-request {:headers {"accept" "application/json"}}) + options (listener-options request) + encode (::ws/encode-fn options) + decode (::ws/decode-fn options) + file-id (uuid/next) + raw (str "{\"type\":\"subscribe-file\",\"requestId\":42," + "\"fileId\":\"" file-id "\"}")] + (let [decoded (decode raw)] + (t/is (= "subscribe-file" (:type decoded))) + (t/is (= 42 (:request-id decoded))) + (t/is (= (str file-id) (:file-id decoded)))) + (let [encoded (encode {:type :join-file + :file-id file-id + :request-id 42})] + (t/is (string? encoded)) + (t/is (= "join-file" (get (json/decode encoded) "type"))) + (t/is (= 42 (get (json/decode encoded) "requestId"))) + (t/is (= (str file-id) (get (json/decode encoded) "fileId")))))) + +(t/deftest http-handler-negotiates-json-codecs-from-fmt-param + (let [request (-> (ws-request {:headers {"accept" "application/transit+json"}}) + (assoc :query-params {:_fmt "json"}) + (assoc-in [:params :_fmt] "json")) + options (listener-options request) + encoded ((::ws/encode-fn options) {:type :join-file})] + (t/is (string? encoded)) + (t/is (= "join-file" (get (json/decode encoded) "type"))))) + +(t/deftest http-handler-defaults-to-transit-codecs + (let [request (ws-request {}) + options (listener-options request) + encode (::ws/encode-fn options) + decode (::ws/decode-fn options) + msg {:type :join-file :file-id (uuid/next)}] + (t/is (= msg (decode (encode msg)))))) diff --git a/frontend/src/app/main/repo.cljs b/frontend/src/app/main/repo.cljs index c3dca3b882..ccdaa02865 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] @@ -218,6 +219,7 @@ [id params options] (let [{:keys [response-type stream? + response-format form-data? raw-transit? query-params @@ -240,11 +242,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))} @@ -285,7 +292,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)