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.

1 files changed, 91 insertions, 15 deletions

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.