Add gun:trailers/3 for HTTP/1.1 and HTTP/2

doc/src/guide/protocols.asciidoc

@@ -111,6 +111,7 @@ current protocol.
 | put        | yes      | yes    | no
 | request    | yes      | yes    | no
 | data       | yes      | yes    | no
+| trailers   | yes      | yes    | no
 | await      | yes      | yes    | no
 | await_body | yes      | yes    | no
 | flush      | yes      | yes    | no

doc/src/manual/gun.asciidoc

@@ -34,6 +34,7 @@ Requests:
 * link:man:gun:headers(3)[gun:headers(3)] - Initiate the given request
 * link:man:gun:request(3)[gun:request(3)] - Perform the given request
 * link:man:gun:data(3)[gun:data(3)] - Stream the body of a request
+* link:man:gun:trailers(3)[gun:trailers(3)] - Send the request trailers
 
 Proxies:
 

doc/src/manual/gun.data.asciidoc

@@ -73,3 +73,4 @@ link:man:gun:patch(3)[gun:patch(3)],
 link:man:gun:post(3)[gun:post(3)],
 link:man:gun:put(3)[gun:put(3)],
 link:man:gun:headers(3)[gun:headers(3)]
+link:man:gun:trailers(3)[gun:trailers(3)]

doc/src/manual/gun.trailers.asciidoc

@@ -0,0 +1,76 @@
+= gun:trailers(3)
+
+== Name
+
+gun:trailers - Send the request trailers
+
+== Description
+
+[source,erlang]
+----
+trailers(ConnPid, StreamRef, Trailers) -> ok
+
+ConnPid   :: pid()
+StreamRef :: gun:stream_ref()
+Trailers  :: gun:req_headers()
+----
+
+Send the request trailers.
+
+This function can be used with
+link:man:gun:data(3)[gun:data(3)] to send trailing
+headers after streaming a body. Sending trailers
+concludes the stream, so all calls to gun:data(3) must
+use the `nofin` flag.
+
+This function can only be used if the original request
+had headers indicating that a body would be streamed.
+
+== Arguments
+
+ConnPid::
+
+The pid of the Gun connection process.
+
+StreamRef::
+
+Identifier of the stream for the original request.
+
+Trailers::
+
+Request trailers.
+
+== Return value
+
+The atom `ok` is returned.
+
+== Changelog
+
+* *2.3*: Function introduced.
+
+== Examples
+
+.Stream the body of a request
+[source,erlang]
+----
+StreamRef = gun:put(ConnPid, "/lang/fr_FR/hello", #{
+	<<"content-type">> => <<"text/plain">>,
+	<<"trailer">> => <<"x-greetings, content-md5">>
+]).
+gun:data(ConnPid, StreamRef, nofin, <<"Bonjour !\n">>).
+gun:data(ConnPid, StreamRef, nofin, <<"Bonsoir !\n">>).
+gun:trailers(ConnPid, StreamRef, #{
+	<<"x-greetings">> => <<"2">>,
+	<<"content-md5">> => <<"8a4f6308bec3e0069d5d59f7aae8fe04">>
+}).
+----
+
+== See also
+
+link:man:gun(3)[gun(3)],
+link:man:gun:data(3)[gun:data(3)]
+link:man:gun:patch(3)[gun:patch(3)],
+link:man:gun:post(3)[gun:post(3)],
+link:man:gun:put(3)[gun:put(3)],
+link:man:gun:headers(3)[gun:headers(3)]
+

src/gun.erl

@@ -59,6 +59,7 @@
 
 %% Streaming data.
 -export([data/4]).
+-export([trailers/3]).
 
 %% User pings.
 -export([ping/1]).
@@ -737,6 +738,11 @@ data(ServerPid, StreamRef, IsFin, Data) ->
 			gen_statem:cast(ServerPid, {data, self(), StreamRef, IsFin, Data})
 	end.
 
+-spec trailers(pid(), stream_ref(), req_headers()) -> ok.
+trailers(ServerPid, StreamRef, Trailers0) ->
+	Trailers = normalize_headers(Trailers0),
+	gen_statem:cast(ServerPid, {trailers, self(), StreamRef, Trailers}).
+
 %% User pings.
 
 -spec ping(pid()) -> reference().
@@ -1563,6 +1569,13 @@ handle_common_connected(cast, {data, ReplyTo, StreamRef, IsFin, Data}, _,
 		dereference_stream_ref(StreamRef, State),
 		ReplyTo, IsFin, Data, EvHandler, EvHandlerState0),
 	commands(Commands, State#state{event_handler_state=EvHandlerState});
+handle_common_connected(cast, {trailers, ReplyTo, StreamRef, Trailers}, _,
+		State=#state{protocol=Protocol, protocol_state=ProtoState,
+			event_handler=EvHandler, event_handler_state=EvHandlerState0}) ->
+	{Commands, EvHandlerState} = Protocol:trailers(ProtoState,
+		dereference_stream_ref(StreamRef, State),
+		ReplyTo, Trailers, EvHandler, EvHandlerState0),
+	commands(Commands, State#state{event_handler_state=EvHandlerState});
 handle_common_connected(info, {timeout, TRef, Name}, _,
 		State=#state{protocol=Protocol, protocol_state=ProtoState}) ->
 	Commands = Protocol:timeout(ProtoState, Name, TRef),

src/gun_http.erl

@@ -30,6 +30,7 @@
 -export([headers/12]).
 -export([request/13]).
 -export([data/7]).
+-export([trailers/6]).
 -export([connect/10]).
 -export([cancel/5]).
 -export([stream_info/2]).
@@ -764,6 +765,47 @@ data(State=#http_state{socket=Socket, transport=Transport, version=Version,
 			{[], EvHandlerState0}
 	end.
 
+%% We can only send trailers on the last created stream.
+trailers(State=#http_state{out=head}, StreamRef, ReplyTo, _, _, EvHandlerState) ->
+	error_stream_closed(State, StreamRef, ReplyTo),
+	{[], EvHandlerState};
+trailers(State=#http_state{streams=[]}, StreamRef, ReplyTo, _, _, EvHandlerState) ->
+	error_stream_not_found(State, StreamRef, ReplyTo),
+	{[], EvHandlerState};
+trailers(State=#http_state{socket=Socket, transport=Transport, version=Version,
+		out=Out, streams=Streams}, StreamRef, ReplyTo, Trailers, EvHandler,
+		EvHandlerState0) ->
+	case lists:last(Streams) of
+		#stream{ref=StreamRef, is_alive=true} ->
+			case Out of
+				body_chunked when Version =:= 'HTTP/1.1' ->
+					DataToSend = [
+						<<"0\r\n">>,
+						cow_http:headers(Trailers),
+						<<"\r\n">>
+					],
+					case Transport:send(Socket, DataToSend) of
+						ok ->
+							RequestEndEvent = #{
+								stream_ref => stream_ref(State, StreamRef),
+								reply_to => ReplyTo
+							},
+							EvHandlerState = EvHandler:request_end(RequestEndEvent,
+								EvHandlerState0),
+							{{state, State#http_state{out=head}}, EvHandlerState};
+						Error={error, _} ->
+							{Error, EvHandlerState0}
+					end;
+				_ ->
+					gun:reply(ReplyTo, {gun_error, self(),
+						stream_ref(State, StreamRef), {badstate,
+						"Trailers can only be used with HTTP/1.1 and chunked transfer-encoding."}})
+			end;
+		_ ->
+			error_stream_not_found(State, StreamRef, ReplyTo),
+			{[], EvHandlerState0}
+	end.
+
 connect(State, StreamRef, ReplyTo, _, _, _, _, CookieStore, _, EvHandlerState)
 		when is_list(StreamRef) ->
 	gun:reply(ReplyTo, {gun_error, self(), stream_ref(State, StreamRef),

src/gun_http2.erl

@@ -31,6 +31,7 @@
 -export([headers/12]).
 -export([request/13]).
 -export([data/7]).
+-export([trailers/6]).
 -export([connect/10]).
 -export([cancel/5]).
 -export([timeout/3]).
@@ -1258,6 +1259,10 @@ data(State, RealStreamRef=[StreamRef|_], ReplyTo, IsFin, Data, EvHandler, EvHand
 			{[], EvHandlerState0}
 	end.
 
+trailers(State, StreamRef, ReplyTo, Trailers, EvHandler, EvHandlerState) ->
+	Data = {trailers, Trailers},
+	data(State, StreamRef, ReplyTo, fin, Data, EvHandler, EvHandlerState).
+
 maybe_send_data(State=#http2_state{http2_machine=HTTP2Machine0}, StreamID, IsFin, Data0,
 		EvHandler, EvHandlerState) ->
 	Data = case is_tuple(Data0) of

src/gun_http3.erl

@@ -32,6 +32,7 @@
 -export([headers/12]).
 -export([request/13]).
 -export([data/7]).
+-export([trailers/6]).
 -export([connect/10]).
 -export([cancel/5]).
 -export([timeout/3]).
@@ -644,6 +645,10 @@ data(State=#http3_state{conn=Conn, transport=Transport}, StreamRef, _ReplyTo, Is
 %			{[], EvHandlerState}
 	end.
 
+trailers(_State, StreamRef, _ReplyTo, _Trailers, _EvHandler, _EvHandlerState)
+		when is_reference(StreamRef) ->
+	error(unimplemented).
+
 -spec connect(_, _, _, _, _, _, _, _, _, _) -> no_return().
 
 connect(_State, StreamRef, _ReplyTo, _Destination, _TunnelInfo, _Headers0,

src/gun_pool.erl

@@ -54,6 +54,7 @@
 
 %% Streaming data.
 -export([data/3]).
+-export([trailers/2]).
 
 %% Tunneling. (HTTP/2+ only.)
 %% @todo -export([connect/2]).
@@ -412,6 +413,10 @@ start_missing_pool(_Authority, _ReqOpts) ->
 data({ConnPid, StreamRef}, IsFin, Data) ->
 	gun:data(ConnPid, StreamRef, IsFin, Data).
 
+-spec trailers(pool_stream_ref(), gun:req_headers()) -> ok.
+trailers({ConnPid, StreamRef}, Trailers) ->
+	gun:trailers(ConnPid, StreamRef, Trailers).
+
 %% Awaiting gun messages.
 
 -spec await(pool_stream_ref()) -> gun:await_result().

test/gun_SUITE.erl

@@ -187,6 +187,73 @@ ignore_empty_data_fin_http2(_) ->
 	>> = Data,
 	gun:close(Pid).
 
+trailers_http(_) ->
+	doc("Trailers can be sent in HTTP/1.1."),
+	{ok, OriginPid, OriginPort} = init_origin(tcp, http),
+	{ok, Pid} = gun:open("localhost", OriginPort),
+	{ok, http} = gun:await_up(Pid),
+	handshake_completed = receive_from(OriginPid),
+	Ref = gun:post(Pid, "/", []),
+	gun:data(Pid, Ref, nofin, "hello"),
+	gun:trailers(Pid, Ref, [{<<"x-checksum">>, <<"abc123">>}]),
+	Data = receive_all_from(OriginPid, 500),
+	Lines = binary:split(Data, <<"\r\n">>, [global]),
+	%% Final chunk terminator, trailer, end (reversed).
+	[<<>>, <<>>, <<"x-checksum: abc123">>, <<"0">> | _] = lists:reverse(Lines),
+	gun:close(Pid).
+
+trailers_http_stream_not_found(_) ->
+	doc("Sending trailers on a non-existing HTTP/1.1 stream results in a gun_error."),
+	{ok, OriginPid, OriginPort} = init_origin(tcp, http),
+	{ok, Pid} = gun:open("localhost", OriginPort),
+	{ok, http} = gun:await_up(Pid),
+	handshake_completed = receive_from(OriginPid),
+	FakeRef = make_ref(),
+	gun:trailers(Pid, FakeRef, [{<<"x-checksum">>, <<"abc123">>}]),
+	receive
+		{gun_error, Pid, _, _} -> ok
+	after 1000 ->
+		error(timeout)
+	end,
+	gun:close(Pid).
+
+trailers_http2(_) ->
+	doc("Trailers are sent as a HEADERS frame with END_STREAM in HTTP/2."),
+	{ok, OriginPid, OriginPort} = init_origin(tcp, http2),
+	{ok, Pid} = gun:open("localhost", OriginPort, #{protocols => [http2]}),
+	{ok, http2} = gun:await_up(Pid),
+	handshake_completed = receive_from(OriginPid),
+	Ref = gun:post(Pid, "/", []),
+	gun:data(Pid, Ref, nofin, "hello"),
+	gun:trailers(Pid, Ref, [{<<"x-checksum">>, <<"abc123">>}]),
+	Data = receive_all_from(OriginPid, 500),
+	<<
+		%% HEADERS frame (request).
+		Len1:24, 1, _:40, _:Len1/unit:8,
+		%% DATA frame (nofin).
+		5:24, 0, _:7, 0:1, _:32, "hello",
+		%% HEADERS frame (trailers, END_STREAM).
+		Len2:24, 1, _:7, 1:1, _:32, _:Len2/unit:8
+	>> = Data,
+	gun:close(Pid).
+
+trailers_http2_without_data(_) ->
+	doc("Trailers can be sent without prior data in HTTP/2."),
+	{ok, OriginPid, OriginPort} = init_origin(tcp, http2),
+	{ok, Pid} = gun:open("localhost", OriginPort, #{protocols => [http2]}),
+	{ok, http2} = gun:await_up(Pid),
+	handshake_completed = receive_from(OriginPid),
+	Ref = gun:post(Pid, "/", []),
+	gun:trailers(Pid, Ref, [{<<"x-checksum">>, <<"abc123">>}]),
+	Data = receive_all_from(OriginPid, 500),
+	<<
+		%% HEADERS frame (request).
+		Len1:24, 1, _:40, _:Len1/unit:8,
+		%% HEADERS frame (trailers, END_STREAM).
+		Len2:24, 1, _:7, 1:1, _:32, _:Len2/unit:8
+	>> = Data,
+	gun:close(Pid).
+
 info(_) ->
 	doc("Get info from the Gun connection."),
 	{ok, ListenSocket} = gen_tcp:listen(0, [binary, {active, false}]),