Skip to content

Commit 2f3180a

Browse files
ioquatixioquatix
authored
Streaming tests. (#182)
* Wait for input to be consumed before continuing. * Update dependencies.
1 parent 1c4b5ab commit 2f3180a

File tree

7 files changed

+333
-5
lines changed

7 files changed

+333
-5
lines changed

‎async-http.gemspec

+1-1
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ Gem::Specification.new do |spec|
2828
spec.add_dependency "async-pool", "~> 0.7"
2929
spec.add_dependency "io-endpoint", "~> 0.11"
3030
spec.add_dependency "io-stream", "~> 0.4"
31-
spec.add_dependency "protocol-http", "~> 0.34"
31+
spec.add_dependency "protocol-http", "~> 0.35"
3232
spec.add_dependency "protocol-http1", "~> 0.20"
3333
spec.add_dependency "protocol-http2", "~> 0.18"
3434
spec.add_dependency "traces", ">= 0.10"

‎lib/async/http/body/finishable.rb

+39
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
# frozen_string_literal: true
2+
3+
# Released under the MIT License.
4+
# Copyright, 2019-2023, by Samuel Williams.
5+
6+
require 'protocol/http/body/wrapper'
7+
require 'async/variable'
8+
9+
module Async
10+
module HTTP
11+
module Body
12+
class Finishable < ::Protocol::HTTP::Body::Wrapper
13+
def initialize(body)
14+
super(body)
15+
16+
@closed = Async::Variable.new
17+
@error = nil
18+
end
19+
20+
def close(error = nil)
21+
unless @closed.resolved?
22+
@error = error
23+
@closed.value = true
24+
end
25+
26+
super
27+
end
28+
29+
def wait
30+
@closed.wait
31+
end
32+
33+
def inspect
34+
"#<#{self.class} closed=#{@closed} error=#{@error}> | #{super}"
35+
end
36+
end
37+
end
38+
end
39+
end

‎lib/async/http/client.rb

+3
Original file line numberDiff line numberDiff line change
@@ -188,6 +188,9 @@ def make_response(request, connection)
188188

189189
# The connection won't be released until the body is completely read/released.
190190
::Protocol::HTTP::Body::Completable.wrap(response) do
191+
# TODO: We should probably wait until the request is fully consumed and/or the connection is ready before releasing it back into the pool.
192+
193+
# Release the connection back into the pool:
191194
@pool.release(connection)
192195
end
193196

‎lib/async/http/protocol/http1/server.rb

+12-4
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,8 @@
77
# Copyright, 2024, by Anton Zhuravsky.
88

99
require_relative 'connection'
10+
require_relative '../../body/finishable'
11+
1012
require 'console/event/failure'
1113

1214
module Async
@@ -46,6 +48,11 @@ def each(task: Task.current)
4648
task.annotate("Reading #{self.version} requests for #{self.class}.")
4749

4850
while request = next_request
51+
if body = request.body
52+
finishable = Body::Finishable.new(body)
53+
request.body = finishable
54+
end
55+
4956
response = yield(request, self)
5057
version = request.version
5158
body = response&.body
@@ -102,23 +109,24 @@ def each(task: Task.current)
102109
head = request.head?
103110

104111
# Same as above:
105-
request = nil unless request.body
112+
request = nil
106113
response = nil
107114

108115
write_body(version, body, head, trailer)
109116
end
110117
end
111118

112-
# We are done with the body, you shouldn't need to call close on it:
119+
# We are done with the body:
113120
body = nil
114121
else
115122
# If the request failed to generate a response, it was an internal server error:
116123
write_response(@version, 500, {})
117124
write_body(version, nil)
125+
126+
request&.finish
118127
end
119128

120-
# Gracefully finish reading the request body if it was not already done so.
121-
request&.each{}
129+
finishable&.wait
122130

123131
# This ensures we yield at least once every iteration of the loop and allow other fibers to execute.
124132
task.yield

‎test/async/http/middleware/location_redirector.rb

+2
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818
with '301' do
1919
let(:app) do
2020
Protocol::HTTP::Middleware.for do |request|
21+
request.finish # TODO: request.discard - or some default handling?
22+
2123
case request.path
2224
when '/home'
2325
Protocol::HTTP::Response[301, {'location' => '/'}, []]

‎test/protocol/http/body/stream.rb

+142
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,142 @@
1+
# frozen_string_literal: true
2+
3+
# Released under the MIT License.
4+
# Copyright, 2024, by Samuel Williams.
5+
6+
require "async/http/protocol/http"
7+
require "protocol/http/body/streamable"
8+
require "sus/fixtures/async/http"
9+
10+
AnEchoServer = Sus::Shared("an echo server") do
11+
let(:app) do
12+
::Protocol::HTTP::Middleware.for do |request|
13+
output = ::Protocol::HTTP::Body::Writable.new
14+
15+
Async do
16+
stream = ::Protocol::HTTP::Body::Stream.new(request.body, output)
17+
18+
Console.debug(self, "Echoing chunks...")
19+
while chunk = stream.readpartial(1024)
20+
Console.debug(self, "Reading chunk:", chunk: chunk)
21+
stream.write(chunk)
22+
end
23+
rescue EOFError
24+
Console.debug(self, "EOF.")
25+
# Ignore.
26+
ensure
27+
Console.debug(self, "Closing stream.")
28+
stream.close
29+
end
30+
31+
::Protocol::HTTP::Response[200, {}, output]
32+
end
33+
end
34+
35+
it "should echo the request body" do
36+
chunks = ["Hello,", "World!"]
37+
response_chunks = Queue.new
38+
39+
output = ::Protocol::HTTP::Body::Writable.new
40+
response = client.post("/", body: output)
41+
stream = ::Protocol::HTTP::Body::Stream.new(response.body, output)
42+
43+
begin
44+
Console.debug(self, "Echoing chunks...")
45+
chunks.each do |chunk|
46+
Console.debug(self, "Writing chunk:", chunk: chunk)
47+
stream.write(chunk)
48+
end
49+
50+
Console.debug(self, "Closing write.")
51+
stream.close_write
52+
53+
Console.debug(self, "Reading chunks...")
54+
while chunk = stream.readpartial(1024)
55+
Console.debug(self, "Reading chunk:", chunk: chunk)
56+
response_chunks << chunk
57+
end
58+
rescue EOFError
59+
Console.debug(self, "EOF.")
60+
# Ignore.
61+
ensure
62+
Console.debug(self, "Closing stream.")
63+
stream.close
64+
response_chunks.close
65+
end
66+
67+
chunks.each do |chunk|
68+
expect(response_chunks.pop).to be == chunk
69+
end
70+
end
71+
end
72+
73+
AnEchoClient = Sus::Shared("an echo client") do
74+
let(:chunks) {["Hello,", "World!"]}
75+
let(:response_chunks) {Queue.new}
76+
77+
let(:app) do
78+
::Protocol::HTTP::Middleware.for do |request|
79+
output = ::Protocol::HTTP::Body::Writable.new
80+
81+
Async do
82+
stream = ::Protocol::HTTP::Body::Stream.new(request.body, output)
83+
84+
Console.debug(self, "Echoing chunks...")
85+
chunks.each do |chunk|
86+
stream.write(chunk)
87+
end
88+
89+
Console.debug(self, "Closing write.")
90+
stream.close_write
91+
92+
Console.debug(self, "Reading chunks...")
93+
while chunk = stream.readpartial(1024)
94+
Console.debug(self, "Reading chunk:", chunk: chunk)
95+
response_chunks << chunk
96+
end
97+
rescue EOFError
98+
Console.debug(self, "EOF.")
99+
# Ignore.
100+
ensure
101+
Console.debug(self, "Closing stream.")
102+
stream.close
103+
end
104+
105+
::Protocol::HTTP::Response[200, {}, output]
106+
end
107+
end
108+
109+
it "should echo the response body" do
110+
output = ::Protocol::HTTP::Body::Writable.new
111+
response = client.post("/", body: output)
112+
stream = ::Protocol::HTTP::Body::Stream.new(response.body, output)
113+
114+
begin
115+
Console.debug(self, "Echoing chunks...")
116+
while chunk = stream.readpartial(1024)
117+
stream.write(chunk)
118+
end
119+
rescue EOFError
120+
Console.debug(self, "EOF.")
121+
# Ignore.
122+
ensure
123+
Console.debug(self, "Closing stream.")
124+
stream.close
125+
end
126+
127+
chunks.each do |chunk|
128+
expect(response_chunks.pop).to be == chunk
129+
end
130+
end
131+
end
132+
133+
[Async::HTTP::Protocol::HTTP1, Async::HTTP::Protocol::HTTP2].each do |protocol|
134+
describe protocol, unique: protocol.name do
135+
include Sus::Fixtures::Async::HTTP::ServerContext
136+
137+
let(:protocol) {subject}
138+
139+
it_behaves_like AnEchoServer
140+
it_behaves_like AnEchoClient
141+
end
142+
end

‎test/protocol/http/body/streamable.rb

+134
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
# frozen_string_literal: true
2+
3+
# Released under the MIT License.
4+
# Copyright, 2024, by Samuel Williams.
5+
6+
require "async/http/protocol/http"
7+
require "protocol/http/body/streamable"
8+
require "sus/fixtures/async/http"
9+
10+
AnEchoServer = Sus::Shared("an echo server") do
11+
let(:app) do
12+
::Protocol::HTTP::Middleware.for do |request|
13+
streamable = ::Protocol::HTTP::Body::Streamable.response(request) do |stream|
14+
Console.debug(self, "Echoing chunks...")
15+
while chunk = stream.readpartial(1024)
16+
Console.debug(self, "Reading chunk:", chunk: chunk)
17+
stream.write(chunk)
18+
end
19+
rescue EOFError
20+
Console.debug(self, "EOF.")
21+
# Ignore.
22+
ensure
23+
Console.debug(self, "Closing stream.")
24+
stream.close
25+
end
26+
27+
::Protocol::HTTP::Response[200, {}, streamable]
28+
end
29+
end
30+
31+
it "should echo the request body" do
32+
chunks = ["Hello,", "World!"]
33+
response_chunks = Queue.new
34+
35+
output = ::Protocol::HTTP::Body::Writable.new
36+
response = client.post("/", body: output)
37+
stream = ::Protocol::HTTP::Body::Stream.new(response.body, output)
38+
39+
begin
40+
Console.debug(self, "Echoing chunks...")
41+
chunks.each do |chunk|
42+
Console.debug(self, "Writing chunk:", chunk: chunk)
43+
stream.write(chunk)
44+
end
45+
46+
Console.debug(self, "Closing write.")
47+
stream.close_write
48+
49+
Console.debug(self, "Reading chunks...")
50+
while chunk = stream.readpartial(1024)
51+
Console.debug(self, "Reading chunk:", chunk: chunk)
52+
response_chunks << chunk
53+
end
54+
rescue EOFError
55+
Console.debug(self, "EOF.")
56+
# Ignore.
57+
ensure
58+
Console.debug(self, "Closing stream.")
59+
stream.close
60+
response_chunks.close
61+
end
62+
63+
chunks.each do |chunk|
64+
expect(response_chunks.pop).to be == chunk
65+
end
66+
end
67+
end
68+
69+
AnEchoClient = Sus::Shared("an echo client") do
70+
let(:chunks) {["Hello,", "World!"]}
71+
let(:response_chunks) {Queue.new}
72+
73+
let(:app) do
74+
::Protocol::HTTP::Middleware.for do |request|
75+
streamable = ::Protocol::HTTP::Body::Streamable.response(request) do |stream|
76+
Console.debug(self, "Echoing chunks...")
77+
chunks.each do |chunk|
78+
stream.write(chunk)
79+
end
80+
81+
Console.debug(self, "Closing write.")
82+
stream.close_write
83+
84+
Console.debug(self, "Reading chunks...")
85+
while chunk = stream.readpartial(1024)
86+
Console.debug(self, "Reading chunk:", chunk: chunk)
87+
response_chunks << chunk
88+
end
89+
rescue EOFError
90+
Console.debug(self, "EOF.")
91+
# Ignore.
92+
ensure
93+
Console.debug(self, "Closing stream.")
94+
stream.close
95+
end
96+
97+
::Protocol::HTTP::Response[200, {}, streamable]
98+
end
99+
end
100+
101+
it "should echo the response body" do
102+
output = ::Protocol::HTTP::Body::Writable.new
103+
response = client.post("/", body: output)
104+
stream = ::Protocol::HTTP::Body::Stream.new(response.body, output)
105+
106+
begin
107+
Console.debug(self, "Echoing chunks...")
108+
while chunk = stream.readpartial(1024)
109+
stream.write(chunk)
110+
end
111+
rescue EOFError
112+
Console.debug(self, "EOF.")
113+
# Ignore.
114+
ensure
115+
Console.debug(self, "Closing stream.")
116+
stream.close
117+
end
118+
119+
chunks.each do |chunk|
120+
expect(response_chunks.pop).to be == chunk
121+
end
122+
end
123+
end
124+
125+
[Async::HTTP::Protocol::HTTP1, Async::HTTP::Protocol::HTTP2].each do |protocol|
126+
describe protocol, unique: protocol.name do
127+
include Sus::Fixtures::Async::HTTP::ServerContext
128+
129+
let(:protocol) {subject}
130+
131+
it_behaves_like AnEchoServer
132+
it_behaves_like AnEchoClient
133+
end
134+
end

0 commit comments

Comments
 (0)