Add a compress_buffering option to cowboy_compress_h

Also changes the behavior to disable buffering by default, so
that the default works in all cases, including server-sent events.

src/cowboy_compress_h.erl

@@ -25,7 +25,8 @@
 	next :: any(),
 	threshold :: non_neg_integer() | undefined,
 	compress = undefined :: undefined | gzip,
-	deflate = undefined :: undefined | zlib:zstream()
+	deflate = undefined :: undefined | zlib:zstream(),
+	deflate_flush = sync :: none | sync
 }).
 
 -spec init(cowboy_stream:streamid(), cowboy_req:req(), cowboy:opts())
@@ -33,8 +34,14 @@
 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,
 	{Commands0, Next} = cowboy_stream:init(StreamID, Req, Opts),
-	fold(Commands0, State0#state{next=Next, threshold=CompressThreshold}).
+	fold(Commands0, State0#state{next=Next,
+		threshold=CompressThreshold,
+		deflate_flush=DeflateFlush}).
 
 -spec data(cowboy_stream:streamid(), cowboy_stream:fin(), cowboy_req:resp_body(), State)
 	-> {cowboy_stream:commands(), State} when State::#state{}.
@@ -176,9 +183,10 @@ gzip_headers({headers, Status, Headers0}, State) ->
 %% includes a checksum at the end of the stream. We have
 %% to read the file in memory, making this not suitable for
 %% large files.
-gzip_data({data, nofin, Sendfile={sendfile, _, _, _}}, State=#state{deflate=Z}) ->
+gzip_data({data, nofin, Sendfile={sendfile, _, _, _}},
+		State=#state{deflate=Z, deflate_flush=Flush}) ->
 	{ok, Data0} = read_file(Sendfile),
-	Data = zlib:deflate(Z, Data0),
+	Data = zlib:deflate(Z, Data0, Flush),
 	{{data, nofin, Data}, State};
 gzip_data({data, fin, Sendfile={sendfile, _, _, _}}, State=#state{deflate=Z}) ->
 	{ok, Data0} = read_file(Sendfile),
@@ -186,8 +194,8 @@ gzip_data({data, fin, Sendfile={sendfile, _, _, _}}, State=#state{deflate=Z}) ->
 	zlib:deflateEnd(Z),
 	zlib:close(Z),
 	{{data, fin, Data}, State#state{deflate=undefined}};
-gzip_data({data, nofin, Data0}, State=#state{deflate=Z}) ->
-	Data = zlib:deflate(Z, Data0),
+gzip_data({data, nofin, Data0}, State=#state{deflate=Z, deflate_flush=Flush}) ->
+	Data = zlib:deflate(Z, Data0, Flush),
 	{{data, nofin, Data}, State};
 gzip_data({data, fin, Data0}, State=#state{deflate=Z}) ->
 	Data = zlib:deflate(Z, Data0, finish),

src/cowboy_http.erl

@@ -25,6 +25,7 @@
 -export([system_code_change/4]).
 
 -type opts() :: #{
+	compress_buffering => boolean(),
 	compress_threshold => non_neg_integer(),
 	connection_type => worker | supervisor,
 	env => cowboy_middleware:env(),

src/cowboy_http2.erl

@@ -27,6 +27,7 @@
 -export([system_code_change/4]).
 
 -type opts() :: #{
+	compress_buffering => boolean(),
 	compress_threshold => non_neg_integer(),
 	connection_type => worker | supervisor,
 	enable_connect_protocol => boolean(),

test/compress_SUITE.erl

@@ -18,6 +18,7 @@
 
 -import(ct_helper, [config/2]).
 -import(ct_helper, [doc/1]).
+-import(ct_helper, [name/0]).
 -import(cowboy_test, [gun_open/1]).
 
 %% ct.
@@ -144,3 +145,57 @@ gzip_stream_reply_content_encoding(Config) ->
 	{_, <<"compress">>} = lists:keyfind(<<"content-encoding">>, 1, Headers),
 	100000 = iolist_size(Body),
 	ok.
+
+opts_compress_buffering_false(Config0) ->
+	doc("Confirm that the compress_buffering option can be set to false, "
+		"which is the default."),
+	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),
+	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()).
+
+opts_compress_buffering_true(Config0) ->
+	doc("Confirm that the compress_buffering option can be set to true, "
+		"and that the data received is buffered."),
+	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),
+	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),
+	cowboy:stop_listener(name()).

test/handlers/compress_h.erl

@@ -50,6 +50,14 @@ init(Req0, State=stream_reply) ->
 			cowboy_req:stream_body({sendfile, 0, Size, AppFile}, nofin, Req1),
 			cowboy_req:stream_body(Data, nofin, Req1),
 			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
 	end,
 	{ok, Req, State}.