diff --git a/Project.toml b/Project.toml index 4fda915..25b6e82 100644 --- a/Project.toml +++ b/Project.toml @@ -4,7 +4,7 @@ keywords = ["Swagger", "OpenAPI", "REST"] license = "MIT" desc = "OpenAPI server and client helper for Julia" authors = ["JuliaHub Inc."] -version = "0.2.6" +version = "0.2.7" [deps] Base64 = "2a0f44e3-6c83-55bd-87e4-b1978d98bd5f" @@ -35,9 +35,10 @@ p7zip_jll = "17" Dates = "ade2ca70-3891-5945-98fb-dc099432e06a" HTTP = "cd3eb016-35fb-5094-929b-558a96fad6f3" Random = "9a3f8284-a2c9-5f02-9a11-845980a1fd5c" +Sockets = "6462fe0b-24de-5631-8697-dd941f90decc" Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40" TimeZones = "f269a46b-ccf7-5d73-abea-4c690281aa53" URIs = "5c2747f8-b7ea-4ff2-ba2e-563bfd36b1d4" [targets] -test = ["Test", "Random", "URIs", "HTTP", "Dates", "TimeZones"] +test = ["Test", "Random", "URIs", "HTTP", "Dates", "TimeZones", "Sockets"] diff --git a/src/client/httplibs/juliaweb_http.jl b/src/client/httplibs/juliaweb_http.jl index c1b4e4c..cadf0a0 100644 --- a/src/client/httplibs/juliaweb_http.jl +++ b/src/client/httplibs/juliaweb_http.jl @@ -317,13 +317,17 @@ function _http_streaming_request(ctx, method, url, headers, body, timeout, bytes write(io, body) captured_response[] = http_response = startread(io) try - # `read(io, n)` is unavailable on 2.0 streams; read into a reusable - # buffer with `readbytes!`, which works on both 1.x and 2.x. - buf = Vector{UInt8}(undef, 8192) # 8KB chunks + # `readavailable`, not `readbytes!`: on both 1.x and 2.x streams + # `readbytes!(io, buf)` blocks until the whole buffer is filled + # (or the body ends), so a response chunk smaller than the buffer + # (e.g. a single Kubernetes watch event) is not forwarded until + # enough later data accumulates — an indefinite stall on quiet + # streams. `eof` blocks until data is available; `readavailable` + # then returns whatever has arrived without further blocking. while !eof(io) - n = readbytes!(io, buf) - n == 0 && break - bytesread[] += write(output, view(buf, 1:n)) + bytes = readavailable(io) + isempty(bytes) && continue + bytesread[] += write(output, bytes) end finally close(output) diff --git a/test/runtests.jl b/test/runtests.jl index 49f661f..6ab75ad 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -3,6 +3,7 @@ using Test, HTTP import OpenAPI include("chunkreader_tests.jl") +include("streaming_latency_tests.jl") include("testutils.jl") include("modelgen/testmodelgen.jl") include("client/runtests.jl") @@ -19,6 +20,9 @@ include("deep_object/deep_client.jl") @testset "Chunk Readers" begin ChunkReaderTests.runtests() end + @testset "Streaming Latency" begin + StreamingLatencyTests.runtests() + end @testset "Petstore Client" begin try if run_tests_with_servers && !openapi_generator_env diff --git a/test/streaming_latency_tests.jl b/test/streaming_latency_tests.jl new file mode 100644 index 0000000..3d366e3 --- /dev/null +++ b/test/streaming_latency_tests.jl @@ -0,0 +1,110 @@ +module StreamingLatencyTests + +using Test +using Sockets + +import OpenAPI +import OpenAPI.Clients: Client, Ctx, exec, LineChunkReader + +# Streamed chunks must be forwarded to the consumer as soon as they arrive, +# not held back until an internal read buffer fills. Regression test for the +# :http backend stalling small chunks: `readbytes!(io, buf)` blocks until the +# whole 8KB buffer is filled, so a chunk smaller than that (e.g. a Kubernetes +# watch event) was only delivered once enough later data accumulated — on a +# quiet stream, effectively never. +# +# The server is a minimal raw-TCP chunked HTTP/1.1 responder so that the test +# controls exactly when bytes hit the wire, independent of HTTP.jl's server +# API (which differs between 1.x and 2.x). Two endpoints: +# /quick - both chunks back to back (used to warm up/compile the client path) +# /stall - first chunk immediately, second only after `delay` seconds + +# JSON string literals: in the streaming path the client parses each chunk as +# JSON (no response object to take a content-type from), so a quoted string +# round-trips to a `String` return value. +const CHUNK1 = "\"tick-1\"\n" +const CHUNK2 = "\"tick-2\"\n" + +function write_chunk(sock, data) + write(sock, string(sizeof(data), base=16), "\r\n", data, "\r\n") + flush(sock) +end + +function handle_connection(sock, delay) + try + request_line = readline(sock) + path = split(request_line, ' ')[2] + while true + line = readline(sock) + (isempty(line) || line == "\r") && break + end + # `connection: close` so the client does not try to reuse the socket, + # which this bare-bones server closes after each response + write(sock, "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\nconnection: close\r\ntransfer-encoding: chunked\r\n\r\n") + flush(sock) + write_chunk(sock, CHUNK1) + endswith(path, "/stall") && sleep(delay) + write_chunk(sock, CHUNK2) + write(sock, "0\r\n\r\n") + flush(sock) + catch + # client went away; nothing to do + finally + close(sock) + end +end + +function start_server(delay) + listener = Sockets.listen(Sockets.localhost, 0) + _, port = Sockets.getsockname(listener) + @async begin + while true + sock = try + accept(listener) + catch + break # listener closed + end + @async handle_connection(sock, delay) + end + end + listener, Int(port) +end + +# Run a streaming GET and return (first chunk, seconds until it arrived, +# remaining chunks). +function first_chunk_latency(httplib, port, path) + client = Client("http://127.0.0.1:$port"; httplib=httplib) + return_types = Dict{Regex,Type}(Regex("^200\$") => String) + ctx = Ctx(client, "GET", return_types, path, String[]; chunk_reader_type=LineChunkReader) + events = Channel{Any}(64) + t0 = time() + task = @async exec(ctx, events) + first = take!(events) + latency = time() - t0 + remaining = collect(events) + wait(task) + (first, latency, remaining) +end + +function runtests() + delay = 4.0 + listener, port = start_server(delay) + try + @testset "first chunk not delayed until buffer fills ($httplib)" for httplib in values(OpenAPI.Clients.HTTPLib) + # warm up: compile the full streaming request path so the timing + # below measures I/O behavior, not JIT + first_chunk_latency(httplib, port, "/quick") + + first, latency, remaining = first_chunk_latency(httplib, port, "/stall") + @test first == "tick-1" + # the first chunk must arrive while the server is still stalling + # before the second chunk, not when the response completes + @test latency < delay / 2 + @test remaining == ["tick-2"] + end + finally + close(listener) + end +end + +end # module StreamingLatencyTests