Fix shutdown for HTTP/1.1 pipeline

src/cowboy_http.erl

@@ -1445,10 +1445,12 @@ early_error(StatusCode0, #state{socket=Socket, transport=Transport,
 
 initiate_closing(State=#state{streams=[]}, Reason) ->
 	terminate(State, Reason);
-initiate_closing(State=#state{streams=[_Stream|Streams],
+initiate_closing(State=#state{streams=Streams,
 		out_streamid=OutStreamID}, Reason) ->
-	terminate_all_streams(State, Streams, Reason),
-	State#state{last_streamid=OutStreamID}.
+	{value, Stream, RestStreams}
+		= lists:keytake(OutStreamID, #stream.id, Streams),
+	terminate_all_streams(State, RestStreams, Reason),
+	State#state{streams=[Stream], last_streamid=OutStreamID}.
 
 -spec terminate(_, _) -> no_return().
 terminate(undefined, Reason) ->

test/cowboy_test.erl

@@ -157,6 +157,12 @@ raw_recv_head(Socket, Transport, Buffer) ->
 			Buffer
 	end.
 
+raw_recv_rest({raw_client, _, _}, Length, Buffer) when Length =:= byte_size(Buffer) ->
+	Buffer;
+raw_recv_rest({raw_client, Socket, Transport}, Length, Buffer) when Length > byte_size(Buffer) ->
+	{ok, Data} = Transport:recv(Socket, Length - byte_size(Buffer), 10000),
+	<< Buffer/binary, Data/binary >>.
+
 raw_recv({raw_client, Socket, Transport}, Length, Timeout) ->
 	Transport:recv(Socket, Length, Timeout).
 

test/http_SUITE.erl

@@ -24,6 +24,7 @@
 -import(cowboy_test, [raw_open/1]).
 -import(cowboy_test, [raw_send/2]).
 -import(cowboy_test, [raw_recv_head/1]).
+-import(cowboy_test, [raw_recv_rest/3]).
 -import(cowboy_test, [raw_recv/3]).
 -import(cowboy_test, [raw_expect_recv/2]).
 
@@ -449,33 +450,38 @@ graceful_shutdown_connection(Config) ->
 	doc("Check that the current request is handled before gracefully "
 	    "shutting down a connection."),
 	Dispatch = cowboy_router:compile([{"localhost", [
+		{"/hello", delay_hello_h,
+			#{delay => 0, notify_received => self()}},
 		{"/delay_hello", delay_hello_h,
-			#{delay => 500, notify_received => self()}},
-		{"/long_delay_hello", delay_hello_h,
-			#{delay => 10000, notify_received => self()}}
+			#{delay => 1000, notify_received => self()}}
 	]}]),
 	ProtoOpts = #{
 		env => #{dispatch => Dispatch}
 	},
 	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [{port, 0}], ProtoOpts),
 	Port = ranch:get_port(?FUNCTION_NAME),
 	try
-		ConnPid = gun_open([{type, tcp}, {protocol, http}, {port, Port}|Config]),
-		{ok, http} = gun:await_up(ConnPid),
-		#{socket := Socket} = gun:info(ConnPid),
-		CowboyConnPid = get_remote_pid_tcp(Socket),
-		CowboyConnRef = erlang:monitor(process, CowboyConnPid),
-		Ref1 = gun:get(ConnPid, "/delay_hello"),
-		Ref2 = gun:get(ConnPid, "/delay_hello"),
-		receive {request_received, <<"/delay_hello">>} -> ok end,
+		Client = raw_open([{type, tcp}, {port, Port}, {opts, []}|Config]),
+		ok = raw_send(Client,
+			"GET /delay_hello HTTP/1.1\r\n"
+			"Host: localhost\r\n\r\n"
+			"GET /hello HTTP/1.1\r\n"
+			"Host: localhost\r\n\r\n"),
 		receive {request_received, <<"/delay_hello">>} -> ok end,
+		receive {request_received, <<"/hello">>} -> ok end,
+		CowboyConnPid = get_remote_pid_tcp(element(2, Client)),
+		CowboyConnRef = erlang:monitor(process, CowboyConnPid),
 		ok = sys:terminate(CowboyConnPid, system_is_going_down),
-		{response, nofin, 200, RespHeaders} = gun:await(ConnPid, Ref1),
-		<<"close">> = proplists:get_value(<<"connection">>, RespHeaders),
-		{ok, RespBody} = gun:await_body(ConnPid, Ref1),
-		<<"Hello world!">> = iolist_to_binary(RespBody),
-		{error, {stream_error, _}} = gun:await(ConnPid, Ref2),
-		ok = gun_down(ConnPid),
+		Rest = case catch raw_recv_head(Client) of
+			{'EXIT', _} -> error(closed);
+			Data ->
+				{'HTTP/1.1', 200, _, Rest0} = cow_http:parse_status_line(Data),
+				{Headers, Rest1} = cow_http:parse_headers(Rest0),
+				<<"close">> = proplists:get_value(<<"connection">>, Headers),
+				Rest1
+		end,
+		<<"Hello world!">> = raw_recv_rest(Client, byte_size(<<"Hello world!">>), Rest),
+		{error, closed} = raw_recv(Client, 0, 1000),
 		receive
 			{'DOWN', CowboyConnRef, process, CowboyConnPid, _Reason} ->
 				ok

test/rfc7230_SUITE.erl

@@ -22,6 +22,7 @@
 -import(cowboy_test, [raw_open/1]).
 -import(cowboy_test, [raw_send/2]).
 -import(cowboy_test, [raw_recv_head/1]).
+-import(cowboy_test, [raw_recv_rest/3]).
 -import(cowboy_test, [raw_recv/3]).
 
 suite() ->
@@ -63,13 +64,7 @@ do_raw(Config, Data) ->
 	{Headers, Rest2} = cow_http:parse_headers(Rest),
 	case lists:keyfind(<<"content-length">>, 1, Headers) of
 		{_, LengthBin} when LengthBin =/= <<"0">> ->
-			Length = binary_to_integer(LengthBin),
-			Body = if
-				byte_size(Rest2) =:= Length -> Rest2;
-				true ->
-					{ok, Body0} = raw_recv(Client, Length - byte_size(Rest2), 5000),
-					<< Rest2/bits, Body0/bits >>
-			end,
+			Body = raw_recv_rest(Client, binary_to_integer(LengthBin), Rest2),
 			#{client => Client, version => Version, code => Code, reason => Reason, headers => Headers, body => Body};
 		_ ->
 			#{client => Client, version => Version, code => Code, reason => Reason, headers => Headers, body => <<>>}