Test proxy pool with both http and https

src/Connections.jl

@@ -451,7 +451,11 @@ function newconnection(wrapconnection::Function,
 
     connection_limit_warning(connection_limit)
 
-    key = (url.host, url.port, proxy, require_ssl_verification, keepalive, true)
+    key = if isnothing(proxy) || IOType == socket_type_tls
+        (url.host, url.port, proxy, require_ssl_verification, keepalive, true)
+    else
+        (URI(proxy).host, URI(proxy).port, nothing, require_ssl_verification, keepalive, true)
+    end
 
     acquire(
             getpool(pool, IOType),
@@ -461,6 +465,9 @@ function newconnection(wrapconnection::Function,
 
         if proxy !== nothing
             url = URI(proxy)
+            if IOType != socket_type_tls
+                proxy = nothing
+            end
         end
 
         innerIOType = sockettype(url, socket_type, socket_type_tls)
@@ -479,7 +486,7 @@ function newconnection(wrapconnection::Function,
         end
 
         if connectionkey(io) != key
-            throw(ErrorException(string("Connection error ", (;expected = connectionkey(io), key))))
+            throw(ErrorException(string("Connection error ", (;expected = key, got = connectionkey(io)))))
         end
 
         io

test/client.jl

@@ -553,52 +553,116 @@ end
     end
 end
 
-@testset "HTTP CONNECT Proxy pool" begin
-    # Stores the http request passed by the client
-    messages = []
-    streams = Set()
-
-    function forwardstream(src, dst)
-        while isopen(dst) && isopen(src) && !eof(src)
-            buff = readavailable(src)
-            !isempty(buff) && isopen(dst) && write(dst, buff)
-        end
+function forwardstream(src, dst)
+    while isopen(dst) && isopen(src) && !eof(src)
+        buff = readavailable(src)
+        !isempty(buff) && isopen(dst) && write(dst, buff)
+    end
+end
 
-        close(src)
-        close(dst)
+@testset "HTTP CONNECT Proxy pool" begin
+    function forwardclosetask(src, dst)
+        errormonitor(@async begin
+            forwardstream(src, dst)
+            close(src)
+            close(dst)
+        end)
     end
 
-    # Simple implementation of a proxy server
+    # Stores the http message passed by the client
+    messages = []
+    upstreams = Set()
+
+    # Simple implementation of a https proxy server
     proxy = HTTP.listen!(IPv4(0), 8082; stream = true) do http::HTTP.Stream
         push!(messages, http.message)
-        host, port = split(http.message.target, ":")
-        targetstream = connect(host, parse(Int, port))
-        HTTP.setstatus(http, 200)
-        HTTP.startwrite(http)
-        up = @async forwardstream(http.stream.io, targetstream)
-        down = @async forwardstream(targetstream, http.stream.io)
-        push!(streams, targetstream)
-        wait(up)
-        wait(down)
-        delete!(streams, targetstream)
+
+        hostport = split(http.message.target, ":")
+        targetstream = connect(hostport[1], parse(Int, get(hostport, 2, "443")))
+        push!(upstreams, targetstream)
+        try
+            HTTP.setstatus(http, 200)
+            HTTP.startwrite(http)
+            up = forwardclosetask(http.stream.io, targetstream)
+            down = forwardclosetask(targetstream, http.stream.io)
+
+            wait(up)
+            wait(down)
+        finally
+            delete!(upstreams, targetstream)
+        end
     end
 
     try
         # Make the HTTP request
-        r1 = HTTP.get("https://example.com:443"; proxy="http://localhost:8082", retry=false, status_exception=true)
+        r1 = HTTP.get("https://$httpbin/ip"; proxy="http://localhost:8082", retry=false, status_exception=true)
         @test length(messages) == 1
         @test first(messages).method == "CONNECT"
-        @test length(streams) == 1 && isopen(first(streams)) # still alive
+        @test length(upstreams) == 1 && isopen(first(upstreams)) # still alive
 
         # Make another request
         # This should reuse the connection pool and not make another request to the proxy
         empty!(messages)
-        r2 = HTTP.get("https://example.com:443"; proxy="http://localhost:8082", retry=false, status_exception=true)
-        @test isempty(messages)
-        @test r1.body == r2.body # no new message to the proxy yet successfully get the same response from the target server
-        @test length(streams) == 1 && isopen(first(streams)) # still only one stream alive
+        r2 = HTTP.get("https://$httpbin/ip"; proxy="http://localhost:8082", retry=false, status_exception=true)
+        @test isempty(messages) # no new message to the proxy
+        @test length(upstreams) == 1 && isopen(first(upstreams)) # still only one stream alive
+    finally
+        close.(upstreams)
+        close(proxy)
+        HTTP.Connections.closeall()
+        wait(proxy)
+    end
+end
+
+@testset "HTTP Proxy pool" begin
+    # Stores the http request passed by the client
+    downstreamconnections = Set{HTTP.Connections.Connection}()
+    upstreamconnections = Set{HTTP.Connections.Connection}()
+    finished_request = Base.Event(true)
+
+    # Simple implementation of a http proxy server
+    proxy = HTTP.listen!(IPv4(0), 8082; stream = true) do http::HTTP.Stream
+        push!(downstreamconnections, http.stream)
+
+        HTTP.open(http.message.method, http.message.target, http.message.headers;
+                  decompress = false, version = http.message.version, retry=false,
+                  redirect = false) do targetstream
+            push!(upstreamconnections, targetstream.stream)
+
+            up = errormonitor(@async forwardstream(http, targetstream))
+            targetresponse = startread(targetstream)
+
+            HTTP.setstatus(http, targetresponse.status)
+            for h in targetresponse.headers
+                HTTP.setheader(http, h)
+            end
+
+            HTTP.startwrite(http)
+            down = errormonitor(@async forwardstream(targetstream, http))
+
+            wait(up)
+            wait(down)
+
+            notify(finished_request)
+        end
+    end
+
+    try
+        # Make the HTTP request
+        r1 = HTTP.get("http://$httpbin/ip"; proxy="http://localhost:8082", retry=false, redirect = false, status_exception=true)
+        wait(finished_request)
+        @test length(downstreamconnections) == 1
+        @test length(upstreamconnections) == 1
+
+        # Make another request
+        # This should reuse a connection pool in both the client and proxy
+        r2 = HTTP.get("http://$httpbin/ip"; proxy="http://localhost:8082", retry=false, redirect = false, status_exception=true)
+
+        # Check that notify was actually called, but that the set of connections remains of size 1
+        wait(finished_request)
+        @test length(downstreamconnections) == 1
+        @test length(upstreamconnections) == 1
     finally
-        close.(streams)
         close(proxy)
         HTTP.Connections.closeall()
         wait(proxy)