From 279b0133a025614bd313817295fc2ced9ebfd712 Mon Sep 17 00:00:00 2001
From: Samuel Williams <samuel.williams@oriontransfer.co.nz>
Date: Mon, 22 Jan 2024 16:47:54 +1300
Subject: [PATCH] WIP

---
 fixtures/async/http/a_protocol.rb           | 38 +++++++++++++--------
 lib/async/http/body/pipe.rb                 |  2 +-
 lib/async/http/client.rb                    |  5 +--
 lib/async/http/protocol/http1/client.rb     |  2 +-
 lib/async/http/protocol/http1/connection.rb |  2 +-
 lib/async/http/protocol/http2/connection.rb |  4 +--
 test/async/http/proxy.rb                    |  6 ++--
 7 files changed, 34 insertions(+), 25 deletions(-)

diff --git a/fixtures/async/http/a_protocol.rb b/fixtures/async/http/a_protocol.rb
index 7d4c8e59..1910e82c 100644
--- a/fixtures/async/http/a_protocol.rb
+++ b/fixtures/async/http/a_protocol.rb
@@ -544,33 +544,41 @@ def around
 					Console.logger.level = current
 				end
 				
+				def timeout = nil
+				
 				it "doesn't cancel all requests" do
-					tasks = []
 					task = Async::Task.current
+					tasks = []
 					stopped = []
 					
 					10.times do
-						tasks << task.async {
-							begin
-								loop do
-									client.get('http://127.0.0.1:8080/a').finish
-								end
+						task.async do |child|
+							tasks << child
+							
+							loop do
+								response = client.get('/a')
+								response.finish
 							ensure
-								stopped << 'a'
+								response&.close
 							end
-						}
+						ensure
+							stopped << 'a'
+						end
 					end
 					
 					10.times do
-						tasks << task.async {
-							begin
-								loop do
-									client.get('http://127.0.0.1:8080/b').finish
-								end
+						task.async do |child|
+							tasks << child
+							
+							loop do
+								response = client.get('/b')
+								response.finish
 							ensure
-								stopped << 'b'
+								response&.close
 							end
-						}
+						ensure
+							stopped << 'b'
+						end
 					end
 					
 					tasks.each do |child|
diff --git a/lib/async/http/body/pipe.rb b/lib/async/http/body/pipe.rb
index da57c3c8..6ce07315 100644
--- a/lib/async/http/body/pipe.rb
+++ b/lib/async/http/body/pipe.rb
@@ -65,7 +65,7 @@ def writer(task)
 					
 					task.annotate "#{self.class} writer."
 					
-					while chunk = @head.read_partial
+					while chunk = @head.readpartial(1024)
 						@output.write(chunk)
 					end
 				ensure
diff --git a/lib/async/http/client.rb b/lib/async/http/client.rb
index 2c571588..0c3cf87b 100755
--- a/lib/async/http/client.rb
+++ b/lib/async/http/client.rb
@@ -93,7 +93,6 @@ def call(request)
 					
 					# This signals that the ensure block below should not try to release the connection, because it's bound into the response which will be returned:
 					connection = nil
-					
 					return response
 				rescue Protocol::RequestFailed
 					# This is a specific case where the entire request wasn't sent before a failure occurred. So, we can even resend non-idempotent requests.
@@ -119,7 +118,9 @@ def call(request)
 						raise
 					end
 				ensure
-					@pool.release(connection) if connection
+					if connection
+						@pool.release(connection)
+					end
 				end
 			end
 			
diff --git a/lib/async/http/protocol/http1/client.rb b/lib/async/http/protocol/http1/client.rb
index 0139d203..68dad790 100644
--- a/lib/async/http/protocol/http1/client.rb
+++ b/lib/async/http/protocol/http1/client.rb
@@ -64,7 +64,7 @@ def call(request, task: Task.current)
 						@ready = true
 						
 						return response
-					rescue
+					rescue => error
 						# This will ensure that #reusable? returns false.
 						@stream.close
 						
diff --git a/lib/async/http/protocol/http1/connection.rb b/lib/async/http/protocol/http1/connection.rb
index ea943a29..d1b02d3e 100755
--- a/lib/async/http/protocol/http1/connection.rb
+++ b/lib/async/http/protocol/http1/connection.rb
@@ -41,7 +41,7 @@ def read_line
 					end
 					
 					def peer
-						@stream.io
+						@stream
 					end
 					
 					attr :count
diff --git a/lib/async/http/protocol/http2/connection.rb b/lib/async/http/protocol/http2/connection.rb
index 1829a8ac..07c68616 100644
--- a/lib/async/http/protocol/http2/connection.rb
+++ b/lib/async/http/protocol/http2/connection.rb
@@ -97,7 +97,7 @@ def read_in_background(parent: Task.current)
 									Console.debug(self, @framer) {"Reading frame..."}
 									self.read_frame
 								end
-							rescue SocketError, IOError, EOFError, Errno::ECONNRESET, Errno::EPIPE, Async::Wrapper::Cancelled => error
+							rescue SocketError, EOFError, Errno::ECONNRESET, Errno::EPIPE, Async::Wrapper::Cancelled => error
 								# Ignore.
 								Console.warn(self, error)
 							rescue ::Protocol::HTTP2::GoawayError => error
@@ -119,7 +119,7 @@ def read_in_background(parent: Task.current)
 					attr :promises
 					
 					def peer
-						@stream.io
+						@stream
 					end
 					
 					attr :count
diff --git a/test/async/http/proxy.rb b/test/async/http/proxy.rb
index ef677cc9..10e0ba5e 100644
--- a/test/async/http/proxy.rb
+++ b/test/async/http/proxy.rb
@@ -69,7 +69,7 @@
 				expect(request.path).to be == "localhost:1"
 				
 				Async::HTTP::Body::Hijack.response(request, 200, {}) do |stream|
-					while chunk = stream.read_partial(1024)
+					while chunk = stream.readpartial(1024)
 						stream.write(chunk)
 						stream.flush
 					end
@@ -135,7 +135,7 @@
 					reader = Async do |task|
 						task.annotate "Upstream reader."
 						
-						while chunk = upstream.read_partial
+						while chunk = upstream.readpartial(1024)
 							stream.write(chunk)
 							stream.flush
 						end
@@ -147,7 +147,7 @@
 					writer = Async do |task|
 						task.annotate "Upstream writer."
 						
-						while chunk = stream.read_partial
+						while chunk = stream.readpartial(1024)
 							upstream.write(chunk)
 							upstream.flush
 						end