Refactor: getting rid of the convoluted logic in the jetty implementation

The previous implementation used a complicated logic involving locks to
protect against an api usage that should not happen. The protection is
now removed, the code is much simpler.

Author: JeremS

libraries/sdk-ring/src/main/starfederation/datastar/clojure/adapter/ring.clj

@@ -23,12 +23,18 @@
   "Returns a ring response that will start a SSE stream.
 
   The status code will be either 200 or the user provided one.
+
   Specific SSE headers are set automatically, the user provided ones will be
   merged. The response body is a sse generator implementing
   `ring.core.protocols/StreamableResponseBody`.
 
+
+  The body is a special value that is part of the SSE machinery and should not
+  be used directly. In other words you must only interact with the SSEGen
+  provided as argument of the different callbacks described below.
+
   In sync mode, the connection is closed automatically when the handler is
-  done running. You need to explicitely close it in rinc async.
+  done running. You need to explicitly close it in rinc async.
 
   Opts:
   - `:status`: status for the HTTP response, defaults to 200
@@ -52,9 +58,7 @@
   "
   [ring-request {:keys [status] :as opts}]
   {:pre [(ac/on-open opts)]}
-  (let [sse-gen (impl/->sse-gen)]
-    {:status (or status 200)
-     :headers (ac/headers ring-request opts)
-     :body sse-gen
-     ::impl/opts opts}))
-
+  {:status     (or status 200)
+   :headers    (ac/headers ring-request opts)
+   :body       (impl/->sse-gen)
+   ::impl/opts opts})

libraries/sdk-ring/src/main/starfederation/datastar/clojure/adapter/ring/impl.clj

@@ -34,42 +34,22 @@
                        ^:unsynchronized-mutable on-exception]
   rp/StreamableResponseBody
   (write-body-to-stream [this response output-stream]
-    (.lock lock)
+    (let [opts (::opts response)
+          on-open (ac/on-open opts)]
 
-    ;; already initialized, unlock and throw, we are out
-    (when send!
-      (.unlock lock)
-      (throw (ex-info "Reused SSE-gen as several ring responses body. Don't do this." {})))
+      ;; Set the SSEGenerator's state
+      (set! send! (->send output-stream opts))
+      (set! on-exception (or (ac/on-exception opts)
+                             ac/default-on-exception))
+      (when-let [cb (ac/on-close opts)]
+        (set! on-close cb))
 
-    (let [!error (volatile! nil)]
-      (try
-        ;; initializing the internal state
-        (let [opts (::opts response)]
-          (set! send! (->send output-stream opts))
-          (set! on-exception (or (ac/on-exception opts)
-                                 ac/default-on-exception))
-          (when-let [cb (ac/on-close opts)]
-            (set! on-close cb)))
+      ;; flush the HTTP headers as soon as possible
+      (.flush ^OutputStream output-stream)
+
+      ;; Call on-open
+      (on-open this)))
 
-        ;; flush the HTTP headers
-        (.flush ^OutputStream output-stream)
-        true ;; dummy return
-        ;; We catch everything here, if not a Throwable may pass through
-        ;; !error won't catch it, on-open would be called
-        (catch Throwable t
-          (vreset! !error t))
-        (finally
-          ;; Any exception should have been caught,
-          ;; the setup the internal state is done,
-          ;; the HTTP headers are sent
-          ;; we can now release the lock now
-          (.unlock lock)
-          (if-let [e @!error]
-            (throw e) ;; if error throw, the lock is already released
-            ;; if all is ok call on-open, it can safely throw...
-            (when-let [on-open (-> response ::opts ac/on-open)]
-              (on-open this)))))))
- 
   p/SSEGenerator
   (send-event! [this event-type data-lines opts]
     (u/lock! lock
@@ -109,10 +89,10 @@
   (close [this]
     (p/close-sse! this)))
 
-
-(defn ->sse-gen []
+(defn ->sse-gen
+  {:tag SSEGenerator}
+  []
   (SSEGenerator. nil
                  (ReentrantLock.)
                  nil
                  nil))
-

src/test/adapter-ring/starfederation/datastar/clojure/adapter/ring/impl_test.clj

@@ -10,108 +10,6 @@
   (:import
     [java.io ByteArrayOutputStream]))
 
-;; -----------------------------------------------------------------------------
-;; Testing the gnarly adapter setup by simulating it
-;; -----------------------------------------------------------------------------
-(defn ->lock [] (volatile! 0))
-(defn lock!   [!l] (vswap! !l inc))
-(defn unlock! [!l] (vswap! !l dec))
-
-
-(defn throw-key [k] (throw (ex-info "" {:k k})))
-(defn throw-already-used  [] (throw-key :already-used-error))
-(defn throw-writter-error [] (throw-key :writer-ctr-error))
-(defn throw-flush-errror  [] (throw-key :flush-error))
-(defn throw-on-open       [] (throw-key :on-open-error))
-
-
-(defn set-writer! [!ref behavior]
-  (case behavior
-    :ok (vreset! !ref :new-writer)
-    (throw-writter-error)))
-
-(defn flush-headers! [v]
-  (case v
-    :ok nil
-    (throw-flush-errror)))
-
-(defn on-open [v]
-  (case v
-    :ok nil
-    (throw-on-open)))
-
-
-(defn write-body-to-stream-simulation
-  "Here we mimick the behavior of the initialisation of ring SSE responses
-  We capture lock, internal state, errors and return value to check several
-  properties."
-  [writer set-writer-v flush-v on-open-v]
-  (let [!l (->lock)
-        !writer (volatile! :old-writer)
-        !return (volatile! nil)]
- 
-    ;; The code actually mimicking
-    (try
-      (lock! !l)
-      (when writer
-        (unlock! !l)
-        (throw-already-used))
-
-      (let [!error (volatile! nil)]
-        (try
-          (set-writer! !writer set-writer-v)
-          (flush-headers! flush-v)
-          (vreset! !return :success)
-          (catch Throwable t
-            (vreset! !error t))
-          (finally
-            (unlock! !l)
-            (if-let [e @!error]
-              (throw e)
-              (on-open on-open-v)))))
-      (catch Throwable t
-        (vreset! !return t)))
-
-    {:lock @!l
-     :writer @!writer
-     :return (let [r  @!return]
-               (or
-                 (-> r ex-data :k)
-                 r))}))
-
-
-(defn make-all-cases []
-  (for [writer     [nil :old-writer]
-        set-writer [:ok :throw]
-        flush      [:ok :throw]
-        on-open    [:ok :throw]]
-    [writer set-writer flush on-open]))
-
-
-(defn expected [writer set-writer! flush on-open]
-  (cond
-    writer                 {:lock 0 :writer :old-writer :return :already-used-error}
-    (= set-writer! :throw) {:lock 0 :writer :old-writer :return :writer-ctr-error}
-    (= flush       :throw) {:lock 0 :writer :new-writer :return :flush-error}
-    (= on-open     :throw) {:lock 0 :writer :new-writer :return :on-open-error}
-    :else                  {:lock 0 :writer :new-writer :return :success}))
-
-
-(defn run-test-case [test-case]
-  {:res (apply write-body-to-stream-simulation test-case)
-   :expected  (apply expected test-case)})
-
-
-(defn case-coherent? [test-case]
-  (let [res (run-test-case test-case)]
-    (= (:res res) (:expected res))))
-
-
-(defdescribe simulate-write-body-to-stream
-  (it "manages locks and errors properly"
-   (doseq [test-case (make-all-cases)]
-     (expect (case-coherent? test-case) (str test-case)))))        
-
 
 ;; -----------------------------------------------------------------------------
 ;; Basic sending of a SSE event without any server
@@ -120,7 +18,8 @@
   (d*/patch-elements! (at/->sse-gen) "msg"))
 
 (defn send-SSE-event [response]
-  (let [baos (ByteArrayOutputStream.)]
+  (let [response (assoc-in response [::impl/opts ac/on-open] (fn [_sse]))
+        baos (ByteArrayOutputStream.)]
     (with-open [sse-gen (impl/->sse-gen)
                 baos baos]
       (p/write-body-to-stream sse-gen response baos)
@@ -133,23 +32,22 @@
 (defdescribe simple-test
   (it "can send events with a using temp buffers"
     (send-SSE-event {}))
- 
+
   (it "can send events with a using a persistent buffered reader"
     (send-SSE-event {::impl/opts {ac/write-profile ac/buffered-writer-profile}}))
 
   (it "can send gziped events with a using temp buffers"
     (send-SSE-event {::impl/opts {ac/write-profile ac/gzip-profile
                                   :gzip? true}}))
- 
+
   (it "can send gziped events with a using a persistent buffered reader"
     (send-SSE-event {::impl/opts {ac/write-profile ac/gzip-buffered-writer-profile
                                   :gzip? true}})))
- 
+
 
 
 
 (comment
+  (user/reload!)
   (require '[lazytest.repl :as ltr])
-  (ltr/run-test-var #'simulate-write-body-to-stream)
   (ltr/run-test-var #'simple-test))
-