Skip to content

Stream the HttpURLConnection body off the wire #37

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yogthos merged 4 commits into from
Oct 13, 2026
Merged

Stream the HttpURLConnection body off the wire #37

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a5789f6
Stream the HttpURLConnection body off the wire
yogthos Oct 13, 2026
cdd7f55
Start the stream test's server once per namespace
yogthos Oct 13, 2026
06c9f0a
Run CI on jolt's nightly build
yogthos Oct 13, 2026
7ea9bbd
Count only the stream's own closes and frees in the TLS release tests
yogthos Oct 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 15 additions & 16 deletions .github/workflows/tests.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,29 +1,28 @@
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:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
jolt: ['0.8.16']
jolt: ['nightly']
name: test (jolt ${{ matrix.jolt }})
env:
JOLT_VERSION: ${{ matrix.jolt }}
Expand Down
13 changes: 8 additions & 5 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
101 changes: 83 additions & 18 deletions src/jolt/http/platform.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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
Expand All @@ -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)
Expand All @@ -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 ---------------------------------------------------------------
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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))
Expand All @@ -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)
Expand Down
18 changes: 17 additions & 1 deletion test/jolt/http/core_test.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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")))]
Expand Down
2 changes: 2 additions & 0 deletions test/jolt/http/test_runner.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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]))
Expand All @@ -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)]
Expand Down
39 changes: 29 additions & 10 deletions test/jolt/http/test_server.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand All @@ -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}))

Expand Down Expand Up @@ -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)

Expand Down
Loading
Loading