From 240da3f2d9bc02611b23c1ea0e7bbe8db9d99cd8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Hoguin?= Date: Thu, 15 Nov 2018 18:53:42 +0100 Subject: 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 | 19 +++++-- src/cowboy_stream.erl | 1 + src/cowboy_stream_h.erl | 3 + test/compress_SUITE.erl | 133 ++++++++++++++++++++++++++++++++++--------- test/handlers/compress_h.erl | 34 ++++++++--- 5 files changed, 151 insertions(+), 39 deletions(-) diff --git a/src/cowboy_compress_h.erl b/src/cowboy_compress_h.erl index bd3df42..95c49d3 100644 --- a/src/cowboy_compress_h.erl +++ b/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 diff --git a/src/cowboy_stream.erl b/src/cowboy_stream.erl index 49d1bb2..2dad6d0 100644 --- a/src/cowboy_stream.erl +++ b/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]). diff --git a/src/cowboy_stream_h.erl b/src/cowboy_stream_h.erl index 55a1ca2..6c10517 100644 --- a/src/cowboy_stream_h.erl +++ b/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). diff --git a/test/compress_SUITE.erl b/test/compress_SUITE.erl index 052d82d..1b82cc6 100644 --- a/test/compress_SUITE.erl +++ b/test/compress_SUITE.erl @@ -149,53 +149,132 @@ 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), - ConnPid = gun_open(Config), - Ref = gun:get(ConnPid, "/stream_reply/delayed", - [{<<"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), - cowboy:stop_listener(name()). + try + ConnPid = gun_open(Config), + Ref = gun:get(ConnPid, "/stream_reply/delayed", + [{<<"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. 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), - ConnPid = gun_open(Config), - Ref = gun:get(ConnPid, "/stream_reply/delayed", - [{<<"accept-encoding">>, <<"gzip">>}]), - {response, nofin, 200, Headers} = gun:await(ConnPid, Ref), + try + ConnPid = gun_open(Config), + Ref = gun:get(ConnPid, "/stream_reply/delayed", + [{<<"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_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), - 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), - cowboy:stop_listener(name()). + _ = zlib:gunzip(GzBody), + ok. diff --git a/test/handlers/compress_h.erl b/test/handlers/compress_h.erl index ffea05f..32830d9 100644 --- a/test/handlers/compress_h.erl +++ b/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. -- cgit v1.2.3