From 0f8452cafaa27aeffb103019564c086eacfd34f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Hoguin?= Date: Mon, 16 Jan 2017 14:22:43 +0100 Subject: Add support for multiple stream handlers The stream handlers can be specified using the protocol option 'stream_handlers'. It defaults to [cowboy_stream_h]. The cowboy_stream_h module currently does not forward the calls to further stream handlers. It feels like an edge case; usually we'd want to put our own handlers between the protocol code and the request process. I am therefore going to focus on other things for now. The various types and specifications for stream handlers have been updated and the cowboy_stream module can now be safely used as a behavior. The interface might change a little more, though. This commit does not include tests or documentation. They will follow separately. --- src/cowboy_stream.erl | 106 +++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 91 insertions(+), 15 deletions(-) (limited to 'src/cowboy_stream.erl') diff --git a/src/cowboy_stream.erl b/src/cowboy_stream.erl index 04601af..19b3663 100644 --- a/src/cowboy_stream.erl +++ b/src/cowboy_stream.erl @@ -14,31 +14,46 @@ -module(cowboy_stream). +-type state() :: any(). +-type human_reason() :: atom(). + -type streamid() :: any(). +-export_type([streamid/0]). + -type fin() :: fin | nofin. --type headers() :: map(). %% @todo cowboy:http_headers() when they're maps +-export_type([fin/0]). --type status_code() :: 100..999. %% @todo cowboy:http_status() when not binary --type state() :: any(). +%% @todo Perhaps it makes more sense to have resp_body in this module? --type commands() :: [{response, fin(), status_code(), headers()} +-type commands() :: [{response, cowboy:http_status(), cowboy:http_headers(), cowboy_req:resp_body()} + | {headers, cowboy:http_status(), cowboy:http_headers()} | {data, fin(), iodata()} - | {promise, binary(), binary(), binary(), binary(), headers()} + | {push, binary(), binary(), binary(), inet:port_number(), + binary(), binary(), cowboy:http_headers()} | {flow, auto | integer()} - | {spawn, pid()} - | {upgrade, module(), state()}]. + | {spawn, pid(), timeout()} + | {error_response, cowboy:http_status(), cowboy:http_headers(), iodata()} + | {internal_error, any(), human_reason()} + | {switch_protocol, cowboy:http_headers(), module(), state()} + %% @todo I'm not convinced we need this 'stop' command. + %% It's used on crashes, but error_response should + %% terminate the request instead. It's also used on + %% normal exits of children. I'm not sure what to do + %% there yet. Investigate. + | stop]. +-export_type([commands/0]). --type human_reason() :: atom(). --type reason() :: [{internal_error, timeout | {error | exit | throw, any()}, human_reason()} +-type reason() :: normal + | {internal_error, timeout | {error | exit | throw, any()}, human_reason()} | {socket_error, closed | atom(), human_reason()} - | {stream_error, cow_http2:error_reason(), human_reason()} - | {connection_error, cow_http2:error_reason(), human_reason()} - | {stop, cow_http2:frame(), human_reason()}]. + | {stream_error, cow_http2:error(), human_reason()} + | {connection_error, cow_http2:error(), human_reason()} + | {stop, cow_http2:frame(), human_reason()}. +-export_type([reason/0]). --callback init(streamid(), fin(), binary(), binary(), binary(), binary(), - headers(), cowboy:opts()) -> {commands(), state()}. +-callback init(streamid(), cowboy_req:req(), cowboy:opts()) -> {commands(), state()}. -callback data(streamid(), fin(), binary(), State) -> {commands(), State} when State::state(). --callback info(streamid(), any(), state()) -> {commands(), State} when State::state(). +-callback info(streamid(), any(), State) -> {commands(), State} when State::state(). -callback terminate(streamid(), reason(), state()) -> any(). %% @todo To optimize the number of active timers we could have a command @@ -51,3 +66,64 @@ %% %% This same timer can be used to try and send PING frames to help detect %% that the connection is indeed unresponsive. + +-export([init/3]). +-export([data/4]). +-export([info/3]). +-export([terminate/3]). + +%% Note that this and other functions in this module do NOT catch +%% exceptions. We want the exception to go all the way down to the +%% protocol code. +%% +%% OK the failure scenario is not so clear. The problem is +%% that the failure at any point in init/3 will result in the +%% corresponding state being lost. I am unfortunately not +%% confident we can do anything about this. If the crashing +%% handler just created a process, we'll never know about it. +%% Therefore at this time I choose to leave all failure handling +%% to the protocol process. +%% +%% Note that a failure in init/3 will result in terminate/3 +%% NOT being called. This is because the state is not available. + +-spec init(streamid(), cowboy_req:req(), cowboy:opts()) + -> {commands(), {module(), state()} | undefined}. +init(StreamID, Req, Opts) -> + case maps:get(stream_handlers, Opts, [cowboy_stream_h]) of + [] -> + {[], undefined}; + [Handler|Tail] -> + %% We call the next handler and remove it from the list of + %% stream handlers. This means that handlers that run after + %% it have no knowledge it exists. Should user require this + %% knowledge they can just define a separate option that will + %% be left untouched. + {Commands, State} = Handler:init(StreamID, Req, Opts#{stream_handlers => Tail}), + {Commands, {Handler, State}} + end. + +-spec data(streamid(), fin(), binary(), {Handler, State} | undefined) + -> {commands(), {Handler, State} | undefined} + when Handler::module(), State::state(). +data(_, _, _, undefined) -> + {[], undefined}; +data(StreamID, IsFin, Data, {Handler, State0}) -> + {Commands, State} = Handler:data(StreamID, IsFin, Data, State0), + {Commands, {Handler, State}}. + +-spec info(streamid(), any(), {Handler, State} | undefined) + -> {commands(), {Handler, State} | undefined} + when Handler::module(), State::state(). +info(_, _, undefined) -> + {[], undefined}; +info(StreamID, Info, {Handler, State0}) -> + {Commands, State} = Handler:info(StreamID, Info, State0), + {Commands, {Handler, State}}. + +-spec terminate(streamid(), reason(), {module(), state()} | undefined) -> ok. +terminate(_, _, undefined) -> + ok; +terminate(StreamID, Reason, {Handler, State}) -> + _ = Handler:terminate(StreamID, Reason, State), + ok. -- cgit v1.2.3