feat: streaming responses support.

src/erf.erl

@@ -88,12 +88,15 @@
 -type response() :: {
     StatusCode :: pos_integer(),
     Headers :: [header()],
-    Body :: body() | {file, binary()}
+    Body :: body() | {file, binary()} | stream_body()
 }.
 -type route_patterns() :: [{Route :: binary(), RouteRegEx :: binary()}].
+-type send_chunk_fun() :: fun((iodata()) -> ok | {error, closed | timeout}).
 -type static_dir() :: {dir, binary()}.
 -type static_file() :: {file, binary()}.
 -type static_route() :: {Path :: binary(), Resource :: static_file() | static_dir()}.
+-type stream_body() :: {stream, stream_producer()}.
+-type stream_producer() :: fun((send_chunk_fun()) -> any()).
 
 %%% TYPE EXPORTS
 -export_type([
@@ -107,7 +110,10 @@
     request/0,
     response/0,
     route_patterns/0,
-    static_route/0
+    send_chunk_fun/0,
+    static_route/0,
+    stream_body/0,
+    stream_producer/0
 ]).
 
 %%% MACROS

src/erf_http_server/erf_http_server_elli.erl

@@ -91,7 +91,12 @@ init(Req, [Name]) ->
 handle(ElliRequest, [Name]) ->
     ErfRequest = preprocess(Name, ElliRequest),
     ErfResponse = erf_router:handle(Name, ErfRequest),
-    postprocess(ErfRequest, ErfResponse).
+    case ErfResponse of
+        {Status, Headers, {stream, Producer}} ->
+            handle_stream(Name, ElliRequest, Status, Headers, Producer);
+        _ ->
+            postprocess(ErfRequest, ErfResponse)
+    end.
 
 -spec handle_event(Event, Args, CallbackArgs) -> ok when
     Event :: elli_handler:event(),
@@ -131,6 +136,61 @@ handle_event(_Event, _Args, _CallbackArgs) ->
 %%%-----------------------------------------------------------------------------
 %%% INTERNAL FUNCTIONS
 %%%-----------------------------------------------------------------------------
+-spec handle_stream(Name, ElliRequest, Status, Headers, Producer) -> Result when
+    Name :: atom(),
+    ElliRequest :: elli:req(),
+    Status :: pos_integer(),
+    Headers :: [erf:header()],
+    Producer :: erf:stream_producer(),
+    Result :: elli_handler:result().
+handle_stream(Name, ElliRequest, Status, Headers, Producer) ->
+    case elli_request:chunk_ref(ElliRequest) of
+        {error, not_supported} ->
+            {ok, LogLevel} = erf_conf:log_level(Name),
+            ?LOG(
+                LogLevel,
+                "[erf] Streaming response requested on HTTP/1.0; returning 505"
+            ),
+            {505, [], <<>>};
+        ChunkRef ->
+            case Status of
+                200 ->
+                    ok;
+                _ ->
+                    {ok, LogLevel} = erf_conf:log_level(Name),
+                    ?LOG(
+                        LogLevel,
+                        "[erf] elli backend forces chunked responses to HTTP 200; "
+                        "status ~p from callback ignored",
+                        [Status]
+                    )
+            end,
+            _ = erlang:spawn(fun() -> drive_stream(Name, ChunkRef, Producer) end),
+            {chunk, Headers}
+    end.
+
+-spec drive_stream(Name, ChunkRef, Producer) -> ok when
+    Name :: atom(),
+    ChunkRef :: pid(),
+    Producer :: erf:stream_producer().
+drive_stream(Name, ChunkRef, Producer) ->
+    Send = fun(Data) -> elli_request:send_chunk(ChunkRef, Data) end,
+    try
+        _ = Producer(Send),
+        ok
+    catch
+        Class:Reason:Stack ->
+            {ok, LogLevel} = erf_conf:log_level(Name),
+            ?LOG(
+                LogLevel,
+                "[erf] Stream producer crashed ~p:~p~nStacktrace:~n~p",
+                [Class, Reason, Stack]
+            ),
+            ok
+    after
+        _ = elli_request:close_chunk(ChunkRef)
+    end.
+
 -spec build_elli_conf(Name, HTTPServerConf, ExtraElliConf) -> ElliConf when
     Name :: atom(),
     HTTPServerConf :: erf_http_server:conf(),

src/erf_router.erl

@@ -879,6 +879,8 @@ load_binary(ModuleName, Bin) ->
     Response :: erf:response().
 postprocess(_Request, {_Status, _Headers, {file, _Path}} = Response) ->
     Response;
+postprocess(_Request, {_Status, _Headers, {stream, _Producer}} = Response) ->
+    Response;
 postprocess(_Request, {Status, RawHeaders, RawBody}) ->
     ContentTypeHeader = string:casefold(<<"content-type">>),
     case proplists:get_value(ContentTypeHeader, RawHeaders, undefined) of