diff --git a/frontend/src/app/main/data/workspace.cljs b/frontend/src/app/main/data/workspace.cljs index a42c367c3b..28e14bc3bd 100644 --- a/frontend/src/app/main/data/workspace.cljs +++ b/frontend/src/app/main/data/workspace.cljs @@ -416,7 +416,10 @@ ptk/WatchEvent (watch [_ state stream] - (let [stoper-s (rx/filter (ptk/type? ::finalize-workspace) stream) + (let [;; A reload starts a new `initialize-workspace`, which replaces these streams + stoper-s (rx/filter #(or (ptk/type? ::finalize-workspace %) + (ptk/type? ::initialize-workspace %)) + stream) rparams (rt/get-params state) features (features/get-enabled-features state team-id) render-wasm-enabled? (features/active-feature? state "render-wasm/v1") diff --git a/frontend/src/app/main/data/workspace/notifications.cljs b/frontend/src/app/main/data/workspace/notifications.cljs index b2637bbce7..2cd1562589 100644 --- a/frontend/src/app/main/data/workspace/notifications.cljs +++ b/frontend/src/app/main/data/workspace/notifications.cljs @@ -48,12 +48,29 @@ (declare handle-pointer-send) (declare handle-export-update) +(defn- resubscribed? + "True when `event` is a new subscription to `file-id`, which a reload of + the open file starts." + [file-id event] + (and (ptk/type? ::initialize event) + (= file-id (:file-id (deref event))))) + (defn initialize + "Subscribes to the notifications of `file-id` and its team until the file + is left or a new subscription starts. A new subscription to the same file + replaces this one, so a reload never doubles the handled messages." [team-id file-id] (ptk/reify ::initialize + IDeref + (-deref [_] {:file-id file-id}) + ptk/WatchEvent (watch [_ state stream] - (let [stopper (rx/filter (ptk/type? ::finalize) stream) + (let [stopped-by (volatile! nil) + stopper (->> stream + (rx/filter #(or (ptk/type? ::finalize %) + (ptk/type? ::initialize %))) + (rx/tap #(vreset! stopped-by %))) profile-id (:profile-id state) organization-id (dm/get-in state [:teams team-id :organization :id]) @@ -107,7 +124,13 @@ (rx/take-until stopper))] - (rx/concat stream (rx/of (dws/send endmsg))))))) + (rx/concat + stream + ;; The server keeps one file subscription per connection: an + ;; unsubscribe sent after the new subscribe would end it + (rx/defer #(if (resubscribed? file-id @stopped-by) + (rx/empty) + (rx/of (dws/send endmsg))))))))) (defn- handle-change-team-role [{:keys [role] :as msg}] diff --git a/frontend/src/app/main/data/workspace/shape_layout.cljs b/frontend/src/app/main/data/workspace/shape_layout.cljs index df614dfe7a..9062ad1c5c 100644 --- a/frontend/src/app/main/data/workspace/shape_layout.cljs +++ b/frontend/src/app/main/data/workspace/shape_layout.cljs @@ -150,7 +150,10 @@ (ptk/reify ::initialize-shape-layout ptk/WatchEvent (watch [_ _ stream] - (let [stopper (rx/filter (ptk/type? ::finalize-shape-layout) stream)] + (let [;; A workspace reload starts a new one, which replaces this one + stopper (rx/filter #(or (ptk/type? ::finalize-shape-layout %) + (ptk/type? ::initialize-shape-layout %)) + stream)] (->> stream (rx/filter (ptk/type? :layout/update)) ;; Keeps each update's page and the shapes that can be laid out; diff --git a/frontend/src/app/main/data/workspace/texts.cljs b/frontend/src/app/main/data/workspace/texts.cljs index 4b533c8fa5..0464cee98b 100644 --- a/frontend/src/app/main/data/workspace/texts.cljs +++ b/frontend/src/app/main/data/workspace/texts.cljs @@ -96,7 +96,10 @@ (ptk/reify ::initialize-text-reflow ptk/WatchEvent (watch [_ _ stream] - (let [stopper (rx/filter (ptk/type? ::finalize-text-reflow) stream) + (let [;; A workspace reload starts a new one, which replaces this one + stopper (rx/filter #(or (ptk/type? ::finalize-text-reflow %) + (ptk/type? ::initialize-text-reflow %)) + stream) page-stopper (rx/filter page-finalize? stream)] (->> stream (rx/filter (ptk/type? :text/reflow)) diff --git a/frontend/test/frontend_tests/data/workspace_notifications_test.cljs b/frontend/test/frontend_tests/data/workspace_notifications_test.cljs index 7c6d362941..0bb9d87f95 100644 --- a/frontend/test/frontend_tests/data/workspace_notifications_test.cljs +++ b/frontend/test/frontend_tests/data/workspace_notifications_test.cljs @@ -12,12 +12,14 @@ [app.main.data.changes :as dch] [app.main.data.modal :as modal] [app.main.data.notifications :as ntf] + [app.main.data.websocket :as dws] [app.main.data.workspace :as-alias dw] [app.main.data.workspace.notifications :as dwntf] [app.util.i18n :as i18n] [beicon.v2.core :as rx] [cljs.test :as t :include-macros true] [frontend-tests.helpers.async :as async] + [frontend-tests.helpers.mock :as mock] [potok.v2.core :as ptk])) (defn- file-change-msg @@ -143,3 +145,95 @@ :session-id (uuid/next) :vern 2}))] (t/is (empty? events)))) + +(defn- ^:async subscription-messages + "Runs `scenario` with a store over `team-id`, `dws/send` and the pointer + handler stubbed. `scenario` gets the store and a function that delivers + a peer's pointer message for a file. Resolves to the messages sent over + the websocket and the number of handled pointer messages." + [team-id scenario] + (let [sent (atom []) + handled (atom 0) + result (atom nil) + store (ptk/store {:state {:profile-id (uuid/next) + :session-id (uuid/next) + :teams {team-id {:id team-id}}} + :on-error #(t/is false (str %))}) + pointer (fn [file-id] + (ptk/emit! store (ptk/data-event ::dws/message + {:type :pointer-update + :topic file-id + :file-id file-id + :session-id (uuid/next)})))] + (await + (mock/with-mocks* + {dws/send (mock/stub (fn [msg] + (swap! sent conj msg) + (ptk/data-event ::sent msg))) + dwntf/handle-pointer-update (mock/stub (fn [_] + (swap! handled inc) + (ptk/data-event ::handled {})))} + (try + (await (scenario store pointer)) + ;; Taken before the store ends, which ends the subscriptions too + (reset! result {:sent (mapv #(select-keys % [:type :file-id]) @sent) + :handled @handled}) + (finally + (rx/dispose! store))))) + @result)) + +;; Scenario: the open file reloads three times (a peer's restore, the +;; restorer's own version reload, the rollback fallback), each starting a +;; new subscription, and then a peer's message arrives. Proves: one live +;; subscription per open file, so each message is handled once, and no +;; unsubscribe ends the server's subscription of the reloaded file. +(t/deftest ^:async a-reload-replaces-the-file-subscription + (let [team-id (uuid/next) + file-id (uuid/next) + result (await + (subscription-messages + team-id + (^:async fn [store pointer] + (ptk/emit! store + (dwntf/initialize team-id file-id) + (dwntf/initialize team-id file-id) + (dwntf/initialize team-id file-id)) + (await (async/settle)) + (pointer file-id) + (await (async/settle)))))] + (t/is (= 1 (:handled result)) "the message is handled once") + (t/is (= 3 (count (filter #(= :subscribe-file (:type %)) (:sent result))))) + (t/is (not-any? #(= :unsubscribe-file (:type %)) (:sent result))))) + +;; Scenario: the user leaves a file, and in another case opens another file +;; without leaving the first one. Proves: the subscription ends with an +;; unsubscribe of the file it was for, and the old file's messages are no +;; longer handled. +(t/deftest ^:async leaving-or-switching-files-ends-the-file-subscription + (let [team-id (uuid/next) + file-a (uuid/next) + file-b (uuid/next) + left (await + (subscription-messages + team-id + (^:async fn [store _] + (ptk/emit! store + (dwntf/initialize team-id file-a) + (dwntf/finalize file-a)) + (await (async/settle))))) + switch (await + (subscription-messages + team-id + (^:async fn [store pointer] + (ptk/emit! store + (dwntf/initialize team-id file-a) + (dwntf/initialize team-id file-b)) + (await (async/settle)) + (pointer file-a) + (pointer file-b) + (await (async/settle)))))] + (t/is (= [{:type :unsubscribe-file :file-id file-a}] + (filterv #(= :unsubscribe-file (:type %)) (:sent left)))) + (t/is (= [{:type :unsubscribe-file :file-id file-a}] + (filterv #(= :unsubscribe-file (:type %)) (:sent switch)))) + (t/is (= 1 (:handled switch)) "only the new file's message is handled"))) diff --git a/frontend/test/frontend_tests/data/workspace_undo_test.cljs b/frontend/test/frontend_tests/data/workspace_undo_test.cljs index 465e0b01ba..55a1ae0cca 100644 --- a/frontend/test/frontend_tests/data/workspace_undo_test.cljs +++ b/frontend/test/frontend_tests/data/workspace_undo_test.cljs @@ -12,9 +12,14 @@ [app.common.test-helpers.ids-map :as cthi] [app.common.uuid :as uuid] [app.main.data.changes :as dch] + [app.main.data.comments :as dcmt] + [app.main.data.fonts :as df] + [app.main.data.workspace :as dw] [app.main.data.workspace.undo :as dwu] [beicon.v2.core :as rx] [cljs.test :as t :include-macros true] + [frontend-tests.helpers.async :as async] + [frontend-tests.helpers.mock :as mock] [frontend-tests.helpers.state :as ths] [potok.v2.core :as ptk])) @@ -236,3 +241,34 @@ (t/is (= 2 (:index (forget undo-state #{(uuid/next)} 2)))) (t/is (= 0 (:index (forget undo-state #{g2} 2))) "a cut history cannot point past its end"))) + +;; Scenario: the workspace opens and then reloads twice (a peer's restore, +;; the rollback fallback), each reload running `initialize-workspace` again +;; without leaving the workspace; then one commit is made. Proves: each +;; reload replaces the workspace streams, so the commit adds one undo entry, +;; not one per reload. +(t/deftest ^:async a-workspace-reload-adds-one-undo-entry-per-commit + (let [file (cthf/sample-file :file1 :page-label :page1) + team-id (uuid/next) + store (ths/setup-store file {:renderer :svg}) + loaded (fn [type] (mock/stub (fn [& _] (ptk/data-event type {}))))] + (await + (mock/with-mocks* + {dw/initialize-file (loaded ::file-requested) + dcmt/retrieve-comment-threads (loaded ::threads-requested) + dcmt/fetch-profiles (loaded ::profiles-requested) + df/fetch-fonts (loaded ::fonts-requested)} + (try + (ptk/emit! store + #(assoc % :current-team-id team-id) + (dw/initialize-workspace team-id (:id file)) + (ptk/event ::dw/reload-current-file) + (ptk/event ::dw/reload-current-file)) + (await (async/settle)) + (ptk/emit! store (rename-page (cthf/current-page-id file) "Renamed")) + (await (async/wait-for #(seq (get-in @store [:workspace-undo :items])) + "the commit reaches the undo history")) + (await (async/settle)) + (t/is (= 1 (count (get-in @store [:workspace-undo :items])))) + (finally + (rx/dispose! store)))))))