Cowboy源码解读

1、cowboy主要依赖ranch和cowlib，
ranch在前面已经讲解了，cowlib主要是处理http协议里的一些数据
代码先行

-spec start_clear(ranch:ref(), ranch:opts(), opts())
    -> {ok, pid()} | {error, any()}.
start_clear(Ref, TransOpts0, ProtoOpts0) ->
    TransOpts1 = ranch:normalize_opts(TransOpts0),
    {TransOpts, ConnectionType} = ensure_connection_type(TransOpts1),
    ProtoOpts = ProtoOpts0#{connection_type => ConnectionType},
    ranch:start_listener(Ref, ranch_tcp, TransOpts, cowboy_clear, ProtoOpts).

回调模块cowboy_clear，收到的数据都会从这里开始

-module(cowboy_clear).
-behavior(ranch_protocol).

-export([start_link/4]).
-export([connection_process/4]).

-spec start_link(ranch:ref(), inet:socket(), module(), cowboy:opts()) -> {ok, pid()}.
start_link(Ref, _Socket, Transport, Opts) ->
    Pid = proc_lib:spawn_link(?MODULE, connection_process,
        [self(), Ref, Transport, Opts]),
    {ok, Pid}.

-spec connection_process(pid(), ranch:ref(), module(), cowboy:opts()) -> ok.
connection_process(Parent, Ref, Transport, Opts) ->
    ProxyInfo = case maps:get(proxy_header, Opts, false) of
        true ->
            {ok, ProxyInfo0} = ranch:recv_proxy_header(Ref, 1000),
            ProxyInfo0;
        false ->
            undefined
    end,
    {ok, Socket} = ranch:handshake(Ref),
    init(Parent, Ref, Socket, Transport, ProxyInfo, Opts, cowboy_http).

init(Parent, Ref, Socket, Transport, ProxyInfo, Opts, Protocol) ->
    _ = case maps:get(connection_type, Opts, supervisor) of
        worker -> ok;
        supervisor -> process_flag(trap_exit, true)
    end,
    Protocol:init(Parent, Ref, Socket, Transport, ProxyInfo, Opts).

调用 cowboy_http:init -> before_loop -> loop

loop(State=#state{parent=Parent, socket=Socket, transport=Transport, opts=Opts,
        timer=TimerRef, children=Children, in_streamid=InStreamID,
        last_streamid=LastStreamID, streams=Streams}, Buffer) ->
    {OK, Closed, Error} = Transport:messages(),
    InactivityTimeout = maps:get(inactivity_timeout, Opts, 300000),
    receive
        %% Discard data coming in after the last request
        %% we want to process was received fully.
        {OK, Socket, _} when InStreamID > LastStreamID ->
            before_loop(State, Buffer);
        %% Socket messages.
        {OK, Socket, Data} ->
            %% Only reset the timeout if it is idle_timeout (active streams).
            State1 = case Streams of
                [] -> State;
                _ -> set_timeout(State)
            end,
            parse(<< Buffer/binary, Data/binary >>, State1);
        {Closed, Socket} ->
            terminate(State, {socket_error, closed, 'The socket has been closed.'});
        {Error, Socket, Reason} ->
            terminate(State, {socket_error, Reason, 'An error has occurred on the socket.'});
        %% Timeouts.
        {timeout, Ref, {shutdown, Pid}} ->
            cowboy_children:shutdown_timeout(Children, Ref, Pid),
            loop(State, Buffer);
        {timeout, TimerRef, Reason} ->
            timeout(State, Reason);
        {timeout, _, _} ->
            loop(State, Buffer);
        %% System messages.
        {'EXIT', Parent, Reason} ->
            terminate(State, {stop, {exit, Reason}, 'Parent process terminated.'});
        {system, From, Request} ->
            sys:handle_system_msg(Request, From, Parent, ?MODULE, [], {State, Buffer});
        %% Messages pertaining to a stream.
        {{Pid, StreamID}, Msg} when Pid =:= self() ->
            loop(info(State, StreamID, Msg), Buffer);
        %% Exit signal from children.
        Msg = {'EXIT', Pid, _} ->
            loop(down(State, Pid, Msg), Buffer);
        %% Calls from supervisor module.
        {'$gen_call', From, Call} ->
            cowboy_children:handle_supervisor_call(Call, From, Children, ?MODULE),
            loop(State, Buffer);
        %% Unknown messages.
        Msg ->
            cowboy:log(warning, "Received stray message ~p.~n", [Msg], Opts),
            loop(State, Buffer)
    after InactivityTimeout ->
        terminate(State, {internal_error, timeout, 'No message or data received before timeout.'})
    end.

在这等着收socket数据
当匹配到{OK, Socket, Data} ，然后处理数据

parse(<<>>, State) ->
    before_loop(State, <<>>);
%% Do not process requests that come in after the last request
%% and discard the buffer if any to save memory.
parse(_, State=#state{in_streamid=InStreamID, in_state=#ps_request_line{},
        last_streamid=LastStreamID}) when InStreamID > LastStreamID ->
    before_loop(State, <<>>);
parse(Buffer, State=#state{in_state=#ps_request_line{empty_lines=EmptyLines}}) ->
    after_parse(parse_request(Buffer, State, EmptyLines));
parse(Buffer, State=#state{in_state=PS=#ps_header{headers=Headers, name=undefined}}) ->
    after_parse(parse_header(Buffer,
        State#state{in_state=PS#ps_header{headers=undefined}},
        Headers));
parse(Buffer, State=#state{in_state=PS=#ps_header{headers=Headers, name=Name}}) ->
    after_parse(parse_hd_before_value(Buffer,
        State#state{in_state=PS#ps_header{headers=undefined, name=undefined}},
        Headers, Name));
parse(Buffer, State=#state{in_state=#ps_body{}}) ->
    %% @todo We do not want to get the body automatically if the request doesn't ask for it.
    %% We may want to get bodies that are below a threshold without waiting, and buffer them
    %% until the request asks, though.
    after_parse(parse_body(Buffer, State)).

我们直接看after_parse

after_parse({request, Req=#{streamid := StreamID, method := Method,
        headers := Headers, version := Version},
        State0=#state{opts=Opts, streams=Streams0}, Buffer}) ->
            io:format("after_parse request~n"),
    try cowboy_stream:init(StreamID, Req, Opts) of
        {Commands, StreamState} ->
            TE = maps:get(<<"te">>, Headers, undefined),
            Streams = [#stream{id=StreamID, state=StreamState,
                method=Method, version=Version, te=TE}|Streams0],
            State1 = case maybe_req_close(State0, Headers, Version) of
                close -> State0#state{streams=Streams, last_streamid=StreamID};
                keepalive -> State0#state{streams=Streams}
            end,
            State = set_timeout(State1),
            parse(Buffer, commands(State, StreamID, Commands))
    catch Class:Exception ->
        cowboy:log(cowboy_stream:make_error_log(init,
            [StreamID, Req, Opts],
            Class, Exception, erlang:get_stacktrace()), Opts),
        early_error(500, State0, {internal_error, {Class, Exception},
            'Unhandled exception in cowboy_stream:init/3.'}, Req),
        parse(Buffer, State0)
    end;

cowboy_stream:init，需要注意下处理完了，还会去解析parse(Buffer, commands(State, StreamID, Commands))，如：body里面的数据

-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.

默认会调用 cowboy_stream_h:init

-spec init(cowboy_stream:streamid(), cowboy_req:req(), cowboy:opts())
    -> {[{spawn, pid(), timeout()}], #state{}}.
init(StreamID, Req=#{ref := Ref}, Opts) ->
    Env = maps:get(env, Opts, #{}),
    Middlewares = maps:get(middlewares, Opts, [cowboy_router, cowboy_handler]),
    Shutdown = maps:get(shutdown_timeout, Opts, 5000),
    Pid = proc_lib:spawn_link(?MODULE, request_process, [Req, Env, Middlewares]),
    Expect = expect(Req),
    {Commands, Next} = cowboy_stream:init(StreamID, Req, Opts),
    {[{spawn, Pid, Shutdown}|Commands],
        #state{next=Next, ref=Ref, pid=Pid, expect=Expect}}.

注意下
Middlewares = maps:get(middlewares, Opts, [cowboy_router, cowboy_handler]),
默认的中间件模块
然后启动一个进程，执行 request_process 这个函数

%% Request process.

%% We catch all exceptions in order to add the stacktrace to
%% the exit reason as it is not propagated by proc_lib otherwise
%% and therefore not present in the 'EXIT' message. We want
%% the stacktrace in order to simplify debugging of errors.
%%
%% This + the behavior in proc_lib means that we will get a
%% {Reason, Stacktrace} tuple for every exceptions, instead of
%% just for errors and throws.
%%
%% @todo Better spec.
-spec request_process(_, _, _) -> _.
request_process(Req, Env, Middlewares) ->
    OTP = erlang:system_info(otp_release),
    try
        execute(Req, Env, Middlewares)
    catch
        exit:Reason ->
            Stacktrace = erlang:get_stacktrace(),
            erlang:raise(exit, {Reason, Stacktrace}, Stacktrace);
        %% OTP 19 does not propagate any exception stacktraces,
        %% we therefore add it for every class of exception.
        _:Reason when OTP =:= "19" ->
            Stacktrace = erlang:get_stacktrace(),
            erlang:raise(exit, {Reason, Stacktrace}, Stacktrace);
        %% @todo I don't think this clause is necessary.
        Class:Reason ->
            erlang:raise(Class, Reason, erlang:get_stacktrace())
    end.

%% @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.

然后执行cowboy_router:execute 这个主要是解析匹配路由数据

-spec execute(Req, Env)
    -> {ok, Req, Env} | {stop, Req}
    when Req::cowboy_req:req(), Env::cowboy_middleware:env().
execute(Req=#{host := Host, path := Path}, Env=#{dispatch := Dispatch}) ->
    case match(Dispatch, Host, Path) of
        {ok, Handler, HandlerOpts, Bindings, HostInfo, PathInfo} ->
            {ok, Req#{
                host_info => HostInfo,
                path_info => PathInfo,
                bindings => Bindings
            }, Env#{
                handler => Handler,
                handler_opts => HandlerOpts
            }};
        {error, notfound, host} ->
            {stop, cowboy_req:reply(400, Req)};
        {error, badrequest, path} ->
            {stop, cowboy_req:reply(400, Req)};
        {error, notfound, path} ->
            {stop, cowboy_req:reply(404, Req)}
    end.

匹配到处理的Handler 放入Env
再执行cowboy_handler:execute

-spec execute(Req, Env) -> {ok, Req, Env}
    when Req::cowboy_req:req(), Env::cowboy_middleware:env().
execute(Req, Env=#{handler := Handler, handler_opts := HandlerOpts}) ->
    try Handler:init(Req, HandlerOpts) of
        {ok, Req2, State} ->
            Result = terminate(normal, Req2, State, Handler),
            {ok, Req2, Env#{result => Result}};
        {Mod, Req2, State} ->
            Mod:upgrade(Req2, Env, Handler, State);
        {Mod, Req2, State, Opts} ->
            Mod:upgrade(Req2, Env, Handler, State, Opts)
    catch Class:Reason ->
        terminate({crash, Class, Reason}, Req, HandlerOpts, Handler),
        erlang:raise(Class, Reason, erlang:get_stacktrace())
    end.

然后再调用你路由表中的Handler模块的 init 函数

如果是websocket会有个upgrade，注意下

提升方式：使用cowboy下的examples，然后自己跟踪下代码