Add the set_options stream handler command

The first two options to benefit from this are the
cowboy_compress_h options.

src/cowboy_compress_h.erl

@@ -34,10 +34,7 @@
 init(StreamID, Req, Opts) ->
 	State0 = check_req(Req),
 	CompressThreshold = maps:get(compress_threshold, Opts, 300),
-	DeflateFlush = case maps:get(compress_buffering, Opts, false) of
-		false -> sync;
-		true -> none
-	end,
+	DeflateFlush = buffering_to_zflush(maps:get(compress_buffering, Opts, false)),
 	{Commands0, Next} = cowboy_stream:init(StreamID, Req, Opts),
 	fold(Commands0, State0#state{next=Next,
 		threshold=CompressThreshold,
@@ -143,10 +140,24 @@ fold([Data0={data, _, _}|Tail], State0=#state{compress=gzip}, Acc) ->
 fold([Trailers={trailers, _}|Tail], State0=#state{compress=gzip}, Acc) ->
 	{{data, fin, Data}, State} = gzip_data({data, fin, <<>>}, State0),
 	fold(Tail, State, [Trailers, {data, nofin, Data}|Acc]);
+%% All the options from this handler can be updated for the current stream.
+fold([{set_options, Opts}|Tail], State=#state{
+		threshold=CompressThreshold0, deflate_flush=DeflateFlush0}, Acc) ->
+	CompressThreshold = maps:get(compress_threshold, Opts, CompressThreshold0),
+	DeflateFlush = case Opts of
+		#{compress_buffering := CompressBuffering} ->
+			buffering_to_zflush(CompressBuffering);
+		_ ->
+			DeflateFlush0
+	end,
+	fold(Tail, State#state{threshold=CompressThreshold, deflate_flush=DeflateFlush}, Acc);
 %% Otherwise, we have an unrelated command or compression is disabled.
 fold([Command|Tail], State, Acc) ->
 	fold(Tail, State, [Command|Acc]).
 
+buffering_to_zflush(true) -> none;
+buffering_to_zflush(false) -> sync.
+
 gzip_response({response, Status, Headers, Body}, State) ->
 	%% We can't call zlib:gzip/1 because it does an
 	%% iolist_to_binary(GzBody) at the end to return

src/cowboy_stream.erl

@@ -41,6 +41,7 @@
 	| {error_response, cowboy:http_status(), cowboy:http_headers(), iodata()}
 	| {switch_protocol, cowboy:http_headers(), module(), state()}
 	| {internal_error, any(), human_reason()}
+	| {set_options, map()}
 	| {log, logger:level(), io:format(), list()}
 	| stop].
 -export_type([commands/0]).

src/cowboy_stream_h.erl

@@ -227,6 +227,9 @@ info(StreamID, Push={push, _, _, _, _, _, _, _}, State) ->
 	do_info(StreamID, Push, [Push], State);
 info(StreamID, SwitchProtocol={switch_protocol, _, _, _}, State) ->
 	do_info(StreamID, SwitchProtocol, [SwitchProtocol], State#state{expect=undefined});
+%% Convert the set_options message to a command.
+info(StreamID, SetOptions={set_options, _}, State) ->
+	do_info(StreamID, SetOptions, [SetOptions], State);
 %% Unknown message, either stray or meant for a handler down the line.
 info(StreamID, Info, State) ->
 	do_info(StreamID, Info, [], State).

test/compress_SUITE.erl

@@ -149,16 +149,18 @@ gzip_stream_reply_content_encoding(Config) ->
 opts_compress_buffering_false(Config0) ->
 	doc("Confirm that the compress_buffering option can be set to false, "
 		"which is the default."),
+	Name = name(),
 	Fun = case config(ref, Config0) of
 		https_compress -> init_https;
 		h2_compress -> init_http2;
 		_ -> init_http
 	end,
-	Config = cowboy_test:Fun(name(), #{
+	Config = cowboy_test:Fun(Name, #{
 		env => #{dispatch => init_dispatch(Config0)},
 		stream_handlers => [cowboy_compress_h, cowboy_stream_h],
 		compress_buffering => false
 	}, Config0),
+	try
 		ConnPid = gun_open(Config),
 		Ref = gun:get(ConnPid, "/stream_reply/delayed",
 			[{<<"accept-encoding">>, <<"gzip">>}]),
@@ -171,22 +173,26 @@ opts_compress_buffering_false(Config0) ->
 		timer:sleep(1000),
 		{data, nofin, Data2} = gun:await(ConnPid, Ref, 100),
 		<<"data: World!\r\n\r\n">> = iolist_to_binary(zlib:inflate(Z, Data2)),
-	gun:close(ConnPid),
-	cowboy:stop_listener(name()).
+		gun:close(ConnPid)
+	after
+		cowboy:stop_listener(Name)
+	end.
 
 opts_compress_buffering_true(Config0) ->
 	doc("Confirm that the compress_buffering option can be set to true, "
 		"and that the data received is buffered."),
+	Name = name(),
 	Fun = case config(ref, Config0) of
 		https_compress -> init_https;
 		h2_compress -> init_http2;
 		_ -> init_http
 	end,
-	Config = cowboy_test:Fun(name(), #{
+	Config = cowboy_test:Fun(Name, #{
 		env => #{dispatch => init_dispatch(Config0)},
 		stream_handlers => [cowboy_compress_h, cowboy_stream_h],
 		compress_buffering => true
 	}, Config0),
+	try
 		ConnPid = gun_open(Config),
 		Ref = gun:get(ConnPid, "/stream_reply/delayed",
 			[{<<"accept-encoding">>, <<"gzip">>}]),
@@ -197,5 +203,78 @@ opts_compress_buffering_true(Config0) ->
 		%% The data gets buffered because it is too small.
 		{data, nofin, Data1} = gun:await(ConnPid, Ref, 100),
 		<<>> = iolist_to_binary(zlib:inflate(Z, Data1)),
-	gun:close(ConnPid),
-	cowboy:stop_listener(name()).
+		gun:close(ConnPid)
+	after
+		cowboy:stop_listener(Name)
+	end.
+
+set_options_compress_buffering_false(Config0) ->
+	doc("Confirm that the compress_buffering option can be dynamically "
+		"set to false by a handler and that the data received is not buffered."),
+	Name = name(),
+	Fun = case config(ref, Config0) of
+		https_compress -> init_https;
+		h2_compress -> init_http2;
+		_ -> init_http
+	end,
+	Config = cowboy_test:Fun(Name, #{
+		env => #{dispatch => init_dispatch(Config0)},
+		stream_handlers => [cowboy_compress_h, cowboy_stream_h],
+		compress_buffering => true
+	}, Config0),
+	try
+		ConnPid = gun_open(Config),
+		Ref = gun:get(ConnPid, "/stream_reply/set_options_buffering_false",
+			[{<<"accept-encoding">>, <<"gzip">>}]),
+		{response, nofin, 200, Headers} = gun:await(ConnPid, Ref),
+		{_, <<"gzip">>} = lists:keyfind(<<"content-encoding">>, 1, Headers),
+		Z = zlib:open(),
+		zlib:inflateInit(Z, 31),
+		{data, nofin, Data1} = gun:await(ConnPid, Ref, 100),
+		<<"data: Hello!\r\n\r\n">> = iolist_to_binary(zlib:inflate(Z, Data1)),
+		timer:sleep(1000),
+		{data, nofin, Data2} = gun:await(ConnPid, Ref, 100),
+		<<"data: World!\r\n\r\n">> = iolist_to_binary(zlib:inflate(Z, Data2)),
+		gun:close(ConnPid)
+	after
+		cowboy:stop_listener(Name)
+	end.
+
+set_options_compress_buffering_true(Config0) ->
+	doc("Confirm that the compress_buffering option can be dynamically "
+		"set to true by a handler and that the data received is buffered."),
+	Name = name(),
+	Fun = case config(ref, Config0) of
+		https_compress -> init_https;
+		h2_compress -> init_http2;
+		_ -> init_http
+	end,
+	Config = cowboy_test:Fun(Name, #{
+		env => #{dispatch => init_dispatch(Config0)},
+		stream_handlers => [cowboy_compress_h, cowboy_stream_h],
+		compress_buffering => false
+	}, Config0),
+	try
+		ConnPid = gun_open(Config),
+		Ref = gun:get(ConnPid, "/stream_reply/set_options_buffering_true",
+			[{<<"accept-encoding">>, <<"gzip">>}]),
+		{response, nofin, 200, Headers} = gun:await(ConnPid, Ref),
+		{_, <<"gzip">>} = lists:keyfind(<<"content-encoding">>, 1, Headers),
+		Z = zlib:open(),
+		zlib:inflateInit(Z, 31),
+		%% The data gets buffered because it is too small.
+		{data, nofin, Data1} = gun:await(ConnPid, Ref, 100),
+		<<>> = iolist_to_binary(zlib:inflate(Z, Data1)),
+		gun:close(ConnPid)
+	after
+		cowboy:stop_listener(Name)
+	end.
+
+set_options_compress_threshold_0(Config) ->
+	doc("Confirm that the compress_threshold option can be dynamically "
+		"set to change how large response bodies must be to be compressed."),
+	{200, Headers, GzBody} = do_get("/reply/set_options_threshold0",
+		[{<<"accept-encoding">>, <<"gzip">>}], Config),
+	{_, <<"gzip">>} = lists:keyfind(<<"content-encoding">>, 1, Headers),
+	_ = zlib:gunzip(GzBody),
+	ok.

test/handlers/compress_h.erl

@@ -19,7 +19,12 @@ init(Req0, State=reply) ->
 		<<"sendfile">> ->
 			AppFile = code:where_is_file("cowboy.app"),
 			Size = filelib:file_size(AppFile),
-			cowboy_req:reply(200, #{}, {sendfile, 0, Size, AppFile}, Req0)
+			cowboy_req:reply(200, #{}, {sendfile, 0, Size, AppFile}, Req0);
+		<<"set_options_threshold0">> ->
+			%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast.
+			#{pid := Pid, streamid := StreamID} = Req0,
+			Pid ! {{Pid, StreamID}, {set_options, #{compress_threshold => 0}}},
+			cowboy_req:reply(200, #{}, lists:duplicate(100, $a), Req0)
 	end,
 	{ok, Req, State};
 init(Req0, State=stream_reply) ->
@@ -52,13 +57,17 @@ init(Req0, State=stream_reply) ->
 			cowboy_req:stream_body({sendfile, 0, Size, AppFile}, fin, Req1),
 			Req1;
 		<<"delayed">> ->
-			Req1 = cowboy_req:stream_reply(200, Req0),
-			cowboy_req:stream_body(<<"data: Hello!\r\n\r\n">>, nofin, Req1),
-			timer:sleep(1000),
-			cowboy_req:stream_body(<<"data: World!\r\n\r\n">>, nofin, Req1),
-			timer:sleep(1000),
-			cowboy_req:stream_body(<<"data: Closing!\r\n\r\n">>, fin, Req1),
-			Req1
+			stream_delayed(Req0);
+		<<"set_options_buffering_false">> ->
+			%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast.
+			#{pid := Pid, streamid := StreamID} = Req0,
+			Pid ! {{Pid, StreamID}, {set_options, #{compress_buffering => false}}},
+			stream_delayed(Req0);
+		<<"set_options_buffering_true">> ->
+			%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast.
+			#{pid := Pid, streamid := StreamID} = Req0,
+			Pid ! {{Pid, StreamID}, {set_options, #{compress_buffering => true}}},
+			stream_delayed(Req0)
 	end,
 	{ok, Req, State}.
 
@@ -68,3 +77,12 @@ stream_reply(Headers, Req0) ->
 	_ = [cowboy_req:stream_body(Data, nofin, Req) || _ <- lists:seq(1,9)],
 	cowboy_req:stream_body(Data, fin, Req),
 	Req.
+
+stream_delayed(Req0) ->
+	Req = cowboy_req:stream_reply(200, Req0),
+	cowboy_req:stream_body(<<"data: Hello!\r\n\r\n">>, nofin, Req),
+	timer:sleep(1000),
+	cowboy_req:stream_body(<<"data: World!\r\n\r\n">>, nofin, Req),
+	timer:sleep(1000),
+	cowboy_req:stream_body(<<"data: Closing!\r\n\r\n">>, fin, Req),
+	Req.