diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 52e1c8b..c2f1f2f 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -1,21 +1,20 @@ name: tests on: [push, pull_request] -# One pinned jolt, not whatever the latest release happens to be. The supported -# floor is 0.8.16, which deps.edn declares as :jolt/min-version: that is where -# java.util.concurrent.CompletableFuture entered the runtime and this library -# stopped claiming it. 0.8.9 is where java.util.zip entered the runtime, on -# the zlib every jolt binary links, and where this library stopped shipping -# gzip/deflate shims of its own over the system libz. An older jolt has neither the classes nor the runtime's answer -# for them, so a gzip response would fail to decode there; the floor refuses -# the project instead. Earlier floors, for the record: 0.8.1 (java.util.concurrent's -# executor interfaces, which babashka.http-client's ->Executor type-checks), -# 0.8.0 (jolt.ffi/write's argument order), 0.7.19 (jolt's own java.io byte -# streams answering the full surface). +# CI tracks jolt's nightly build (release tag vnightly, the rolling build of +# jolt's main), so a jolt change that breaks this library shows up the day it +# lands rather than at the next release. # -# The pin is the DECLARED FLOOR, not the newest release, so CI tests the oldest -# jolt deps.edn claims to support. Raise both together. One leg is enough now: -# the `:as :stream` live-body path that used to need a second, newer leg -# (0.8.8's abstract-class method inheritance) is below the floor. +# That means CI no longer exercises the declared floor. deps.edn still declares +# :jolt/min-version 0.8.16, which is where java.util.concurrent.CompletableFuture +# entered the runtime and this library stopped claiming it. Earlier floors, for +# the record: 0.8.9 (java.util.zip entered the runtime, on the zlib every jolt +# binary links, and this library stopped shipping gzip/deflate shims of its own +# over the system libz), 0.8.1 (java.util.concurrent's executor interfaces, +# which babashka.http-client's ->Executor type-checks), 0.8.0 (jolt.ffi/write's +# argument order), 0.7.19 (jolt's own java.io byte streams answering the full +# surface). The `:as :stream` live-body path needs 0.8.8's abstract-class method +# inheritance, which is below the floor. Raising the floor is a deps.edn change +# and should be checked against that release by hand. jobs: test: @@ -23,7 +22,7 @@ jobs: strategy: fail-fast: false matrix: - jolt: ['0.8.16'] + jolt: ['nightly'] name: test (jolt ${{ matrix.jolt }}) env: JOLT_VERSION: ${{ matrix.jolt }} diff --git a/README.md b/README.md index 99e63bb..5350c5b 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,8 @@ send runs on jolt's own future pool. Each is recorded on the client and read back, so a caller that sets and inspects one sees what it set. WebSocket negotiates no extensions, so no `permessage-deflate`. -`:as :stream` (`BodyHandlers/ofInputStream`) is live **on jolt 0.8.8 and up**: +`:as :stream` is live on both clients — `BodyHandlers/ofInputStream` on +`babashka.http-client`, `HttpURLConnection.getInputStream` under `clj-http-lite`: the response comes back as soon as the headers are in and the body is pulled off the socket as it is read, so an SSE feed, an LLM token stream or a log tail works. The stream is framed by the response — `Content-Length`, chunked, or the peer's close — so it ends where @@ -96,10 +97,12 @@ rather than being cut (#26). For a hard total cap, streamed or not. The live stream is a reify `java.io.InputStream`, which needs the abstract-class -method inheritance jolt gained in 0.8.8. Below that — the declared floor is -0.8.1 — the transport probes for one, does not find it, and keeps reading bodies -to completion, so `:as :stream` is a stream over a finished body as it was -before. CI runs both. +method inheritance jolt gained in 0.8.8. The declared floor is above that now, +but the transport still probes for one, and on a jolt without it keeps reading +bodies to completion, so `:as :stream` is a stream over a finished body. + +`clj-http-lite` reads every other `:as` by draining that same stream, so its +buffered coercions, gzip/deflate decoding included, are unchanged. Two things to know. The connection belongs to the stream until the body is read to its end or the stream is closed, so a caller that abandons a stream should diff --git a/src/jolt/http/platform.clj b/src/jolt/http/platform.clj index e0b707a..1ee6c83 100644 --- a/src/jolt/http/platform.clj +++ b/src/jolt/http/platform.clj @@ -55,6 +55,33 @@ (core/set-max-response-ms! ms)) +;; The body is left on the wire. getResponseCode and the header getters only +;; need the head, and clj-http-lite reaches the body through getInputStream / +;; getErrorStream alone, so the call can come back at the headers and the body +;; can be read as it arrives: `:as :stream` over an SSE feed or an LLM token +;; stream used to wait for the whole response before returning anything. The +;; buffered coercions lose nothing — clj-http-lite drains the stream to EOF +;; itself, and gzip/deflate are decoded from those bytes afterwards. +;; +;; Unlike the java.net.http path, the socket's read timeout stays armed for the +;; body (read-response's body-unbounded? is false here): on HttpURLConnection +;; setReadTimeout is SO_RCVTIMEO and bounds every read of the body, which is +;; what lets an SSE reader bound the gap between events. set-max-response-ms! +;; still caps the body's total time. +;; +;; Where the runtime cannot carry a live body (jolt below 0.8.8, see +;; core/reify-input-streams?) read-response hands back a buffered body and +;; nothing here changes. + +(defn- discard-body! + "Drain and let go of a streamed body nobody will read — the response to a + redirect being followed. Left alone, its connection stays open until GC and + could never go back to the pool." + [resp] + (when-let [pull (:body-pull resp)] + ((:release! resp) (try (loop [] (if (nil? (pull)) true (recur))) + (catch Throwable _ false))))) + (defn- perform! [conn] (loop [url (tget conn :url) method (tget conn :method) @@ -66,17 +93,30 @@ open! (fn [] (connect-stream (tget url :host) (effective-port url) https? (tget conn :insecure) (tget conn :read-timeout) (tget conn :connect-timeout))) + ;; A buffered response is done with its connection here: pooled when + ;; the response says it may be kept, closed otherwise, including on + ;; the way out of a failure. A streamed one still has its body on the + ;; socket, so ownership passes to the body stream through :release!, + ;; which takes true once the body was read to its end. once (fn [stream received] - (let [ok (atom false)] + (let [ok (atom false) + streaming (atom false)] (try (s-write stream (build-request method url (tget conn :req-headers) body)) - (let [r (read-response stream nil method received)] - (reset! ok (:reusable? r)) - r) + (let [r (read-response stream nil method received true)] + (if (:body-pull r) + (do (reset! streaming true) + (assoc r :release! + (fn [complete?] + (if (and complete? (:reusable? r)) + (core/pool-release! key stream) + (try (s-close stream) (catch Throwable _ nil)))))) + (do (reset! ok (:reusable? r)) r))) (finally - (if @ok - (core/pool-release! key stream) - (try (s-close stream) (catch Throwable _ nil))))))) + (when-not @streaming + (if @ok + (core/pool-release! key stream) + (try (s-close stream) (catch Throwable _ nil)))))))) resp (if-let [pooled (core/pool-acquire key)] ;; a connection the peer retired since the last request fails ;; before answering — cleanly or with a reset, depending on the @@ -94,15 +134,35 @@ (redirect-statuses (:status resp)) (or (= method "GET") (= method "HEAD")) loc (< redirects 20)) - (recur (resolve-location url loc) + (do (discard-body! resp) + (recur (resolve-location url loc) (if (= (:status resp) 303) "GET" method) - (inc redirects)) + (inc redirects))) (do (tput! conn :response resp) (tput! conn :performed true) resp))))) (defn- ensure-performed! [conn] (when-not (tget conn :performed) (perform! conn)) (tget conn :response)) +(defn- body-stream + "The response body as an InputStream: live when the transport left it on the + wire, over the bytes otherwise. Made once, so getInputStream, getErrorStream + and disconnect all see the same stream." + [conn] + (or (tget conn :body-stream) + (let [resp (ensure-performed! conn) + s (if-let [pull (:body-pull resp)] + (core/make-body-stream pull (:release! resp)) + (make-bais (:body resp)))] + (tput! conn :body-stream s) + s))) + +(defn- input-stream [conn] + (let [resp (ensure-performed! conn)] + (if (>= (:status resp) 400) + (throw-typed "java.io.IOException" (str "Server returned HTTP response code: " (:status resp))) + (body-stream conn)))) + (defn- open-connection [url] (let [c (tt :jolt/http-url-connection)] (tput! c :url url) @@ -112,6 +172,7 @@ (tput! c :read-timeout nil) (tput! c :connect-timeout nil) (tput! c :insecure false) (tput! c :out-buffer nil) (tput! c :performed false) (tput! c :response nil) + (tput! c :body-stream nil) c)) ;; --- install --------------------------------------------------------------- @@ -239,7 +300,9 @@ "toString" (fn [self] (tget self :spec)) "toExternalForm" (fn [self] (tget self :spec)) "openConnection" (fn [self] (open-connection self)) - "openStream" (fn [self] (make-bais (:body (ensure-performed! (open-connection self)))))}) + ;; openConnection().getInputStream(), as on the JVM: an error status + ;; throws rather than handing back the error page as if it were the body. + "openStream" (fn [self] (input-stream (open-connection self)))}) ;; java.net.HttpURLConnection (__register-class-methods! :jolt/http-url-connection @@ -268,7 +331,13 @@ "setChunkedStreamingMode" (fn [self _n] nil) "setFixedLengthStreamingMode" (fn [self _n] nil) "connect" (fn [self] nil) - "disconnect" (fn [self] nil) + ;; Gives up a body still on the wire: its connection is closed rather + ;; than pooled, since what is left of the response is unread. + "disconnect" (fn [self] + (when (tget self :performed) + (try (.close ^java.io.InputStream (body-stream self)) + (catch Throwable _ nil))) + nil) "getOutputStream" (fn [self] (when (nil? (tget self :out-buffer)) (tput! self :out-buffer (make-baos))) (tget self :out-buffer)) @@ -280,14 +349,10 @@ "getHeaderField" (fn [self i] (let [pairs (:header-pairs (ensure-performed! self))] (when (and (>= i 1) (<= i (count pairs))) (second (nth pairs (dec i)))))) - "getInputStream" (fn [self] - (let [resp (ensure-performed! self)] - (if (>= (:status resp) 400) - (throw-typed "java.io.IOException" (str "Server returned HTTP response code: " (:status resp))) - (make-bais (:body resp))))) + "getInputStream" (fn [self] (input-stream self)) "getErrorStream" (fn [self] - (let [resp (ensure-performed! self)] - (when (>= (:status resp) 400) (make-bais (:body resp))))) + (when (>= (:status (ensure-performed! self)) 400) + (body-stream self))) "getContentLength" (fn [self] (or (parse-long (or (header-ci (:header-pairs (ensure-performed! self)) "content-length") "")) -1)) "setHostnameVerifier" (fn [self v] (tput! self :hostname-verifier v) nil) diff --git a/test/jolt/http/core_test.clj b/test/jolt/http/core_test.clj index aa1249b..7af6293 100644 --- a/test/jolt/http/core_test.clj +++ b/test/jolt/http/core_test.clj @@ -275,6 +275,19 @@ (is (= "bye" (:body (get! "/close")))) (is (zero? (core/pool-count))))))) +(deftest a-body-closed-before-its-end-gives-up-the-connection + (with-pool-server + (fn [_s] + (let [conn (.openConnection (java.net.URL. (str *pool-base* "/chunked"))) + body (.getInputStream conn)] + (is (= 200 (.getResponseCode conn))) + (is (pos? (.read body (byte-array 3) 0 3))) + (.close body) + (is (zero? (core/pool-count)) + "the rest of that response is still on the socket, so it is not pooled")) + (is (= "ok /next" (:body (get! "/next")))) + (is (= 1 (core/pool-count)))))) + (defn- hangup-handler "Answers the first request on a connection and hangs up on the next one without replying — a peer retiring a socket exactly as the client reuses it, @@ -382,7 +395,10 @@ (testing "a first request with a generous timeout leaves a pooled connection" (let [c (.openConnection (java.net.URL. (str "http://localhost:" (:port slow) "/fast")))] (.setReadTimeout c 10000) - (is (= 200 (.getResponseCode c)))) + (is (= 200 (.getResponseCode c))) + ;; the body is read off the wire as the caller reads it, so the + ;; connection goes back to the pool when it has been read to its end + (is (= "ok" (slurp (.getInputStream c))))) (is (= 1 (core/pool-count)))) (testing "the next request's own, shorter timeout is the one that applies" (let [c (.openConnection (java.net.URL. (str "http://localhost:" (:port slow) "/slow")))] diff --git a/test/jolt/http/test_runner.clj b/test/jolt/http/test_runner.clj index e036b2d..e9d9e63 100644 --- a/test/jolt/http/test_runner.clj +++ b/test/jolt/http/test_runner.clj @@ -12,6 +12,7 @@ [jolt.http.net-platform-test] [jolt.http.stream-shim-test] [jolt.http.core-test] + [jolt.http.urlconnection-stream-test] [clj-http.lite.links-test] [clj-http.lite.client-test] [clj-http.lite.integration-test])) @@ -23,6 +24,7 @@ 'jolt.http.net-platform-test 'jolt.http.stream-shim-test 'jolt.http.core-test + 'jolt.http.urlconnection-stream-test 'clj-http.lite.links-test 'clj-http.lite.client-test 'clj-http.lite.integration-test)] diff --git a/test/jolt/http/test_server.clj b/test/jolt/http/test_server.clj index 02bccb5..7727b9d 100644 --- a/test/jolt/http/test_server.clj +++ b/test/jolt/http/test_server.clj @@ -15,6 +15,10 @@ (ffi/defcfn c-listen "listen" [:int :int] :int) (ffi/defcfn c-setsockopt "setsockopt" [:int :int :int :pointer :int] :int) (ffi/defcfn c-accept "accept" [:int :pointer :pointer] :int :blocking) +(ffi/defcfn c-shutdown "shutdown" [:int :int] :int) + +;; SHUT_RDWR is 2 on every platform this runs on +(def ^:private SHUT-RDWR 2) (def ^:private AF-INET 2) (def ^:private SOCK-STREAM 1) @@ -267,8 +271,7 @@ (if (:close resp) (swap! stats update :per-connection conj (inc n)) (recur (subs acc (+ he 4 (:content-length req))) (inc n))))))))))))) - (catch Throwable _ nil) - (finally (try (conn-close conn) (catch Throwable _ nil))))) + (catch Throwable _ nil))) (defn start-persistent "A server that keeps connections open. :stats holds {:requests n @@ -277,11 +280,17 @@ :chunked true to frame the body with Transfer-Encoding, or :close true to ask the client to hang up. - Accepted connections are tracked in :conns and closed by `stop`. Closing only + Accepted connections are tracked in :conns and shut down by `stop`. Closing only the listening socket is not enough for a server whose whole point is that connections outlive a request: Linux refuses to rebind the port while they are still established, even with SO_REUSEADDR, so the next server on that port - failed with \"bind() failed\"." + failed with \"bind() failed\". + + Each connection is CLOSED only by the thread serving it, and only after it is + out of :conns; `stop` shuts the live ones down, which wakes that thread. + Closing from `stop` as well closed every fd twice, and the second close landed + on whatever socket had been given that number in between — a client + connection in the next test, which then read EBADF or another socket's bytes." [port handler] (let [fd (listen-socket port) running? (atom true) @@ -295,7 +304,10 @@ (neg? raw) (when @running? (recur)) :else (do (swap! conns conj raw) (future (try (persistent-serve raw handler stats running?) - (finally (swap! conns disj raw)))) + (finally + (locking conns + (swap! conns disj raw) + (try (net/close raw) (catch Throwable _ nil)))))) (recur)))))) {:fd fd :port port :running running? :stats stats :conns conns})) @@ -331,20 +343,27 @@ (or handler default-handler) concurrent?)) {:fd fd :port port :running running?}))) +(defn- shutdown-conns! + "Shut down every connection a persistent server is still serving. Not close: + the serving thread closes its own fd once it sees the EOF (see + start-persistent), so each fd is closed exactly once." + [server] + (when-let [a (:conns server)] + (locking a + (doseq [c @a] (try (c-shutdown c SHUT-RDWR) (catch Throwable _ nil)))))) + (defn drop-connections! - "Close every accepted connection, leaving the listener up — a peer retiring + "Hang up every accepted connection, leaving the listener up — a peer retiring its idle keep-alive sockets." [server] - (doseq [c (when-let [a (:conns server)] @a)] - (try (net/close c) (catch Throwable _ nil))) + (shutdown-conns! server) nil) (defn stop [server] (reset! (:running server) false) ;; accepted connections first: a persistent server's outlive the request that ;; made them, and the port cannot be rebound while they are established - (doseq [c (when-let [a (:conns server)] @a)] - (try (net/close c) (catch Throwable _ nil))) + (shutdown-conns! server) (net/close (:fd server)) nil) diff --git a/test/jolt/http/tls_test.clj b/test/jolt/http/tls_test.clj index ef63ec4..e4616fc 100644 --- a/test/jolt/http/tls_test.clj +++ b/test/jolt/http/tls_test.clj @@ -143,18 +143,28 @@ ;; closed whatever fd number had been handed out since and freed the SSL twice. ;; Server streams also never freed their per-connection SSL_CTX, and a failed ;; server handshake leaked the SSL, the BIOs and the context. -(defn- counting [f] - (let [closes (atom []) ssl-frees (atom 0) ctx-frees (atom 0) - close0 jolt.http.net/close ssl0 tls/c-SSL-free ctx0 tls/c-SSL-CTX-free] - (with-redefs [jolt.http.net/close (fn [fd] (swap! closes conj fd) (close0 fd)) - tls/c-SSL-free (fn [p] (swap! ssl-frees inc) (ssl0 p)) - tls/c-SSL-CTX-free (fn [p] (swap! ctx-frees inc) (ctx0 p))] - (f)) - {:closes @closes :ssl-frees @ssl-frees :ctx-frees @ctx-frees})) +(defn- counting + "Runs `f` and counts socket closes and SSL / SSL_CTX frees. Given `st`, only + that stream's own socket, SSL and context count: with-redefs is global, and + the in-process test server closes and frees ITS end of the same connection + on its own thread, inside this window on Linux, which counted as a second + close (nightly CI). Without `st`, every call counts." + ([f] (counting nil f)) + ([st f] + (let [closes (atom []) ssls (atom []) ctxs (atom []) + close0 jolt.http.net/close ssl0 tls/c-SSL-free ctx0 tls/c-SSL-CTX-free + own (fn [k] (if st (let [v (jolt.host/ref-get st k)] #(= v %)) (constantly true)))] + (with-redefs [jolt.http.net/close (fn [fd] (swap! closes conj fd) (close0 fd)) + tls/c-SSL-free (fn [p] (swap! ssls conj p) (ssl0 p)) + tls/c-SSL-CTX-free (fn [p] (swap! ctxs conj p) (ctx0 p))] + (f)) + {:closes (filterv (own :sock) @closes) + :ssl-frees (count (filter (own :ssl) @ssls)) + :ctx-frees (count (filter (own :ctx) @ctxs))}))) (deftest closing-a-tls-stream-twice-releases-once (let [st (tls/tls-connect "localhost" https-port true) - n (counting (fn [] ((jolt.host/ref-get st :close)) ((jolt.host/ref-get st :close))))] + n (counting st (fn [] ((jolt.host/ref-get st :close)) ((jolt.host/ref-get st :close))))] (is (= 1 (count (:closes n))) "the socket is closed once") (is (= 1 (:ssl-frees n)) "the SSL is freed once") (is (= 0 (:ctx-frees n)) "a client context is shared out of the cache, never freed"))) @@ -165,7 +175,7 @@ (try (let [client (tls/tls-connect "localhost" 18104 true) st @server - n (counting (fn [] ((jolt.host/ref-get st :close)) ((jolt.host/ref-get st :close))))] + n (counting st (fn [] ((jolt.host/ref-get st :close)) ((jolt.host/ref-get st :close))))] ((jolt.host/ref-get client :close)) (is (= 1 (count (:closes n)))) (is (= 1 (:ssl-frees n))) diff --git a/test/jolt/http/urlconnection_stream_test.clj b/test/jolt/http/urlconnection_stream_test.clj new file mode 100644 index 0000000..85c5a7a --- /dev/null +++ b/test/jolt/http/urlconnection_stream_test.clj @@ -0,0 +1,204 @@ +(ns jolt.http.urlconnection-stream-test + "`:as :stream` through clj-http-lite, i.e. over the HttpURLConnection shim in + jolt.http.platform. The java.net.http path has streamed its body off the wire + since b221fbf; this one read the whole body before getResponseCode returned, + so an SSE feed or an LLM token stream arrived in one piece at the end. + + The routes write their headers at once and the body a tick at a time, so a + client that buffers is visible as a call that takes the whole run." + (:require [clojure.test :refer [deftest is testing use-fixtures]] + [clojure.java.io :as io] + [clj-http.lite.util :as util] + [jolt.http-client :as http] + [jolt.http.core :as core] + [jolt.http.test-server :as srv])) + +(def ^:private port 18110) +(def ^:private base (str "http://localhost:" port)) + +(def ^:private tick-count 6) +(def ^:private tick-ms 150) +(def ^:private ticks-ms (* tick-count tick-ms)) +(def ^:private ticks (mapv (fn [i] (str "tick " i "\n")) (range tick-count))) +(def ^:private ticks-body (apply str ticks)) +;; how long a route holds the socket open after a framed body has ended +(def ^:private hold-ms 1500) + +;; Whether this jolt can carry a live body (a reify java.io.InputStream, jolt +;; 0.8.8+). Below that the transport keeps the buffered path, and the timing +;; assertions here do not apply. +(def ^:private live? core/reify-input-streams?) + +(defn- send! [conn s] (srv/conn-write conn (core/latin1->ba s))) + +(defn- head! [conn status & headers] + (send! conn (apply str "HTTP/1.1 " status " X\r\n" + (concat (map (fn [h] (str h "\r\n")) headers) ["\r\n"])))) + +(defn- chunked-ticks! [conn] + (doseq [t ticks] + (send! conn (str (Integer/toHexString (count t)) "\r\n" t "\r\n")) + (Thread/sleep tick-ms)) + (send! conn "0\r\n\r\n")) + +(defn- handler [{:keys [uri conn]}] + (case uri + "/chunked" + (do (head! conn 200 "Content-Type: text/event-stream" "Transfer-Encoding: chunked" + "Connection: close") + (chunked-ticks! conn) + (Thread/sleep hold-ms) + (srv/conn-close conn) + :hijacked) + + "/sized" + (do (head! conn 200 (str "Content-Length: " (count ticks-body)) "Connection: close") + (doseq [t ticks] (send! conn t) (Thread/sleep tick-ms)) + (Thread/sleep hold-ms) + (srv/conn-close conn) + :hijacked) + + "/to-close" + (do (head! conn 200 "Content-Type: text/plain") + (doseq [t ticks] (send! conn t) (Thread/sleep tick-ms)) + (srv/conn-close conn) + :hijacked) + + "/short" + (do (head! conn 200 "Content-Length: 100" "Connection: close") + (send! conn "hello") + (Thread/sleep 50) + (srv/conn-close conn) + :hijacked) + + "/stall" + (do (head! conn 200 "Transfer-Encoding: chunked" "Connection: close") + (send! conn "7\r\ntick 0\n\r\n") + (Thread/sleep 2000) + (send! conn "0\r\n\r\n") + (srv/conn-close conn) + :hijacked) + + "/error-ticks" + (do (head! conn 500 "Transfer-Encoding: chunked" "Connection: close") + (chunked-ticks! conn) + (srv/conn-close conn) + :hijacked) + + "/gzip" + {:status 200 + :headers {"Content-Encoding" "gzip"} + :body (util/gzip (.getBytes ^String ticks-body "UTF-8"))} + + {:status 404 :body "nope"})) + +;; One server for the whole namespace. stop closes the listener, but on Linux a +;; thread parked in accept() on it keeps the socket listening, so a second +;; server on the same port fails to bind (seen on CI). Each test runs its body +;; as a thunk against the one server. +(use-fixtures :once + (fn [t] + (let [s (srv/start-plain port {:handler handler})] + (Thread/sleep 100) + (try (t) (finally (srv/stop s)))))) + +(defn- with-server [f] (f)) + +(defn- elapsed [t0] (- (System/currentTimeMillis) t0)) + +(deftest stream-returns-at-the-headers + (with-server + (fn [] + (doseq [path ["/chunked" "/sized" "/to-close"]] + (testing (str path ": the call returns before the body has arrived") + (let [t0 (System/currentTimeMillis) + r (http/get (str base path) {:as :stream}) + headers-at (elapsed t0)] + (is (= 200 (:status r))) + (when live? + (is (< headers-at (quot ticks-ms 2)) + (str "returned after " headers-at "ms; the body runs for " ticks-ms "ms"))) + (is (= ticks-body (slurp (:body r)))) + (when (not= path "/to-close") + (is (< (elapsed t0) (+ ticks-ms (quot hold-ms 2))) + "the stream ends where the body ends, not at the peer's close")))))))) + +(deftest stream-reads-return-as-bytes-arrive + (with-server + (fn [] + (doseq [path ["/chunked" "/sized" "/to-close"]] + (testing (str path ": each line comes off the wire as it is sent") + (let [r (http/get (str base path) {:as :stream}) + rdr (java.io.BufferedReader. (io/reader (:body r))) + t0 (System/currentTimeMillis) + first-line (.readLine rdr) + first-at (elapsed t0) + _ (dotimes [_ (- tick-count 2)] (.readLine rdr)) + last-line (.readLine rdr) + last-at (elapsed t0)] + (is (= "tick 0" first-line)) + (is (= (str "tick " (dec tick-count)) last-line)) + (when live? + (is (< first-at (- last-at (* 2 tick-ms))) + (str "first line at " first-at "ms, last at " last-at "ms"))) + (.close rdr))))))) + +(deftest a-body-that-ends-early-is-an-error + (with-server + (fn [] + (testing "Content-Length promised 100 bytes and the peer closed after 5" + (let [r (http/get (str base "/short") {:as :stream})] + (is (= 200 (:status r))) + (is (thrown? java.io.IOException (slurp (:body r)))))) + (testing "the buffered coercions see the same failure" + (is (thrown? java.io.IOException (http/get (str base "/short")))))))) + +(deftest buffered-coercions-still-read-the-whole-body + (with-server + (fn [] + (doseq [path ["/chunked" "/sized" "/to-close"]] + (testing path + (is (= ticks-body (:body (http/get (str base path))))) + (is (= ticks-body (String. ^bytes (:body (http/get (str base path) {:as :byte-array})) + "UTF-8"))))) + (testing "gzip is decoded after the body is read" + (is (= ticks-body (:body (http/get (str base "/gzip"))))))))) + +(deftest an-error-status-streams-too + (with-server + (fn [] + (testing "getErrorStream hands back the live body" + (let [t0 (System/currentTimeMillis) + r (http/get (str base "/error-ticks") {:as :stream :throw-exceptions false}) + headers-at (elapsed t0)] + (is (= 500 (:status r))) + (when live? + (is (< headers-at (quot ticks-ms 2)) (str "returned after " headers-at "ms"))) + (is (= ticks-body (slurp (:body r)))))) + (testing "and the buffered one reads it whole" + (let [r (http/get (str base "/error-ticks") {:throw-exceptions false})] + (is (= 500 (:status r))) + (is (= ticks-body (:body r)))))))) + +(deftest socket-timeout-bounds-the-gap-between-reads + (with-server + (fn [] + (let [t0 (System/currentTimeMillis) + r (http/get (str base "/stall") {:as :stream :socket-timeout 400}) + body (:body r) + buf (byte-array 64)] + (is (= 200 (:status r))) + (is (= "tick 0\n" (String. buf 0 (.read body buf 0 64) "UTF-8"))) + (is (thrown? java.net.SocketTimeoutException (.read body buf 0 64))) + (is (< (elapsed t0) 1800) "the stalled read gave up at the socket timeout"))))) + +(deftest closing-a-stream-early + (with-server + (fn [] + (let [r (http/get (str base "/chunked") {:as :stream}) + body (:body r) + buf (byte-array 32)] + (is (pos? (.read body buf 0 32))) + (.close body) + (is (= -1 (try (.read body buf 0 32) (catch java.io.IOException _ -1))) + "nothing more is read from a closed body")))))