0
Fork 0
mirror of https://github.com/ninenines/cowboy.git synced 2025-07-14 12:20:24 +00:00

Add cowboy_req:cast/2

Better than sending messages manually.
This commit is contained in:
Loïc Hoguin 2019-10-07 13:25:49 +02:00
parent 5cdf78fd57
commit 2e8fcb9a9e
No known key found for this signature in database
GPG key ID: 8A9DF795F6FED764
6 changed files with 105 additions and 49 deletions

View file

@ -90,6 +90,10 @@ Response:
* link:man:cowboy_req:stream_trailers(3)[cowboy_req:stream_trailers(3)] - Send the response trailers * link:man:cowboy_req:stream_trailers(3)[cowboy_req:stream_trailers(3)] - Send the response trailers
* link:man:cowboy_req:push(3)[cowboy_req:push(3)] - Push a resource to the client * link:man:cowboy_req:push(3)[cowboy_req:push(3)] - Push a resource to the client
Stream handlers:
* link:man:cowboy_req:cast(3)[cowboy_req:cast(3)] - Cast a stream handler event
== Types == Types
=== push_opts() === push_opts()

View file

@ -0,0 +1,64 @@
= cowboy_req:cast(3)
== Name
cowboy_req:cast - Cast a stream handler event
== Description
[source,erlang]
----
cast(Event :: any(), Req :: cowboy_req:req()) -> ok
----
Cast a stream handler event.
The event will be passed to stream handlers through the
`info/3` callback.
== Arguments
Event::
The event to be sent to stream handlers.
Req::
The Req object.
== Return value
The atom `ok` is always returned. It can be safely ignored.
== Changelog
* *2.7*: Function introduced.
== Examples
.Increase the HTTP/1.1 idle timeout
[source,erlang]
----
cowboy_req:cast({set_options, #{
idle_timeout => 3600000
}}, Req).
----
.Add user data to metrics
----
cowboy_req:cast({set_options, #{
metrics_user_data => #{handler => ?MODULE}
}}, Req).
----
.Enable compression buffering
----
cowboy_req:cast({set_options, #{
compress_buffering => true
}}, Req).
----
== See also
link:man:cowboy_req(3)[cowboy_req(3)],
link:man:cowboy_stream(3)[cowboy_stream(3)]

View file

@ -277,9 +277,8 @@ relevant option's documentation for details.
Cowboy will forward all messages sent to the stream to Cowboy will forward all messages sent to the stream to
the `info/3` callback. To send a message to a stream, the `info/3` callback. To send a message to a stream,
send a message to the connection process with the form the function link:man:cowboy_req:cast(3)[cowboy_req:cast(3)]
`{{Pid, StreamID}, Msg}`. The connection process will can be used.
then forward `Msg` to the stream handlers.
Cowboy will also forward the exit signals for the Cowboy will also forward the exit signals for the
processes that the stream spawned. processes that the stream spawned.
@ -403,4 +402,5 @@ tuple.
link:man:cowboy(7)[cowboy(7)], link:man:cowboy(7)[cowboy(7)],
link:man:cowboy_http(3)[cowboy_http(3)], link:man:cowboy_http(3)[cowboy_http(3)],
link:man:cowboy_http2(3)[cowboy_http2(3)] link:man:cowboy_http2(3)[cowboy_http2(3)],
link:man:cowboy_req:cast(3)[cowboy_req:cast(3)]

View file

@ -92,6 +92,9 @@
-export([push/3]). -export([push/3]).
-export([push/4]). -export([push/4]).
%% Stream handlers.
-export([cast/2]).
%% Internal. %% Internal.
-export([response_headers/2]). -export([response_headers/2]).
@ -516,12 +519,12 @@ read_body(Req=#{has_body := false}, _) ->
{ok, <<>>, Req}; {ok, <<>>, Req};
read_body(Req=#{has_read_body := true}, _) -> read_body(Req=#{has_read_body := true}, _) ->
{ok, <<>>, Req}; {ok, <<>>, Req};
read_body(Req=#{pid := Pid, streamid := StreamID}, Opts) -> read_body(Req, Opts) ->
Length = maps:get(length, Opts, 8000000), Length = maps:get(length, Opts, 8000000),
Period = maps:get(period, Opts, 15000), Period = maps:get(period, Opts, 15000),
Timeout = maps:get(timeout, Opts, Period + 1000), Timeout = maps:get(timeout, Opts, Period + 1000),
Ref = make_ref(), Ref = make_ref(),
Pid ! {{Pid, StreamID}, {read_body, self(), Ref, Length, Period}}, cast({read_body, self(), Ref, Length, Period}, Req),
receive receive
{request_body, Ref, nofin, Body} -> {request_body, Ref, nofin, Body} ->
{more, Body, Req}; {more, Body, Req};
@ -775,10 +778,8 @@ inform(Status, Req) ->
-spec inform(cowboy:http_status(), cowboy:http_headers(), req()) -> ok. -spec inform(cowboy:http_status(), cowboy:http_headers(), req()) -> ok.
inform(_, _, #{has_sent_resp := _}) -> inform(_, _, #{has_sent_resp := _}) ->
error(function_clause); %% @todo Better error message. error(function_clause); %% @todo Better error message.
inform(Status, Headers, #{pid := Pid, streamid := StreamID}) inform(Status, Headers, Req) when is_integer(Status); is_binary(Status) ->
when is_integer(Status); is_binary(Status) -> cast({inform, Status, Headers}, Req).
Pid ! {{Pid, StreamID}, {inform, Status, Headers}},
ok.
-spec reply(cowboy:http_status(), Req) -> Req when Req::req(). -spec reply(cowboy:http_status(), Req) -> Req when Req::req().
reply(Status, Req) -> reply(Status, Req) ->
@ -823,11 +824,11 @@ reply(Status, Headers, Body, Req)
%% Don't send any body for HEAD responses. While the protocol code is %% Don't send any body for HEAD responses. While the protocol code is
%% supposed to enforce this rule, we prefer to avoid copying too much %% supposed to enforce this rule, we prefer to avoid copying too much
%% data around if we can avoid it. %% data around if we can avoid it.
do_reply(Status, Headers, _, Req=#{pid := Pid, streamid := StreamID, method := <<"HEAD">>}) -> do_reply(Status, Headers, _, Req=#{method := <<"HEAD">>}) ->
Pid ! {{Pid, StreamID}, {response, Status, response_headers(Headers, Req), <<>>}}, cast({response, Status, response_headers(Headers, Req), <<>>}, Req),
done_replying(Req, true); done_replying(Req, true);
do_reply(Status, Headers, Body, Req=#{pid := Pid, streamid := StreamID}) -> do_reply(Status, Headers, Body, Req) ->
Pid ! {{Pid, StreamID}, {response, Status, response_headers(Headers, Req), Body}}, cast({response, Status, response_headers(Headers, Req), Body}, Req),
done_replying(Req, true). done_replying(Req, true).
done_replying(Req, HasSentResp) -> done_replying(Req, HasSentResp) ->
@ -841,9 +842,8 @@ stream_reply(Status, Req) ->
-> Req when Req::req(). -> Req when Req::req().
stream_reply(_, _, #{has_sent_resp := _}) -> stream_reply(_, _, #{has_sent_resp := _}) ->
error(function_clause); error(function_clause);
stream_reply(Status, Headers=#{}, Req=#{pid := Pid, streamid := StreamID}) stream_reply(Status, Headers=#{}, Req) when is_integer(Status); is_binary(Status) ->
when is_integer(Status); is_binary(Status) -> cast({headers, Status, response_headers(Headers, Req)}, Req),
Pid ! {{Pid, StreamID}, {headers, Status, response_headers(Headers, Req)}},
done_replying(Req, headers). done_replying(Req, headers).
-spec stream_body(resp_body(), fin | nofin, req()) -> ok. -spec stream_body(resp_body(), fin | nofin, req()) -> ok.
@ -872,8 +872,8 @@ stream_body(Data, IsFin, Req=#{has_sent_resp := headers})
stream_body({data, self(), IsFin, Data}, Req). stream_body({data, self(), IsFin, Data}, Req).
%% @todo Do we need a timeout? %% @todo Do we need a timeout?
stream_body(Msg, #{pid := Pid, streamid := StreamID}) -> stream_body(Msg, Req=#{pid := Pid}) ->
Pid ! {{Pid, StreamID}, Msg}, cast(Msg, Req),
receive {data_ack, Pid} -> ok end. receive {data_ack, Pid} -> ok end.
-spec stream_events(cow_sse:event() | [cow_sse:event()], fin | nofin, req()) -> ok. -spec stream_events(cow_sse:event() | [cow_sse:event()], fin | nofin, req()) -> ok.
@ -883,9 +883,8 @@ stream_events(Events, IsFin, Req=#{has_sent_resp := headers}) ->
stream_body({data, self(), IsFin, cow_sse:events(Events)}, Req). stream_body({data, self(), IsFin, cow_sse:events(Events)}, Req).
-spec stream_trailers(cowboy:http_headers(), req()) -> ok. -spec stream_trailers(cowboy:http_headers(), req()) -> ok.
stream_trailers(Trailers, #{pid := Pid, streamid := StreamID, has_sent_resp := headers}) -> stream_trailers(Trailers, Req=#{has_sent_resp := headers}) ->
Pid ! {{Pid, StreamID}, {trailers, Trailers}}, cast({trailers, Trailers}, Req).
ok.
-spec push(iodata(), cowboy:http_headers(), req()) -> ok. -spec push(iodata(), cowboy:http_headers(), req()) -> ok.
push(Path, Headers, Req) -> push(Path, Headers, Req) ->
@ -895,14 +894,19 @@ push(Path, Headers, Req) ->
%% @todo Path, Headers, Opts, everything should be in proper binary, %% @todo Path, Headers, Opts, everything should be in proper binary,
%% or normalized when creating the Req object. %% or normalized when creating the Req object.
-spec push(iodata(), cowboy:http_headers(), req(), push_opts()) -> ok. -spec push(iodata(), cowboy:http_headers(), req(), push_opts()) -> ok.
push(Path, Headers, #{pid := Pid, streamid := StreamID, push(Path, Headers, Req=#{scheme := Scheme0, host := Host0, port := Port0}, Opts) ->
scheme := Scheme0, host := Host0, port := Port0}, Opts) ->
Method = maps:get(method, Opts, <<"GET">>), Method = maps:get(method, Opts, <<"GET">>),
Scheme = maps:get(scheme, Opts, Scheme0), Scheme = maps:get(scheme, Opts, Scheme0),
Host = maps:get(host, Opts, Host0), Host = maps:get(host, Opts, Host0),
Port = maps:get(port, Opts, Port0), Port = maps:get(port, Opts, Port0),
Qs = maps:get(qs, Opts, <<>>), Qs = maps:get(qs, Opts, <<>>),
Pid ! {{Pid, StreamID}, {push, Method, Scheme, Host, Port, Path, Qs, Headers}}, cast({push, Method, Scheme, Host, Port, Path, Qs, Headers}, Req).
%% Stream handlers.
-spec cast(any(), req()) -> ok.
cast(Msg, #{pid := Pid, streamid := StreamID}) ->
Pid ! {{Pid, StreamID}, Msg},
ok. ok.
%% Internal. %% Internal.

View file

@ -24,9 +24,7 @@ init(Req0, State=reply) ->
Size = filelib:file_size(AppFile), 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">> -> <<"set_options_threshold0">> ->
%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast. cowboy_req:cast({set_options, #{compress_threshold => 0}}, Req0),
#{pid := Pid, streamid := StreamID} = Req0,
Pid ! {{Pid, StreamID}, {set_options, #{compress_threshold => 0}}},
cowboy_req:reply(200, #{}, lists:duplicate(100, $a), Req0) cowboy_req:reply(200, #{}, lists:duplicate(100, $a), Req0)
end, end,
{ok, Req, State}; {ok, Req, State};
@ -62,14 +60,10 @@ init(Req0, State=stream_reply) ->
<<"delayed">> -> <<"delayed">> ->
stream_delayed(Req0); stream_delayed(Req0);
<<"set_options_buffering_false">> -> <<"set_options_buffering_false">> ->
%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast. cowboy_req:cast({set_options, #{compress_buffering => false}}, Req0),
#{pid := Pid, streamid := StreamID} = Req0,
Pid ! {{Pid, StreamID}, {set_options, #{compress_buffering => false}}},
stream_delayed(Req0); stream_delayed(Req0);
<<"set_options_buffering_true">> -> <<"set_options_buffering_true">> ->
%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast. cowboy_req:cast({set_options, #{compress_buffering => true}}, Req0),
#{pid := Pid, streamid := StreamID} = Req0,
Pid ! {{Pid, StreamID}, {set_options, #{compress_buffering => true}}},
stream_delayed(Req0) stream_delayed(Req0)
end, end,
{ok, Req, State}. {ok, Req, State}.

View file

@ -9,32 +9,22 @@ init(Req, State) ->
set_options(cowboy_req:binding(key, Req), Req, State). set_options(cowboy_req:binding(key, Req), Req, State).
set_options(<<"chunked_false">>, Req0, State) -> set_options(<<"chunked_false">>, Req0, State) ->
%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast. cowboy_req:cast({set_options, #{chunked => false}}, Req0),
#{pid := Pid, streamid := StreamID} = Req0,
Pid ! {{Pid, StreamID}, {set_options, #{chunked => false}}},
Req = cowboy_req:stream_reply(200, Req0), Req = cowboy_req:stream_reply(200, Req0),
cowboy_req:stream_body(<<0:8000000>>, fin, Req), cowboy_req:stream_body(<<0:8000000>>, fin, Req),
{ok, Req, State}; {ok, Req, State};
set_options(<<"chunked_false_ignored">>, Req0, State) -> set_options(<<"chunked_false_ignored">>, Req0, State) ->
%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast. cowboy_req:cast({set_options, #{chunked => false}}, Req0),
#{pid := Pid, streamid := StreamID} = Req0,
Pid ! {{Pid, StreamID}, {set_options, #{chunked => false}}},
Req = cowboy_req:reply(200, #{}, <<"Hello world!">>, Req0), Req = cowboy_req:reply(200, #{}, <<"Hello world!">>, Req0),
{ok, Req, State}; {ok, Req, State};
set_options(<<"idle_timeout_short">>, Req0, State) -> set_options(<<"idle_timeout_short">>, Req0, State) ->
%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast. cowboy_req:cast({set_options, #{idle_timeout => 500}}, Req0),
#{pid := Pid, streamid := StreamID} = Req0,
Pid ! {{Pid, StreamID}, {set_options, #{idle_timeout => 500}}},
{_, Body, Req} = cowboy_req:read_body(Req0), {_, Body, Req} = cowboy_req:read_body(Req0),
{ok, cowboy_req:reply(200, #{}, Body, Req), State}; {ok, cowboy_req:reply(200, #{}, Body, Req), State};
set_options(<<"idle_timeout_long">>, Req0, State) -> set_options(<<"idle_timeout_long">>, Req0, State) ->
%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast. cowboy_req:cast({set_options, #{idle_timeout => 60000}}, Req0),
#{pid := Pid, streamid := StreamID} = Req0,
Pid ! {{Pid, StreamID}, {set_options, #{idle_timeout => 60000}}},
{_, Body, Req} = cowboy_req:read_body(Req0), {_, Body, Req} = cowboy_req:read_body(Req0),
{ok, cowboy_req:reply(200, #{}, Body, Req), State}; {ok, cowboy_req:reply(200, #{}, Body, Req), State};
set_options(<<"metrics_user_data">>, Req, State) -> set_options(<<"metrics_user_data">>, Req, State) ->
%% @todo This should be replaced by a cowboy_req:cast/cowboy_stream:cast. cowboy_req:cast({set_options, #{metrics_user_data => #{handler => ?MODULE}}}, Req),
#{pid := Pid, streamid := StreamID} = Req,
Pid ! {{Pid, StreamID}, {set_options, #{metrics_user_data => #{handler => ?MODULE}}}},
{ok, cowboy_req:reply(200, #{}, <<"Hello world!">>, Req), State}. {ok, cowboy_req:reply(200, #{}, <<"Hello world!">>, Req), State}.