Repository: ring-clojure/ring (clojure)

Issue:
Exception handling for InputStream
Hi there,

I have a problem returning `InputStream` as body using latest Ring(1.8.2) and Jetty.
I noticed that if `InputStream` returned a few bytes and then an Exception was thrown(could happen in my case) then http-client received partial data but thinking that he received all the data.

Here is test/example of this issue
https://gist.github.com/lightZebra/56bc2f21218dbec52214fb0af9adb2b4

If I understood correctly this happened because of [with-open](https://github.com/ring-clojure/ring/blob/master/ring-core/src/ring/core/protocols.clj#L35) inside `StreamableResponseBody` for InputStream.
Exception was thrown and ring closed OutputStream like streaming finished successfully and jetty couldn't finish it correctly
I didn't find in Jetty's documentation information that `Handler` should close `OutputStream`, so I tried `extend-protocol` and removed `with-open` and client received an error, like it should

I'm kinda new to ring and jetty, am I doing something wrong or it is a bug?

Pull request:
OutputStream will not be closed in case of error
so underlying server could handle it properly.
Fixed issue: termination chunk was send after
streaming interruption by error.

Fixes #420.

Patch:
diff --git a/ring-core/src/ring/core/protocols.clj b/ring-core/src/ring/core/protocols.clj
index a2d1c189..4c4c30df 100644
--- a/ring-core/src/ring/core/protocols.clj
+++ b/ring-core/src/ring/core/protocols.clj
@@ -19,25 +19,28 @@
 (extend-protocol StreamableResponseBody
   (Class/forName "[B")
   (write-body-to-stream [body _ ^OutputStream output-stream]
-    (with-open [out output-stream]
-      (.write out ^bytes body)))
+    (.write output-stream ^bytes body)
+    (.close output-stream))
   String
   (write-body-to-stream [body response output-stream]
-    (with-open [writer (response-writer response output-stream)]
-      (.write writer body)))
+    (doto (response-writer response output-stream)
+      (.write body)
+      (.close)))
   clojure.lang.ISeq
   (write-body-to-stream [body response output-stream]
-    (with-open [writer (response-writer response output-stream)]
+    (let [writer (response-writer response output-stream)]
       (doseq [chunk body]
-        (.write writer (str chunk)))))
+        (.write writer (str chunk)))
+      (.close writer)))
   java.io.InputStream
   (write-body-to-stream [body _ ^OutputStream output-stream]
-    (with-open [out output-stream, body body]
-      (io/copy body out)))
+    (with-open [body body]
+      (io/copy body output-stream))
+    (.close output-stream))
   java.io.File
   (write-body-to-stream [body _ ^OutputStream output-stream]
-    (with-open [out output-stream]
-      (io/copy body out)))
+    (io/copy body output-stream)
+    (.close output-stream))
   nil
   (write-body-to-stream [_ _ ^java.io.OutputStream output-stream]
     (.close output-stream)))


Test patch:
diff --git a/ring-core/test/ring/core/test/protocols.clj b/ring-core/test/ring/core/test/protocols.clj
index 508a5993..ba55e871 100644
--- a/ring-core/test/ring/core/test/protocols.clj
+++ b/ring-core/test/ring/core/test/protocols.clj
@@ -1,7 +1,8 @@
 (ns ring.core.test.protocols
   (:require [clojure.test :refer :all]
             [clojure.java.io :as io]
-            [ring.core.protocols :refer :all]))
+            [ring.core.protocols :refer :all])
+  (:import [java.io SequenceInputStream IOException InputStream OutputStream]))
 
 (deftest test-write-body-defaults
   (testing "byte-array"
@@ -53,3 +54,41 @@
           response {:body nil}]
       (write-body-to-stream (:body response) response output)
       (is (= "" (.toString output))))))
+
+(defn output-stream-with-close-flag []
+  (let [stream-closed? (atom false)
+        output-stream  (proxy [OutputStream] []
+                         (write
+                           ([])
+                           ([^bytes _])
+                           ([^bytes _ _ _]))
+                         (close [] (reset! stream-closed? true)))]
+    [output-stream stream-closed?]))
+
+(defn error-input-stream []
+  (proxy [InputStream] []
+    (read
+      ([] (throw (IOException. "test error")))
+      ([^bytes _] (throw (IOException. "test error")))
+      ([^bytes _ _ _] (throw (IOException. "test error"))))))
+
+(deftest test-error-when-writing-body
+  (testing "input streams with error"
+    (let [[output closed?] (output-stream-with-close-flag)
+          resource         (io/resource "ring/assets/hello world.txt")
+          input-stream     (SequenceInputStream.
+                            (io/input-stream resource)
+                            (error-input-stream))
+          response         {:body input-stream}]
+      (is (thrown? IOException
+                   (write-body-to-stream (:body response) response output)))
+      (is (= false @closed?))))
+
+  (testing "seqs with error"
+    (let [[output closed?] (output-stream-with-close-flag)
+          response         {:body (lazy-cat
+                                   ["a"]
+                                   (throw (IOException. "test error")))}]
+      (is (thrown? IOException
+                   (write-body-to-stream (:body response) response output)))
+      (is (= false @closed?)))))
diff --git a/ring-jetty-adapter/test/ring/adapter/test/jetty.clj b/ring-jetty-adapter/test/ring/adapter/test/jetty.clj
index 8e624b92..d1a5832a 100644
--- a/ring-jetty-adapter/test/ring/adapter/test/jetty.clj
+++ b/ring-jetty-adapter/test/ring/adapter/test/jetty.clj
@@ -10,7 +10,10 @@
            [org.eclipse.jetty.server Server Request SslConnectionFactory]
            [org.eclipse.jetty.server.handler AbstractHandler]
            [java.net ServerSocket ConnectException]
-           [java.security KeyStore]))
+           [java.security KeyStore]
+           [java.io SequenceInputStream ByteArrayInputStream InputStream
+                    IOException]
+           [org.apache.http MalformedChunkCodingException]))
 
 (defn- hello-world [request]
   {:status  200
@@ -395,6 +398,46 @@
               (recur (inc i))))
           (is (= thread-count (count (all-threads)))))))))
 
+(defn- chunked-stream-with-error
+  ([request]
+   {:status  200
+    :headers {"Transfer-Encoding" "chunked"}
+    :body    (SequenceInputStream.
+               (ByteArrayInputStream. (.getBytes (str (range 100000)) "UTF-8"))
+               (proxy [InputStream] []
+                 (read
+                   ([] (throw (IOException. "test error")))
+                   ([^bytes _] (throw (IOException. "test error")))
+                   ([^bytes _ _ _] (throw (IOException. "test error"))))))})
+  ([request response raise]
+   (response (chunked-stream-with-error request))))
+
+(defn- chunked-lazy-seq-with-error
+  ([request]
+   {:status  200
+    :headers {"Transfer-Encoding" "chunked"}
+    :body    (lazy-cat (range 100000)
+                       (throw (IOException. "test error")))})
+  ([request response raise]
+   (response (chunked-lazy-seq-with-error request))))
+
+(deftest streaming-with-error
+  (testing "chunked stream without sending termination chunk on error"
+    (with-server chunked-stream-with-error {:port test-port}
+      (is (thrown? MalformedChunkCodingException (http/get test-url)))))
+
+  (testing "chunked sequence without sending termination chunk on error"
+    (with-server chunked-lazy-seq-with-error {:port test-port}
+      (is (thrown? MalformedChunkCodingException (http/get test-url)))))
+
+  (testing "async chunked stream without sending termination chunk on error"
+    (with-server chunked-stream-with-error {:port test-port :async? true}
+      (is (thrown? MalformedChunkCodingException (http/get test-url)))))
+
+  (testing "async chunked sequence without sending termination chunk on error"
+    (with-server chunked-lazy-seq-with-error {:port test-port :async? true}
+      (is (thrown? MalformedChunkCodingException (http/get test-url))))))
+
 (def thread-exceptions (atom []))
 
 (defn- hello-world-cps [request respond raise]