Flush response body stream when writing seqs

Update the StreamableResponseBody implementation for ISeq to flush the
OutputStream after each element. This prevents unnecessary delays when
dealing with blocking seqs, for example, when being used to implement
SSEs.

Author: kumarshantanu

ring-core-protocols/src/ring/core/protocols.clj

@@ -63,7 +63,8 @@
   (write-body-to-stream [body response output-stream]
     (let [writer (response-writer response output-stream)]
       (doseq [chunk body]
-        (.write writer (str chunk)))
+        (.write writer (str chunk))
+        (.flush writer))
       (.close writer)))
   java.io.InputStream
   (write-body-to-stream [body _ ^OutputStream output-stream]

ring-core-protocols/test/ring/core/test/protocols.clj

@@ -2,7 +2,9 @@
   (:require [clojure.test :refer :all]
             [clojure.java.io :as io]
             [ring.core.protocols :refer :all])
-  (:import [java.io SequenceInputStream IOException InputStream OutputStream]))
+  (:import [java.io
+            SequenceInputStream IOException InputStream
+            ByteArrayOutputStream OutputStream]))
 
 (deftest test-write-body-defaults
   (testing "byte-array"
@@ -113,3 +115,32 @@
           response {:body "Hello World"}]
       (write-body-to-stream (:body response) response output)
       (is (not @flushed?)))))
+
+(deftest test-flushing-for-seq
+  (testing "seqs with delayed elements"
+    (let [output    (ByteArrayOutputStream.)
+          counter   (atom 0)
+          event-str "data: sample\n\n"
+          gen-event (fn [] (swap! counter inc) event-str)
+          gen-delay (fn [] (Thread/sleep 100))
+          lazy-exec (fn lazy-exec [[f & more]]
+                      (when f
+                        (cons (f) (lazy-seq (lazy-exec more)))))
+          continue? (atom true)
+          resp-body (->> (repeat gen-event)
+                         (interpose gen-delay)
+                         (take-while (fn [_] @continue?))
+                         (lazy-exec))
+          response  {:body resp-body}]
+      ;; first sequence element is already evaluated by lazy-exec
+      (is (= 1 @counter) "counter bump - first seq element already evaluated")
+      (is (= "" (str output)) "empty output because body not written yet")
+      (try
+        (future ; needs to run concurrently so we can observe flushing
+          (write-body-to-stream (:body response) response output))
+        (Thread/sleep 150)
+        (is (= 2 @counter) "two seq elements evaluated yet")
+        (is (= (str event-str event-str)
+               (str output)) "two events written to output yet")
+        (finally
+          (reset! continue? false))))))