diff --git a/lib/async/http/protocol/http2/connection.rb b/lib/async/http/protocol/http2/connection.rb index 37142ff1..bd3206ff 100644 --- a/lib/async/http/protocol/http2/connection.rb +++ b/lib/async/http/protocol/http2/connection.rb @@ -47,12 +47,21 @@ def synchronize(&block) # Write a single frame, deferring stop until the frame is written. def write_frame(frame) - Task.current.defer_stop{super} + defer_stop {super} end # Write multiple frames, deferring stop until the frames are written. def write_frames - Task.current.defer_stop{super} + defer_stop {super} + end + + # A fiber may be running under the scheduler without being an Async task, e.g. an `Enumerator` reading a response body, in which case there is no task to defer stopping. + private def defer_stop(&block) + if task = Task.current? + task.defer_stop(&block) + else + yield + end end # @returns [String] A string representation of this connection. diff --git a/test/async/http/protocol/http2.rb b/test/async/http/protocol/http2.rb index be6d4cf9..84857a28 100644 --- a/test/async/http/protocol/http2.rb +++ b/test/async/http/protocol/http2.rb @@ -149,6 +149,47 @@ def make_client(endpoint, **options) end end + with "a fiber which is not an Async task" do + let(:app) do + Protocol::HTTP::Middleware.for do |request| + Protocol::HTTP::Response[200, {}, ["x" * 1_000_000] * 10] + end + end + + it "can read a response body larger than the window" do + response = client.get("/") + + # An Enumerator iterates in its own fiber, which has the scheduler but no task, e.g. Falcon reading a Rack body: + chunks = Enumerator.new do |yielder| + response.body.each{|chunk| yielder << chunk} + end + + size = 0 + + loop do + size += chunks.next.bytesize + end + + expect(size).to be == 10_000_000 + ensure + response&.close + end + + it "can cancel a response before its body is read" do + response = client.get("/") + stream = response.stream + + # Closing the response sends RST_STREAM, which is also a write from the non-task fiber: + cancel = Enumerator.new do |yielder| + response.close + yielder << stream.closed? + end + + expect(cancel.next).to be == true + expect{cancel.next}.to raise_exception(StopIteration) + end + end + with "stopping requests" do let(:finished) {Async::Promise.new}