Cowboy Stream Handler Reference Implementation

README.md

Cowboy, the popular Erlang web server, has deprecated middleware and replaced the concept with stream handlers, a more flexible, but more complicated API. This page contains a documented reference implementation of a stream handler, to help others when developing their stream handlers in the future.

How Stream Handlers Work

Stream handlers are daisy-chained such that each callback in the first handler invokes the same callback in the second, which invokes the next, which invokes the next, etc. This recursion is handled in the cowboy_stream module where the functions call the callbacks in the cowboy_stream behavior.

Protocol handlers invoke the stream handler chain for a new stream by calling cowboy_stream:init/3, which calls the init/3 callback in the first stream handler. The callback returns {Commands, State} where Commands is a list of commands to be executed by Cowboy and State is a "State" object created for the stream handler. cowboy_stream:init/3 returns {Commands, {Module, State} where Module is a reference to the stream handler.

Here is an example of this recursion with two stream handlers:

Protocol Handler 
  cowboy_stream:init()         -> {Commands, {handler1, Handler1State}}
    handler1:init()            -> {Commands, Handler1State}
      cowboy_stream:init()     -> {Commands, {handler2, Handler2State}}
        handler2:init()        -> {Commands, Handler2State}
          cowboy_stream:init() -> {[],       undefined}

The data/4, info/3, and terminate/3 callbacks in the cowboy_stream behavior take the State parameter value created by the init/3 callback and perform the same recursion as described, above. But, the corresponding cowboy_stream module functions take {Module, State} as the State parameter and invoke the callback in Module with it's associated State.

This reference implementation uses a standard convention where each callback stores the {Module, State}, for the next handler callback, in the state returned from the current callback as state#{next => {Module, State}}. Thus, each callback passes this value to the cowboy_stream function so it can invoke the next handler callback.

Here is an example of this recursion for the data/4 callback:

Protocol Handler
  cowboy_stream:data(..., {handler1, Handler1State#{next => {handler2, Handler2State}}})
    handler1:data(..., State = #{next => {handler2, Handler2State}})
      cowboy_stream:data(..., {handler2, Handler2State#{next => undefined}})
        handler2:data(..., State = #{next => undefined})
          cowboy_stream:data(..., undefined)

reference_stream_h.erl

-module(reference_stream_h).
-behavior(cowboy_stream).

-export([init/3]).
-export([data/4]).
-export([info/3]).
-export([terminate/3]).
-export([early_error/5]).

-record(state, {next}).

%% 
%% Callback Functions
%%

% This implementation was modeled from imetrics_cowboy_stream_h.
% This is a ready-to-use template that other stream handlers can be
% based on.

% This function is called when the request is first received and the
% headers are parsed and known.
-spec init(cowboy_stream:streamid(), cowboy_req:req(), cowboy:opts())
    -> {cowboy_stream:commands(), #state{}}.
init(StreamID, Req, Opts) ->
    % When we're ready to invoke the next handler in the chain, we call
    % cowboy_stream:init/3. It returns the "Next" tuple and a list of "commands"
    % to send to the upstream HTTP process.
    {Commands0, Next} = cowboy_stream:init(StreamID, Req, Opts),

    % The state record here holds the "Next" tuple, which includes a reference
    % to the next stream handler module and the state created by it's `init/3` 
    % callback.
    State0 = #state{next=Next},

    % We pass these commands thru our "fold" function, which allows us to iterate
    % thru each command and modify it if necessary. We can also add new commands
    % here or in that function if we want to. Ultimately, the list of commands is
    % returned by this function, and passed to the next handler upstream (or cowboy,
    % which passes them to the client, if this is the last handler.)
    fold(Commands0, State0).

% This function is called with the data sent in the request body. It can be called in
% chunks, so make sure to check the "IsFin" flag to decide whether you've received
% all the data.
-spec data(cowboy_stream:streamid(), cowboy_stream:fin(), cowboy_req:resp_body(), State)
    -> {cowboy_stream:commands(), State} when State::#state{}.
data(StreamID, IsFin, Data, State0=#state{next=Next0}) ->
    % Invoke the next stream handler in the chain.
    {Commands0, Next} = cowboy_stream:data(StreamID, IsFin, Data, Next0),

    % Process any commands returned by downstream stream handlers.
    fold(Commands0, State0#state{next=Next}).

% Any messages addressed to the stream will be sent to this function. Cowboy also uses
% this function to inform stream handlers of internal events.
-spec info(cowboy_stream:streamid(), any(), State)
    -> {cowboy_stream:commands(), State} when State::#state{}.
info(StreamID, Info, State0=#state{next=Next0}) ->
    % Invoke the next stream handler in the chain.
    {Commands0, Next} = cowboy_stream:info(StreamID, Info, Next0),

    % Process any commands returned by downstream stream handlers.
    fold(Commands0, State0#state{next=Next}).

% All streams will eventually be terminated and this function will be called. The
% only time when terminate/3 is not called is when an error occurs in init/3 of
% any stream handler, since the state is not available.
-spec terminate(cowboy_stream:streamid(), cowboy_stream:reason(), #state{}) -> any().
terminate(StreamID, Reason, #state{next=Next}) ->
    % Propagate the terminate to downstream stream handlers.
    cowboy_stream:terminate(StreamID, Reason, Next).

% This function will be called when an error occurs before the request-line and
% all headers have been received in an HTTP/1.1 request. It includes the partial
% request, and the response Cowboy intends to send. The return value of this
% function is the response, so it allows you to modify it if necessary.
-spec early_error(cowboy_stream:streamid(), cowboy_stream:reason(),
    cowboy_stream:partial_req(), Resp, cowboy:opts()) -> Resp
    when Resp::cowboy_stream:resp_command().
early_error(StreamID, Reason, PartialReq, Resp, Opts) ->
    % Propagate the error to downstream stream handlers.
    cowboy_stream:early_error(StreamID, Reason, PartialReq, Resp, Opts).

%%
%% Internal Functions
%%

% Any commands (see [0]) returned from downstream handlers are passed thru this
% fold function, giving us the chance to modify, omit, replace, or append to them.
%
% This module transparently passes the commands thru without any changes, but this
% logic remains in place since this is the best documented stream handler in all of
% our codebases, and it should exist as a reference for other stream handlers to hook
% into in the future.
% 
% [0]: https://ninenines.eu/docs/en/cowboy/2.8/manual/cowboy_stream/#commands
fold(Commands, State) ->
    fold(Commands, State, []).

% At the end of the recursion, reverse the accumulator so the original order of
% commands is preserved.
fold([], State, Acc) ->
    {lists:reverse(Acc), State};

% You can add additional cases here to catch specific commands.
% fold([{response, ResponseCode, Headers, Body0}|Tail], State0, Acc) ->
%     % do something with the body
%     Body = do_something(Body0),
%     Response = {response, ResponseCode, Headers, Body},
%     fold(Tail, State0, [Response|Acc]);

% This is the catch-all case for any commands we didn't match above, pass it
% thru unmodified.
fold([Command|Tail], State, Acc) ->
    fold(Tail, State, [Command|Acc]).