Merge pull request #528 from kumarshantanu/master

Flush response body stream when writing seqs

Author: weavejester

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))))))