Initial commit with connection/streams

Breaking changes with previous commit. This is a very large change,
and I am giving up on making a single commit that fixes everything.
More commits will follow slowly adding back features, introducing
new tests and fixing the documentation.

This change contains most of the work toward unifying the interface
for handling both HTTP/1.1 and HTTP/2. HTTP/1.1 connections are now
no longer 1 process per connection; instead by default 1 process per
request is also created. This has a number of pros and cons.

Because it has cons, we also allow users to use a lower-level API
that acts on "streams" (requests/responses) directly at the connection
process-level. If performance is a concern, one can always write a
stream handler. The performance in this case will be even greater
than with Cowboy 1, although all the special handlers are unavailable.

When switching to Websocket, after the handler returns from init/2,
Cowboy stops the stream and the Websocket protocol takes over the
connection process. Websocket then calls websocket_init/2 for any
additional initialization such as timers, because the process is
different in init/2 and websocket_*/* functions. This however would
allow us to use websocket_init/2 for sending messages on connect,
instead of sending ourselves a message and be subject to races.
Note that websocket_init/2 is optional.

This is all a big change and while most of the tests pass, some
functionality currently doesn't. SPDY is broken and will be removed
soon in favor of HTTP/2. Automatic compression is currently disabled.
The cowboy_req interface probably still have a few functions that
need to be updated. The docs and examples do not refer the current
functionality anymore.

Everything will be fixed over time. Feedback is more than welcome.
Open a ticket!

1 files changed, 128 insertions, 0 deletions

diff --git a/src/cowboy_stream_h.erl b/src/cowboy_stream_h.erl
new file mode 100644
index 0000000..f924e28
--- /dev/null
+++ b/src/cowboy_stream_h.erl
@@ -0,0 +1,128 @@
+%% Copyright (c) 2016, Loïc Hoguin <[email protected]>
+%%
+%% Permission to use, copy, modify, and/or distribute this software for any
+%% purpose with or without fee is hereby granted, provided that the above
+%% copyright notice and this permission notice appear in all copies.
+%%
+%% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+%% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+%% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+%% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+%% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+%% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+%% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+-module(cowboy_stream_h).
+%% @todo -behaviour(cowboy_stream).
+
+%% @todo Maybe have a callback for the type of process this is, worker or supervisor.
+-export([init/3]).
+-export([data/4]).
+-export([info/3]).
+-export([terminate/3]).
+
+-export([execute/3]).
+-export([resume/5]).
+
+-record(state, {
+	pid = undefined :: pid(),
+	read_body_ref = undefined :: reference(),
+	read_body_length = 0 :: non_neg_integer(),
+	read_body_is_fin = nofin :: nofin | fin,
+	read_body_buffer = <<>> :: binary()
+}).
+
+%% @todo For shutting down children we need to have a timeout before we terminate
+%% the stream like supervisors do. So here just send a message to yourself first,
+%% and then decide what to do when receiving this message.
+
+%% @todo proper specs
+-spec init(_,_,_) -> _.
+init(_StreamID, Req, Opts) ->
+	Env = maps:get(env, Opts, #{}),
+	Middlewares = maps:get(middlewares, Opts, [cowboy_router, cowboy_handler]),
+	Shutdown = maps:get(shutdown, Opts, 5000),
+	Pid = proc_lib:spawn_link(?MODULE, execute, [Req, Env, Middlewares]),
+	{[{spawn, Pid, Shutdown}], #state{pid=Pid}}.
+
+%% If we receive data and stream is waiting for data:
+%%	If we accumulated enough data or IsFin=fin, send it.
+%%	If not, buffer it.
+%% If not, buffer it.
+
+%% @todo proper specs
+-spec data(_,_,_,_) -> _.
+data(_StreamID, IsFin, Data, State=#state{read_body_ref=undefined, read_body_buffer=Buffer}) ->
+	{[], State#state{read_body_is_fin=IsFin, read_body_buffer= << Buffer/binary, Data/binary >>}};
+data(_StreamID, nofin, Data, State=#state{read_body_length=Length, read_body_buffer=Buffer}) when byte_size(Data) + byte_size(Buffer) < Length ->
+	{[], State#state{read_body_buffer= << Buffer/binary, Data/binary >>}};
+data(_StreamID, IsFin, Data, State=#state{pid=Pid, read_body_ref=Ref, read_body_buffer=Buffer}) ->
+	Pid ! {request_body, Ref, IsFin, << Buffer/binary, Data/binary >>},
+	{[], State#state{read_body_ref=undefined, read_body_buffer= <<>>}}.
+
+%% @todo proper specs
+-spec info(_,_,_) -> _.
+info(_StreamID, {'EXIT', Pid, normal}, State=#state{pid=Pid}) ->
+	{[stop], State};
+info(_StreamID, Reason = {'EXIT', Pid, _}, State=#state{pid=Pid}) ->
+	{[{internal_error, Reason, 'Stream process crashed.'}], State};
+%% Request body, no body buffer but IsFin=fin.
+info(_StreamID, {read_body, Ref, _}, State=#state{pid=Pid, read_body_is_fin=fin, read_body_buffer= <<>>}) ->
+	Pid ! {request_body, Ref, fin, <<>>},
+	{[], State};
+%% Request body, body buffered large enough or complete.
+info(_StreamID, {read_body, Ref, Length}, State=#state{pid=Pid, read_body_is_fin=IsFin, read_body_buffer=Data})
+		when element(1, IsFin) =:= fin; byte_size(Data) >= Length ->
+	Pid ! {request_body, Ref, IsFin, Data},
+	{[], State#state{read_body_buffer= <<>>}};
+%% Request body, not enough to send yet.
+info(_StreamID, {read_body, Ref, Length}, State) ->
+	{[{flow, Length}], State#state{read_body_ref=Ref, read_body_length=Length}};
+%% Response.
+info(_StreamID, Response = {response, _, _, _}, State) ->
+	{[Response], State};
+info(_StreamID, Headers = {headers, _, _}, State) ->
+	{[Headers], State};
+info(_StreamID, Data = {data, _, _}, State) ->
+	{[Data], State};
+info(_StreamID, SwitchProtocol = {switch_protocol, _, _, _}, State) ->
+	{[SwitchProtocol], State};
+%% Stray message.
+info(_StreamID, _Msg, State) ->
+	%% @todo Cleanup if no reply was sent when stream ends.
+	{[], State}.
+
+%% @todo proper specs
+-spec terminate(_,_,_) -> _.
+terminate(_StreamID, _Reason, _State) ->
+	ok.
+
+%% Request process.
+
+%% @todo
+%-spec execute(cowboy_req:req(), #state{}, cowboy_middleware:env(), [module()])
+%	-> ok.
+-spec execute(_, _, _) -> _.
+execute(_, _, []) ->
+	ok; %% @todo Maybe error reason should differ here and there.
+execute(Req, Env, [Middleware|Tail]) ->
+	case Middleware:execute(Req, Env) of
+		{ok, Req2, Env2} ->
+			execute(Req2, Env2, Tail);
+		{suspend, Module, Function, Args} ->
+			proc_lib:hibernate(?MODULE, resume, [Env, Tail, Module, Function, Args]);
+		{stop, _Req2} ->
+			ok %% @todo Maybe error reason should differ here and there.
+	end.
+
+-spec resume(cowboy_middleware:env(), [module()],
+	module(), module(), [any()]) -> ok.
+resume(Env, Tail, Module, Function, Args) ->
+	case apply(Module, Function, Args) of
+		{ok, Req2, Env2} ->
+			execute(Req2, Env2, Tail);
+		{suspend, Module2, Function2, Args2} ->
+			proc_lib:hibernate(?MODULE, resume, [Env, Tail, Module2, Function2, Args2]);
+		{stop, _Req2} ->
+			ok %% @todo Maybe error reason should differ here and there.
+	end.