Add or fix events inside or related to CONNECT tunnels

3 files changed, 1085 insertions, 67 deletions

diff --git a/test/event_SUITE.erl b/test/event_SUITE.erl
index 5510e8c..6f2f1fa 100644
--- a/test/event_SUITE.erl
+++ b/test/event_SUITE.erl
@@ -37,7 +37,7 @@ groups() ->
 	%% We currently do not support Websocket over HTTP/2.
 	WsTests = [T || T <- Tests, lists:sublist(atom_to_list(T), 3) =:= "ws_"],
 	[
-		{http, [parallel], Tests -- [cancel_remote|PushTests]},
+		{http, [parallel], Tests -- [cancel_remote, cancel_remote_connect|PushTests]},
 		{http2, [parallel], (Tests -- WsTests) -- HTTP1Tests}
 	].
 
@@ -193,6 +193,8 @@ connect_end_ok_tls(Config) ->
 	false = maps:is_key(protocol, Event),
 	gun:close(Pid).
 
+%% tls_handshake_start/tls_handshake_end.
+
 tls_handshake_start(Config) ->
 	doc("Confirm that the tls_handshake_start event callback is called."),
 	{ok, Pid, _} = do_gun_open_tls(Config),
@@ -236,17 +238,19 @@ tls_handshake_end_ok(Config) ->
 	true = is_tuple(Socket),
 	gun:close(Pid).
 
-http1_tls_handshake_start_connect(Config) ->
+tls_handshake_start_tcp_connect_tls(Config) ->
 	doc("Confirm that the tls_handshake_start event callback is called "
 		"when using CONNECT to a TLS server via a TCP proxy."),
 	OriginPort = config(tls_origin_port, Config),
-	{ok, _, ProxyPort} = rfc7231_SUITE:do_proxy_start(tcp),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
 	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
 		event_handler => {?MODULE, self()},
-		protocols => [config(name, config(tc_group_properties, Config))],
+		protocols => [Protocol],
 		transport => tcp
 	}),
-	{ok, http} = gun:await_up(ConnPid),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
 	StreamRef = gun:connect(ConnPid, #{
 		host => "localhost",
 		port => OriginPort,
@@ -260,21 +264,26 @@ http1_tls_handshake_start_connect(Config) ->
 		tls_opts := _,
 		timeout := _
 	} = do_receive_event(tls_handshake_start),
-	true = is_port(Socket),
+	true = case Protocol of
+		http -> is_port(Socket);
+		http2 -> is_map(Socket)
+	end,
 	gun:close(ConnPid).
 
-http1_tls_handshake_end_error_connect(Config) ->
+tls_handshake_end_error_tcp_connect_tls(Config) ->
 	doc("Confirm that the tls_handshake_end event callback is called on TLS handshake error "
 		"when using CONNECT to a TLS server via a TCP proxy."),
 	%% We use the wrong port on purpose to trigger a handshake error.
 	OriginPort = config(tcp_origin_port, Config),
-	{ok, _, ProxyPort} = rfc7231_SUITE:do_proxy_start(tcp),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
 	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
 		event_handler => {?MODULE, self()},
-		protocols => [config(name, config(tc_group_properties, Config))],
+		protocols => [Protocol],
 		transport => tcp
 	}),
-	{ok, http} = gun:await_up(ConnPid),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
 	StreamRef = gun:connect(ConnPid, #{
 		host => "localhost",
 		port => OriginPort,
@@ -289,20 +298,25 @@ http1_tls_handshake_end_error_connect(Config) ->
 		timeout := _,
 		error := {tls_alert, _}
 	} = do_receive_event(tls_handshake_end),
-	true = is_port(Socket),
+	true = case Protocol of
+		http -> is_port(Socket);
+		http2 -> is_map(Socket)
+	end,
 	gun:close(ConnPid).
 
-http1_tls_handshake_end_ok_connect(Config) ->
+tls_handshake_end_ok_tcp_connect_tls(Config) ->
 	doc("Confirm that the tls_handshake_end event callback is called on TLS handshake success "
 		"when using CONNECT to a TLS server via a TCP proxy."),
 	OriginPort = config(tls_origin_port, Config),
-	{ok, _, ProxyPort} = rfc7231_SUITE:do_proxy_start(tcp),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
 	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
 		event_handler => {?MODULE, self()},
-		protocols => [config(name, config(tc_group_properties, Config))],
+		protocols => [Protocol],
 		transport => tcp
 	}),
-	{ok, http} = gun:await_up(ConnPid),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
 	StreamRef = gun:connect(ConnPid, #{
 		host => "localhost",
 		port => OriginPort,
@@ -317,20 +331,25 @@ http1_tls_handshake_end_ok_connect(Config) ->
 		timeout := _,
 		protocol := http2
 	} = do_receive_event(tls_handshake_end),
-	true = is_tuple(Socket),
+	true = case Protocol of
+		http -> is_tuple(Socket);
+		http2 -> is_pid(Socket)
+	end,
 	gun:close(ConnPid).
 
-http1_tls_handshake_start_connect_over_https_proxy(Config) ->
+tls_handshake_start_tls_connect_tls(Config) ->
 	doc("Confirm that the tls_handshake_start event callback is called "
 		"when using CONNECT to a TLS server via a TLS proxy."),
 	OriginPort = config(tls_origin_port, Config),
-	{ok, _, ProxyPort} = rfc7231_SUITE:do_proxy_start(tls),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tls),
 	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
 		event_handler => {?MODULE, self()},
-		protocols => [config(name, config(tc_group_properties, Config))],
+		protocols => [Protocol],
 		transport => tls
 	}),
-	{ok, http} = gun:await_up(ConnPid),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
 	%% We skip the TLS handshake event to the TLS proxy.
 	_ = do_receive_event(tls_handshake_start),
 	StreamRef = gun:connect(ConnPid, #{
@@ -346,21 +365,26 @@ http1_tls_handshake_start_connect_over_https_proxy(Config) ->
 		tls_opts := _,
 		timeout := _
 	} = do_receive_event(tls_handshake_start),
-	true = is_tuple(Socket),
+	true = case Protocol of
+		http -> is_tuple(Socket);
+		http2 -> is_map(Socket)
+	end,
 	gun:close(ConnPid).
 
-http1_tls_handshake_end_error_connect_over_https_proxy(Config) ->
+tls_handshake_end_error_tls_connect_tls(Config) ->
 	doc("Confirm that the tls_handshake_end event callback is called on TLS handshake error "
 		"when using CONNECT to a TLS server via a TLS proxy."),
 	%% We use the wrong port on purpose to trigger a handshake error.
 	OriginPort = config(tcp_origin_port, Config),
-	{ok, _, ProxyPort} = rfc7231_SUITE:do_proxy_start(tls),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tls),
 	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
 		event_handler => {?MODULE, self()},
-		protocols => [config(name, config(tc_group_properties, Config))],
+		protocols => [Protocol],
 		transport => tls
 	}),
-	{ok, http} = gun:await_up(ConnPid),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
 	%% We skip the TLS handshake event to the TLS proxy.
 	_ = do_receive_event(tls_handshake_end),
 	StreamRef = gun:connect(ConnPid, #{
@@ -377,20 +401,25 @@ http1_tls_handshake_end_error_connect_over_https_proxy(Config) ->
 		timeout := _,
 		error := {tls_alert, _}
 	} = do_receive_event(tls_handshake_end),
-	true = is_tuple(Socket),
+	true = case Protocol of
+		http -> is_tuple(Socket);
+		http2 -> is_map(Socket)
+	end,
 	gun:close(ConnPid).
 
-http1_tls_handshake_end_ok_connect_over_https_proxy(Config) ->
+tls_handshake_end_ok_tls_connect_tls(Config) ->
 	doc("Confirm that the tls_handshake_end event callback is called on TLS handshake success "
 		"when using CONNECT to a TLS server via a TLS proxy."),
 	OriginPort = config(tls_origin_port, Config),
-	{ok, _, ProxyPort} = rfc7231_SUITE:do_proxy_start(tls),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tls),
 	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
 		event_handler => {?MODULE, self()},
-		protocols => [config(name, config(tc_group_properties, Config))],
+		protocols => [Protocol],
 		transport => tls
 	}),
-	{ok, http} = gun:await_up(ConnPid),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
 	%% We skip the TLS handshake event to the TLS proxy.
 	_ = do_receive_event(tls_handshake_end),
 	StreamRef = gun:connect(ConnPid, #{
@@ -410,6 +439,8 @@ http1_tls_handshake_end_ok_connect_over_https_proxy(Config) ->
 	true = is_pid(Socket),
 	gun:close(ConnPid).
 
+%% request_start/request_headers/request_end.
+
 request_start(Config) ->
 	doc("Confirm that the request_start event callback is called."),
 	do_request_event(Config, ?FUNCTION_NAME),
@@ -458,6 +489,110 @@ do_request_event_headers(Config, EventName) ->
 	Authority = iolist_to_binary(EventAuthority),
 	gun:close(Pid).
 
+request_start_connect(Config) ->
+	doc("Confirm that the request_start event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_request_event_connect(Config, request_start),
+	do_request_event_headers_connect(Config, request_start).
+
+request_headers_connect(Config) ->
+	doc("Confirm that the request_headers event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_request_event_connect(Config, request_headers),
+	do_request_event_headers_connect(Config, request_headers).
+
+do_request_event_connect(Config, EventName) ->
+	OriginPort = config(tcp_origin_port, Config),
+	Authority = iolist_to_binary([<<"localhost:">>, integer_to_list(OriginPort)]),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	#{
+		stream_ref := StreamRef1,
+		reply_to := ReplyTo,
+		function := connect,
+		method := <<"CONNECT">>,
+		authority := EventAuthority1,
+		headers := Headers1
+	} = do_receive_event(EventName),
+	Authority = iolist_to_binary(EventAuthority1),
+	%% Gun doesn't send headers with an HTTP/2 CONNECT request
+	%% so we only check that the headers are given as a list.
+	true = is_list(Headers1),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:get(ConnPid, "/", [], #{tunnel => StreamRef1}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		function := request,
+		method := <<"GET">>,
+		authority := EventAuthority2,
+		path := "/",
+		headers := [_|_]
+	} = do_receive_event(EventName),
+	Authority = iolist_to_binary(EventAuthority2),
+	gun:close(ConnPid).
+
+do_request_event_headers_connect(Config, EventName) ->
+	OriginPort = config(tcp_origin_port, Config),
+	Authority = iolist_to_binary([<<"localhost:">>, integer_to_list(OriginPort)]),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	#{
+		stream_ref := StreamRef1,
+		reply_to := ReplyTo,
+		function := connect,
+		method := <<"CONNECT">>,
+		authority := EventAuthority1,
+		headers := Headers1
+	} = do_receive_event(EventName),
+	Authority = iolist_to_binary(EventAuthority1),
+	%% Gun doesn't send headers with an HTTP/2 CONNECT request
+	%% so we only check that the headers are given as a list.
+	true = is_list(Headers1),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:put(ConnPid, "/", [
+		{<<"content-type">>, <<"text/plain">>}
+	], #{tunnel => StreamRef1}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		function := headers,
+		method := <<"PUT">>,
+		authority := EventAuthority2,
+		path := "/",
+		headers := [_|_]
+	} = do_receive_event(EventName),
+	Authority = iolist_to_binary(EventAuthority2),
+	gun:close(ConnPid).
+
 request_end(Config) ->
 	doc("Confirm that the request_end event callback is called."),
 	do_request_end(Config, ?FUNCTION_NAME),
@@ -522,6 +657,149 @@ do_request_end_headers_content_length_0(Config, EventName) ->
 	} = do_receive_event(EventName),
 	gun:close(Pid).
 
+request_end_connect(Config) ->
+	doc("Confirm that the request_end event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_request_end_connect(Config, request_end),
+	do_request_end_headers_connect(Config, request_end),
+	do_request_end_headers_content_length_connect(Config, request_end),
+	do_request_end_headers_content_length_0_connect(Config, request_end).
+
+do_request_end_connect(Config, EventName) ->
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	#{
+		stream_ref := StreamRef1,
+		reply_to := ReplyTo
+	} = do_receive_event(EventName),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:get(ConnPid, "/", [], #{tunnel => StreamRef1}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo
+	} = do_receive_event(EventName),
+	gun:close(ConnPid).
+
+do_request_end_headers_connect(Config, EventName) ->
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	#{
+		stream_ref := StreamRef1,
+		reply_to := ReplyTo
+	} = do_receive_event(EventName),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:put(ConnPid, "/", [
+		{<<"content-type">>, <<"text/plain">>}
+	], #{tunnel => StreamRef1}),
+	gun:data(ConnPid, StreamRef2, nofin, <<"Hello ">>),
+	gun:data(ConnPid, StreamRef2, fin, <<"world!">>),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo
+	} = do_receive_event(EventName),
+	gun:close(ConnPid).
+
+do_request_end_headers_content_length_connect(Config, EventName) ->
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	#{
+		stream_ref := StreamRef1,
+		reply_to := ReplyTo
+	} = do_receive_event(EventName),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:put(ConnPid, "/", [
+		{<<"content-type">>, <<"text/plain">>},
+		{<<"content-length">>, <<"12">>}
+	], #{tunnel => StreamRef1}),
+	gun:data(ConnPid, StreamRef2, nofin, <<"Hello ">>),
+	gun:data(ConnPid, StreamRef2, fin, <<"world!">>),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo
+	} = do_receive_event(EventName),
+	gun:close(ConnPid).
+
+do_request_end_headers_content_length_0_connect(Config, EventName) ->
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	#{
+		stream_ref := StreamRef1,
+		reply_to := ReplyTo
+	} = do_receive_event(EventName),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:put(ConnPid, "/", [
+		{<<"content-type">>, <<"text/plain">>},
+		{<<"content-length">>, <<"0">>}
+	], #{tunnel => StreamRef1}),
+	gun:data(ConnPid, StreamRef2, fin, <<>>),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo
+	} = do_receive_event(EventName),
+	gun:close(ConnPid).
+
+%% push_promise_start/push_promise_end.
+
 push_promise_start(Config) ->
 	doc("Confirm that the push_promise_start event callback is called."),
 	{ok, Pid, _} = do_gun_open(Config),
@@ -538,6 +816,41 @@ push_promise_start(Config) ->
 	} = do_receive_event(?FUNCTION_NAME),
 	gun:close(Pid).
 
+push_promise_start_connect(Config) ->
+	doc("Confirm that the push_promise_start event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_push_promise_start_connect(Config, http),
+	do_push_promise_start_connect(Config, http2).
+
+do_push_promise_start_connect(Config, ProxyProtocol) ->
+	OriginPort = config(tcp_origin_port, Config),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [ProxyProtocol]
+	}),
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [http2]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, http2} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:get(ConnPid, "/push", [], #{tunnel => StreamRef1}),
+	ReplyTo = self(),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo
+	} = do_receive_event(push_promise_start),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo
+	} = do_receive_event(push_promise_start),
+	gun:close(ConnPid).
+
 push_promise_end(Config) ->
 	doc("Confirm that the push_promise_end event callback is called."),
 	{ok, Pid, _} = do_gun_open(Config),
@@ -562,6 +875,49 @@ push_promise_end(Config) ->
 	} = do_receive_event(?FUNCTION_NAME),
 	gun:close(Pid).
 
+push_promise_end_connect(Config) ->
+	doc("Confirm that the push_promise_end event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_push_promise_end_connect(Config, http),
+	do_push_promise_end_connect(Config, http2).
+
+do_push_promise_end_connect(Config, ProxyProtocol) ->
+	OriginPort = config(tcp_origin_port, Config),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [ProxyProtocol]
+	}),
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [http2]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, http2} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:get(ConnPid, "/push", [], #{tunnel => StreamRef1}),
+	ReplyTo = self(),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		promised_stream_ref := [StreamRef1|_],
+		method := <<"GET">>,
+		uri := <<"http://",_/bits>>,
+		headers := [_|_]
+	} = do_receive_event(push_promise_end),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		promised_stream_ref := [StreamRef1|_],
+		method := <<"GET">>,
+		uri := <<"http://",_/bits>>,
+		headers := [_|_]
+	} = do_receive_event(push_promise_end),
+	gun:close(ConnPid).
+
 push_promise_followed_by_response(Config) ->
 	doc("Confirm that the push_promise_end event callbacks are followed by response_start."),
 	{ok, Pid, _} = do_gun_open(Config),
@@ -574,8 +930,10 @@ push_promise_followed_by_response(Config) ->
 	true = lists:member(PromisedStreamRef, [StreamRef1, StreamRef2, StreamRef3]),
 	gun:close(Pid).
 
+%% response_start/response_inform/response_headers/response_trailers/response_end.
+
 response_start(Config) ->
-	doc("Confirm that the request_start event callback is called."),
+	doc("Confirm that the response_start event callback is called."),
 	{ok, Pid, _} = do_gun_open(Config),
 	{ok, _} = gun:await_up(Pid),
 	StreamRef = gun:get(Pid, "/"),
@@ -586,8 +944,40 @@ response_start(Config) ->
 	} = do_receive_event(?FUNCTION_NAME),
 	gun:close(Pid).
 
+response_start_connect(Config) ->
+	doc("Confirm that the response_start event callback is called "
+		"for requests going through a CONNECT proxy."),
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	#{
+		stream_ref := StreamRef1,
+		reply_to := ReplyTo
+	} = do_receive_event(response_start),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:get(ConnPid, "/", [], #{tunnel => StreamRef1}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo
+	} = do_receive_event(response_start),
+	gun:close(ConnPid).
+
 response_inform(Config) ->
-	doc("Confirm that the request_inform event callback is called."),
+	doc("Confirm that the response_inform event callback is called."),
 	{ok, Pid, _} = do_gun_open(Config),
 	{ok, _} = gun:await_up(Pid),
 	StreamRef = gun:get(Pid, "/inform"),
@@ -606,8 +996,44 @@ response_inform(Config) ->
 	} = do_receive_event(?FUNCTION_NAME),
 	gun:close(Pid).
 
+response_inform_connect(Config) ->
+	doc("Confirm that the response_inform event callback is called "
+		"for requests going through a CONNECT proxy."),
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:get(ConnPid, "/inform", [], #{tunnel => StreamRef1}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		status := 103,
+		headers := [_|_]
+	} = do_receive_event(response_inform),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		status := 103,
+		headers := [_|_]
+	} = do_receive_event(response_inform),
+	gun:close(ConnPid).
+
 response_headers(Config) ->
-	doc("Confirm that the request_headers event callback is called."),
+	doc("Confirm that the response_headers event callback is called."),
 	{ok, Pid, _} = do_gun_open(Config),
 	{ok, _} = gun:await_up(Pid),
 	StreamRef = gun:get(Pid, "/"),
@@ -620,21 +1046,74 @@ response_headers(Config) ->
 	} = do_receive_event(?FUNCTION_NAME),
 	gun:close(Pid).
 
+response_headers_connect(Config) ->
+	doc("Confirm that the response_headers event callback is called "
+		"for requests going through a CONNECT proxy."),
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	#{
+		stream_ref := StreamRef1,
+		reply_to := ReplyTo,
+		status := 200,
+		headers := Headers1
+	} = do_receive_event(response_headers),
+	true = is_list(Headers1),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:get(ConnPid, "/", [], #{tunnel => StreamRef1}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		status := 200,
+		headers := [_|_]
+	} = do_receive_event(response_headers),
+	gun:close(ConnPid).
+
 response_trailers(Config) ->
-	doc("Confirm that the request_trailers event callback is called."),
-	{ok, Pid, _} = do_gun_open(Config),
-	{ok, _} = gun:await_up(Pid),
-	StreamRef = gun:get(Pid, "/trailers", [{<<"te">>, <<"trailers">>}]),
+	doc("Confirm that the response_trailers event callback is called "
+		"for requests going through a CONNECT proxy."),
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
 	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:get(ConnPid, "/trailers", [{<<"te">>, <<"trailers">>}], #{tunnel => StreamRef1}),
 	#{
-		stream_ref := StreamRef,
+		stream_ref := StreamRef2,
 		reply_to := ReplyTo,
 		headers := [_|_]
-	} = do_receive_event(?FUNCTION_NAME),
-	gun:close(Pid).
+	} = do_receive_event(response_trailers),
+	gun:close(ConnPid).
 
 response_end(Config) ->
-	doc("Confirm that the request_headers event callback is called."),
+	doc("Confirm that the response_end event callback is called."),
 	do_response_end(Config, ?FUNCTION_NAME, "/"),
 	do_response_end(Config, ?FUNCTION_NAME, "/empty"),
 	do_response_end(Config, ?FUNCTION_NAME, "/stream"),
@@ -651,8 +1130,47 @@ do_response_end(Config, EventName, Path) ->
 	} = do_receive_event(EventName),
 	gun:close(Pid).
 
+response_end_connect(Config) ->
+	doc("Confirm that the response_end event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_response_end_connect(Config, response_end, "/"),
+	do_response_end_connect(Config, response_end, "/empty"),
+	do_response_end_connect(Config, response_end, "/stream"),
+	do_response_end_connect(Config, response_end, "/trailers").
+
+do_response_end_connect(Config, EventName, Path) ->
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol]
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}, []),
+	%% @todo Figure out whether the response should end when the tunnel is established.
+%	#{
+%		stream_ref := StreamRef1,
+%		reply_to := ReplyTo
+%	} = do_receive_event(EventName),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:get(ConnPid, Path, [{<<"te">>, <<"trailers">>}], #{tunnel => StreamRef1}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo
+	} = do_receive_event(EventName),
+	gun:close(ConnPid).
+
 http1_response_end_body_close(Config) ->
-	doc("Confirm that the request_headers event callback is called "
+	doc("Confirm that the response_end event callback is called "
 		"when using HTTP/1.0 and the content-length header is not set."),
 	OriginPort = config(tcp_origin_port, Config),
 	Opts = #{
@@ -670,6 +1188,43 @@ http1_response_end_body_close(Config) ->
 	} = do_receive_event(response_end),
 	gun:close(Pid).
 
+%% @todo Figure out how to test both this and TLS handshake errors. Maybe a proxy option?
+%response_end_body_close_connect(Config) ->
+%	doc("Confirm that the response_end event callback is called "
+%		"when using HTTP/1.0 and the content-length header is not set "
+%		"for requests going through a CONNECT proxy."),
+%	OriginPort = config(tcp_origin_port, Config),
+%	Protocol = config(name, config(tc_group_properties, Config)),
+%	ReplyTo = self(),
+%	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+%	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+%		event_handler => {?MODULE, self()},
+%		protocols => [Protocol]
+%	}),
+%	{ok, Protocol} = gun:await_up(ConnPid),
+%	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+%	StreamRef1 = gun:connect(ConnPid, #{
+%		host => "localhost",
+%		port => OriginPort,
+%		protocols => [{http, #{version => 'HTTP/1.0'}}]
+%	}, []),
+%	%% @todo Figure out whether the response should end when the tunnel is established.
+%%	#{
+%%		stream_ref := StreamRef1,
+%%		reply_to := ReplyTo
+%%	} = do_receive_event(EventName),
+%	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+%	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+%	{up, http} = gun:await(ConnPid, StreamRef1),
+%	StreamRef2 = gun:get(ConnPid, "/stream", [], #{tunnel => StreamRef1}),
+%	#{
+%		stream_ref := StreamRef2,
+%		reply_to := ReplyTo
+%	} = do_receive_event(response_end),
+%	gun:close(ConnPid).
+
+%% ws_upgrade.
+
 ws_upgrade(Config) ->
 	doc("Confirm that the ws_upgrade event callback is called."),
 	{ok, Pid, _} = do_gun_open(Config),
@@ -683,6 +1238,39 @@ ws_upgrade(Config) ->
 	} = do_receive_event(?FUNCTION_NAME),
 	gun:close(Pid).
 
+ws_upgrade_connect(Config) ->
+	doc("Confirm that the ws_upgrade event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_ws_upgrade_connect(Config, http),
+	do_ws_upgrade_connect(Config, http2).
+
+do_ws_upgrade_connect(Config, ProxyProtocol) ->
+	OriginPort = config(tcp_origin_port, Config),
+	OriginProtocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [ProxyProtocol]
+	}),
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [OriginProtocol]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, OriginProtocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:ws_upgrade(ConnPid, "/ws", [], #{tunnel => StreamRef1}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		opts := #{}
+	} = do_receive_event(ws_upgrade),
+	gun:close(ConnPid).
+
 ws_upgrade_all_events(Config) ->
 	doc("Confirm that a Websocket upgrade triggers all relevant events."),
 	{ok, Pid, OriginPort} = do_gun_open(Config),
@@ -730,10 +1318,92 @@ ws_upgrade_all_events(Config) ->
 		headers := [_|_]
 	} = do_receive_event(response_inform),
 	#{
+		stream_ref := StreamRef,
 		protocol := ws
 	} = do_receive_event(protocol_changed),
 	gun:close(Pid).
 
+ws_upgrade_all_events_connect(Config) ->
+	doc("Confirm that a Websocket upgrade triggers all relevant events "
+		"for requests going through a CONNECT proxy."),
+	do_ws_upgrade_all_events_connect(Config, http),
+	do_ws_upgrade_all_events_connect(Config, http2).
+
+do_ws_upgrade_all_events_connect(Config, ProxyProtocol) ->
+	OriginPort = config(tcp_origin_port, Config),
+	OriginProtocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [ProxyProtocol]
+	}),
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [OriginProtocol]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, OriginProtocol} = gun:await(ConnPid, StreamRef1),
+	%% Skip all CONNECT-related events that may conflict.
+	_ = do_receive_event(request_start),
+	_ = do_receive_event(request_headers),
+	_ = do_receive_event(request_end),
+	_ = do_receive_event(response_start),
+	_ = do_receive_event(protocol_changed),
+	%% Check the Websocket events.
+	StreamRef2 = gun:ws_upgrade(ConnPid, "/ws", [], #{tunnel => StreamRef1}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		opts := #{}
+	} = do_receive_event(ws_upgrade),
+	Authority = iolist_to_binary([<<"localhost:">>, integer_to_list(OriginPort)]),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		function := ws_upgrade,
+		method := <<"GET">>,
+		authority := EventAuthority1,
+		path := "/ws",
+		headers := [_|_]
+	} = do_receive_event(request_start),
+	Authority = iolist_to_binary(EventAuthority1),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		function := ws_upgrade,
+		method := <<"GET">>,
+		authority := EventAuthority2,
+		path := "/ws",
+		headers := [_|_]
+	} = do_receive_event(request_headers),
+	Authority = iolist_to_binary(EventAuthority2),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo
+	} = do_receive_event(request_end),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo
+	} = do_receive_event(response_start),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		status := 101,
+		headers := [_|_]
+	} = do_receive_event(response_inform),
+	#{
+		stream_ref := StreamRef2,
+		protocol := ws
+	} = do_receive_event(protocol_changed),
+	gun:close(ConnPid).
+
+%% ws_recv_frame_start/ws_recv_frame_header/ws_recv_frame_end.
+
 ws_recv_frame_start(Config) ->
 	doc("Confirm that the ws_recv_frame_start event callback is called."),
 	{ok, Pid, _} = do_gun_open(Config),
@@ -750,6 +1420,42 @@ ws_recv_frame_start(Config) ->
 	} = do_receive_event(?FUNCTION_NAME),
 	gun:close(Pid).
 
+ws_recv_frame_start_connect(Config) ->
+	doc("Confirm that the ws_recv_frame_start event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_ws_recv_frame_start_connect(Config, http),
+	do_ws_recv_frame_start_connect(Config, http2).
+
+do_ws_recv_frame_start_connect(Config, ProxyProtocol) ->
+	OriginPort = config(tcp_origin_port, Config),
+	OriginProtocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [ProxyProtocol]
+	}),
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [OriginProtocol]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, OriginProtocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:ws_upgrade(ConnPid, "/ws", [], #{tunnel => StreamRef1}),
+	{upgrade, [<<"websocket">>], _} = gun:await(ConnPid, StreamRef2),
+	gun:ws_send(ConnPid, StreamRef2, {text, <<"Hello!">>}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		frag_state := undefined,
+		extensions := #{}
+	} = do_receive_event(ws_recv_frame_start),
+	gun:close(ConnPid).
+
 ws_recv_frame_header(Config) ->
 	doc("Confirm that the ws_recv_frame_header event callback is called."),
 	{ok, Pid, _} = do_gun_open(Config),
@@ -770,6 +1476,46 @@ ws_recv_frame_header(Config) ->
 	} = do_receive_event(?FUNCTION_NAME),
 	gun:close(Pid).
 
+ws_recv_frame_header_connect(Config) ->
+	doc("Confirm that the ws_recv_frame_header event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_ws_recv_frame_header_connect(Config, http),
+	do_ws_recv_frame_header_connect(Config, http2).
+
+do_ws_recv_frame_header_connect(Config, ProxyProtocol) ->
+	OriginPort = config(tcp_origin_port, Config),
+	OriginProtocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [ProxyProtocol]
+	}),
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [OriginProtocol]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, OriginProtocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:ws_upgrade(ConnPid, "/ws", [], #{tunnel => StreamRef1}),
+	{upgrade, [<<"websocket">>], _} = gun:await(ConnPid, StreamRef2),
+	gun:ws_send(ConnPid, StreamRef2, {text, <<"Hello!">>}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		frag_state := undefined,
+		extensions := #{},
+		type := text,
+		rsv := <<0:3>>,
+		len := 6,
+		mask_key := _
+	} = do_receive_event(ws_recv_frame_header),
+	gun:close(ConnPid).
+
 ws_recv_frame_end(Config) ->
 	doc("Confirm that the ws_recv_frame_end event callback is called."),
 	{ok, Pid, _} = do_gun_open(Config),
@@ -787,6 +1533,45 @@ ws_recv_frame_end(Config) ->
 	} = do_receive_event(?FUNCTION_NAME),
 	gun:close(Pid).
 
+ws_recv_frame_end_connect(Config) ->
+	doc("Confirm that the ws_recv_frame_end event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_ws_recv_frame_end_connect(Config, http),
+	do_ws_recv_frame_end_connect(Config, http2).
+
+do_ws_recv_frame_end_connect(Config, ProxyProtocol) ->
+	OriginPort = config(tcp_origin_port, Config),
+	OriginProtocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [ProxyProtocol]
+	}),
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [OriginProtocol]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, OriginProtocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:ws_upgrade(ConnPid, "/ws", [], #{tunnel => StreamRef1}),
+	{upgrade, [<<"websocket">>], _} = gun:await(ConnPid, StreamRef2),
+	gun:ws_send(ConnPid, StreamRef2, {text, <<"Hello!">>}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		extensions := #{},
+		close_code := undefined,
+		payload := <<"Hello!">>
+	} = do_receive_event(ws_recv_frame_end),
+	gun:close(ConnPid).
+
+%% ws_send_frame_start/ws_send_frame_end.
+
 ws_send_frame_start(Config) ->
 	doc("Confirm that the ws_send_frame_start event callback is called."),
 	do_ws_send_frame(Config, ?FUNCTION_NAME).
@@ -810,6 +1595,50 @@ do_ws_send_frame(Config, EventName) ->
 	} = do_receive_event(EventName),
 	gun:close(Pid).
 
+ws_send_frame_start_connect(Config) ->
+	doc("Confirm that the ws_send_frame_start event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_ws_send_frame_connect(Config, http, ws_send_frame_start),
+	do_ws_send_frame_connect(Config, http2, ws_send_frame_start).
+
+ws_send_frame_end_connect(Config) ->
+	doc("Confirm that the ws_send_frame_end event callback is called "
+		"for requests going through a CONNECT proxy."),
+	do_ws_send_frame_connect(Config, http, ws_send_frame_end),
+	do_ws_send_frame_connect(Config, http2, ws_send_frame_end).
+
+do_ws_send_frame_connect(Config, ProxyProtocol, EventName) ->
+	OriginPort = config(tcp_origin_port, Config),
+	OriginProtocol = config(name, config(tc_group_properties, Config)),
+	ReplyTo = self(),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [ProxyProtocol]
+	}),
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [OriginProtocol]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, OriginProtocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:ws_upgrade(ConnPid, "/ws", [], #{tunnel => StreamRef1}),
+	{upgrade, [<<"websocket">>], _} = gun:await(ConnPid, StreamRef2),
+	gun:ws_send(ConnPid, StreamRef2, {text, <<"Hello!">>}),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		extensions := #{},
+		frame := {text, <<"Hello!">>}
+	} = do_receive_event(EventName),
+	gun:close(ConnPid).
+
+%% protocol_changed.
+
 ws_protocol_changed(Config) ->
 	doc("Confirm that the protocol_changed event callback is called on Websocket upgrade success."),
 	{ok, Pid, _} = do_gun_open(Config),
@@ -820,45 +1649,100 @@ ws_protocol_changed(Config) ->
 	} = do_receive_event(protocol_changed),
 	gun:close(Pid).
 
-http1_protocol_changed_connect(Config) ->
+ws_protocol_changed_connect(Config) ->
+	doc("Confirm that the protocol_changed event callback is called on Websocket upgrade success "
+		"for requests going through a CONNECT proxy."),
+	do_ws_protocol_changed_connect(Config, http),
+	do_ws_protocol_changed_connect(Config, http2).
+
+do_ws_protocol_changed_connect(Config, ProxyProtocol) ->
+	OriginPort = config(tcp_origin_port, Config),
+	OriginProtocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [ProxyProtocol]
+	}),
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [OriginProtocol]
+	}, []),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, OriginProtocol} = gun:await(ConnPid, StreamRef1),
+	#{
+		stream_ref := StreamRef1,
+		protocol := OriginProtocol
+	} = do_receive_event(protocol_changed),
+	StreamRef2 = gun:ws_upgrade(ConnPid, "/ws", [], #{tunnel => StreamRef1}),
+	#{
+		stream_ref := StreamRef2,
+		protocol := ws
+	} = do_receive_event(protocol_changed),
+	gun:close(ConnPid).
+
+protocol_changed_connect(Config) ->
 	doc("Confirm that the protocol_changed event callback is called on CONNECT success "
 		"when connecting through a TCP server via a TCP proxy."),
+	do_protocol_changed_connect(Config, http),
+	do_protocol_changed_connect(Config, http2).
+
+do_protocol_changed_connect(Config, OriginProtocol) ->
 	OriginPort = config(tcp_origin_port, Config),
-	{ok, _, ProxyPort} = rfc7231_SUITE:do_proxy_start(tcp),
+	ProxyProtocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
 	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
 		event_handler => {?MODULE, self()},
-		protocols => [config(name, config(tc_group_properties, Config))],
+		protocols => [ProxyProtocol],
 		transport => tcp
 	}),
-	{ok, http} = gun:await_up(ConnPid),
-	_ = gun:connect(ConnPid, #{
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef = gun:connect(ConnPid, #{
 		host => "localhost",
 		port => OriginPort,
-		protocols => [http2]
+		protocols => [OriginProtocol]
 	}),
-	#{protocol := http2} = do_receive_event(protocol_changed),
+	#{
+		stream_ref := StreamRef,
+		protocol := OriginProtocol
+	} = do_receive_event(protocol_changed),
 	gun:close(ConnPid).
 
-http1_protocol_changed_connect_over_https_proxy(Config) ->
+protocol_changed_tls_connect(Config) ->
 	doc("Confirm that the protocol_changed event callback is called on CONNECT success "
-		"when connecting through a TLS server via a TLS proxy."),
+		"when connecting to a TLS server via a TLS proxy."),
+	do_protocol_changed_tls_connect(Config, http),
+	do_protocol_changed_tls_connect(Config, http2).
+
+do_protocol_changed_tls_connect(Config, OriginProtocol) ->
 	OriginPort = config(tls_origin_port, Config),
-	{ok, _, ProxyPort} = rfc7231_SUITE:do_proxy_start(tls),
+	ProxyProtocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tls),
 	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
 		event_handler => {?MODULE, self()},
-		protocols => [config(name, config(tc_group_properties, Config))],
+		protocols => [ProxyProtocol],
 		transport => tls
 	}),
-	{ok, http} = gun:await_up(ConnPid),
-	_ = gun:connect(ConnPid, #{
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef = gun:connect(ConnPid, #{
 		host => "localhost",
 		port => OriginPort,
 		transport => tls,
-		protocols => [http2]
+		protocols => [OriginProtocol]
 	}),
-	#{protocol := http2} = do_receive_event(protocol_changed),
+	#{
+		stream_ref := StreamRef,
+		protocol := OriginProtocol
+	} = do_receive_event(protocol_changed),
 	gun:close(ConnPid).
 
+%% transport_changed.
+
 http1_transport_changed_connect(Config) ->
 	doc("Confirm that the transport_changed event callback is called on CONNECT success "
 		"when connecting through a TLS server via a TCP proxy."),
@@ -905,28 +1789,87 @@ http1_transport_changed_connect_over_https_proxy(Config) ->
 	true = is_pid(Socket),
 	gun:close(ConnPid).
 
-http1_origin_changed_connect(Config) ->
+%% origin_changed.
+
+origin_changed_connect(Config) ->
 	doc("Confirm that the origin_changed event callback is called on CONNECT success."),
 	OriginPort = config(tcp_origin_port, Config),
-	{ok, _, ProxyPort} = rfc7231_SUITE:do_proxy_start(tcp),
+	ProxyProtocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(ProxyProtocol, tcp),
 	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
 		event_handler => {?MODULE, self()},
-		protocols => [config(name, config(tc_group_properties, Config))],
+		protocols => [ProxyProtocol],
 		transport => tcp
 	}),
-	{ok, http} = gun:await_up(ConnPid),
-	_ = gun:connect(ConnPid, #{
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, ProxyPid),
+	StreamRef = gun:connect(ConnPid, #{
 		host => "localhost",
 		port => OriginPort
 	}),
-	#{
+	Event = #{
 		type := connect,
 		origin_scheme := <<"http">>,
 		origin_host := "localhost",
 		origin_port := OriginPort
 	} = do_receive_event(origin_changed),
+	case ProxyProtocol of
+		http -> ok;
+		http2 ->
+			#{stream_ref := StreamRef} = Event
+	end,
 	gun:close(ConnPid).
 
+origin_changed_connect_connect(Config) ->
+	doc("Confirm that the origin_changed event callback is called on CONNECT success "
+		"when performed inside another CONNECT tunnel."),
+	OriginPort = config(tcp_origin_port, Config),
+	ProxyProtocol = config(name, config(tc_group_properties, Config)),
+	{ok, Proxy1Pid, Proxy1Port} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, Proxy2Pid, Proxy2Port} = do_proxy_start(ProxyProtocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", Proxy1Port, #{
+		event_handler => {?MODULE, self()},
+		protocols => [ProxyProtocol],
+		transport => tcp
+	}),
+	{ok, ProxyProtocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, Proxy1Pid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => Proxy2Port,
+		protocols => [ProxyProtocol]
+	}),
+	Event1 = #{
+		type := connect,
+		origin_scheme := <<"http">>,
+		origin_host := "localhost",
+		origin_port := Proxy2Port
+	} = do_receive_event(origin_changed),
+	case ProxyProtocol of
+		http -> ok;
+		http2 ->
+			#{stream_ref := StreamRef1} = Event1
+	end,
+	tunnel_SUITE:do_handshake_completed(ProxyProtocol, Proxy2Pid),
+	StreamRef2 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort
+	}, [], #{tunnel => StreamRef1}),
+	Event2 = #{
+		type := connect,
+		origin_scheme := <<"http">>,
+		origin_host := "localhost",
+		origin_port := OriginPort
+	} = do_receive_event(origin_changed),
+	case ProxyProtocol of
+		http -> ok;
+		http2 ->
+			#{stream_ref := StreamRef2} = Event2
+	end,
+	gun:close(ConnPid).
+
+%% cancel.
+
 cancel(Config) ->
 	doc("Confirm that the cancel event callback is called when we cancel a stream."),
 	{ok, Pid, _} = do_gun_open(Config),
@@ -957,6 +1900,71 @@ cancel_remote(Config) ->
 	} = do_receive_event(cancel),
 	gun:close(Pid).
 
+cancel_connect(Config) ->
+	doc("Confirm that the cancel event callback is called when we "
+		"cancel a stream running inside a CONNECT proxy."),
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol],
+		transport => tcp
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:post(ConnPid, "/stream", [], #{tunnel => StreamRef1}),
+	gun:cancel(ConnPid, StreamRef2),
+	ReplyTo = self(),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		endpoint := local,
+		reason := cancel
+	} = do_receive_event(cancel),
+	gun:close(ConnPid).
+
+cancel_remote_connect(Config) ->
+	doc("Confirm that the cancel event callback is called when the "
+		"remote endpoint cancels a stream running inside a CONNECT proxy."),
+	OriginPort = config(tcp_origin_port, Config),
+	Protocol = config(name, config(tc_group_properties, Config)),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(Protocol, tcp),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		event_handler => {?MODULE, self()},
+		protocols => [Protocol],
+		transport => tcp
+	}),
+	{ok, Protocol} = gun:await_up(ConnPid),
+	tunnel_SUITE:do_handshake_completed(Protocol, ProxyPid),
+	StreamRef1 = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [Protocol]
+	}),
+	%% @todo _IsFin is 'fin' for HTTP and 'nofin' for HTTP/2...
+	{response, _IsFin, 200, _} = gun:await(ConnPid, StreamRef1),
+	{up, Protocol} = gun:await(ConnPid, StreamRef1),
+	StreamRef2 = gun:post(ConnPid, "/stream", [], #{tunnel => StreamRef1}),
+	ReplyTo = self(),
+	#{
+		stream_ref := StreamRef2,
+		reply_to := ReplyTo,
+		endpoint := remote,
+		reason := _
+	} = do_receive_event(cancel),
+	gun:close(ConnPid).
+
+%% disconnect.
+
 disconnect(Config) ->
 	doc("Confirm that the disconnect event callback is called on disconnect."),
 	{ok, OriginPid, OriginPort} = init_origin(tcp),
@@ -970,6 +1978,8 @@ disconnect(Config) ->
 	} = do_receive_event(?FUNCTION_NAME),
 	gun:close(Pid).
 
+%% terminate.
+
 terminate(Config) ->
 	doc("Confirm that the terminate event callback is called on terminate."),
 	{ok, Pid, _} = do_gun_open(12345, Config),
@@ -1012,6 +2022,12 @@ do_receive_event(Event) ->
 		error(timeout)
 	end.
 
+do_proxy_start(Protocol, Transport) ->
+	case Protocol of
+		http -> rfc7231_SUITE:do_proxy_start(Transport);
+		http2 -> rfc7540_SUITE:do_proxy_start(Transport)
+	end.
+
 %% gun_event callbacks.
 
 init(EventData, Pid) ->
diff --git a/test/rfc7231_SUITE.erl b/test/rfc7231_SUITE.erl
index b1941c6..d6bcafa 100644
--- a/test/rfc7231_SUITE.erl
+++ b/test/rfc7231_SUITE.erl
@@ -115,10 +115,11 @@ do_proxy_loop(Transport, ClientSocket, OriginSocket) ->
 				{error, _} ->
 					ok
 			end;
+		%% Wait forever when a connection gets closed. We will exit with the test process.
 		{tcp_closed, _} ->
-			ok;
+			timer:sleep(infinity);
 		{ssl_closed, _} ->
-			ok;
+			timer:sleep(infinity);
 		Msg ->
 			error(Msg)
 	end.
diff --git a/test/rfc7540_SUITE.erl b/test/rfc7540_SUITE.erl
index a66c5ae..8680031 100644
--- a/test/rfc7540_SUITE.erl
+++ b/test/rfc7540_SUITE.erl
@@ -90,10 +90,11 @@ do_proxy_receive(Buffer, Proxy=#proxy{socket=Socket, transport=Transport}) ->
 			do_proxy_parse(<<Buffer/binary, Data0/bits>>, Proxy);
 		{tcp, OriginSocket, OriginData} ->
 			do_proxy_forward(Buffer, Proxy, OriginSocket, OriginData);
+		%% Wait forever when a connection gets closed. We will exit with the test process.
 		{tcp_closed, _} ->
-			ok;
+			timer:sleep(infinity);
 		{ssl_closed, _} ->
-			ok;
+			timer:sleep(infinity);
 		Msg ->
 			error(Msg)
 	end.