add SSE support

apps/iot/src/http_handlers/event_stream_handler.erl

@@ -0,0 +1,46 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 08. 5月 2025 13:00
+%%%-------------------------------------------------------------------
+-module(event_stream_handler).
+-author("anlicheng").
+
+%% API
+-export([init/2]).
+
+init(Req0, Opts) ->
+    Method = binary_to_list(cowboy_req:method(Req0)),
+    Path = binary_to_list(cowboy_req:path(Req0)),
+    GetParams0 = cowboy_req:parse_qs(Req0),
+    GetParams = maps:from_list(GetParams0),
+
+    #{<<"task_id">> := TaskId0} = GetParams,
+    TaskId = binary_to_integer(TaskId0),
+
+    lager:debug("method: ~p, path: ~p,  get: ~p", [Method, Path, GetParams]),
+    Req1 = cowboy_req:stream_reply(200, #{
+        <<"content-type">> => <<"text/event-stream">>,
+        <<"cache-control">> => <<"no-cache">>
+    }, Req0),
+
+    ok = iot_event_stream_observer:add_listener(self(), TaskId),
+    receiver_events(TaskId, Req1),
+
+    {ok, Req1, Opts}.
+
+receiver_events(TaskId, Req) ->
+    receive
+        {stream_data, TaskId, Type, Stream} ->
+            Data = jiffy:encode(#{<<"type">> => Type, <<"stream">> => Stream}, [force_utf8]),
+            Body = iolist_to_binary([<<"event: message\n">>, <<"data: ", Data/binary, "\n">>, <<"\n">>]),
+            ok = cowboy_req:stream_body(Body, nofin, Req),
+
+            receiver_events(TaskId, Req);
+        {stream_close, TaskId} ->
+            CloseFrame = iolist_to_binary([<<"event: close\n">>, <<"data: bye\n">>, <<"\n">>]),
+            ok = cowboy_req:stream_body(CloseFrame, fin, Req)
+    end.

apps/iot/src/iot_app.erl

@@ -49,7 +49,8 @@ start_http_server() ->
         {'_', [
             {"/host/[...]", http_protocol, [host_handler]},
             {"/container/[...]", http_protocol, [container_handler]},
-            {"/device/[...]", http_protocol, [device_handler]}
+            {"/device/[...]", http_protocol, [device_handler]},
+            {"/event_stream", event_stream_handler, []}
         ]}
     ]),
 

apps/iot/src/iot_event_stream_observer.erl

@@ -0,0 +1,128 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 26. 9月 2025 12:19
+%%%-------------------------------------------------------------------
+-module(iot_event_stream_observer).
+-author("anlicheng").
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0]).
+-export([add_listener/2, stream_data/3, stream_close/1]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+-define(SERVER, ?MODULE).
+
+-record(state, {
+    listeners = #{}
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+-spec add_listener(ListenerPid :: pid(), TaskId :: integer()) -> ok.
+add_listener(ListenerPid, TaskId) when is_pid(ListenerPid), is_integer(TaskId) ->
+    gen_server:call(?SERVER, {add_listener, ListenerPid, TaskId}).
+
+-spec stream_data(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> no_return().
+stream_data(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
+    gen_server:cast(?SERVER, {stream_data, TaskId, Type, Stream}).
+
+-spec stream_close(TaskId :: integer()) -> no_return().
+stream_close(TaskId) when is_integer(TaskId) ->
+    gen_server:cast(?SERVER, {stream_close, TaskId}).
+
+%% @doc Spawns the server and registers the local name (unique)
+-spec(start_link() ->
+    {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+%% @private
+%% @doc Initializes the server
+-spec(init(Args :: term()) ->
+    {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term()} | ignore).
+init([]) ->
+    {ok, #state{}}.
+
+%% @private
+%% @doc Handling call messages
+-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
+    State :: #state{}) ->
+    {reply, Reply :: term(), NewState :: #state{}} |
+    {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
+    {noreply, NewState :: #state{}} |
+    {noreply, NewState :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
+    {stop, Reason :: term(), NewState :: #state{}}).
+handle_call({add_listener, ListenerPid, TaskId}, _From, State = #state{listeners = Listeners}) ->
+    erlang:monitor(process, ListenerPid),
+    {reply, ok, State#state{listeners = maps:put(TaskId, ListenerPid, Listeners)}}.
+
+%% @private
+%% @doc Handling cast messages
+-spec(handle_cast(Request :: term(), State :: #state{}) ->
+    {noreply, NewState :: #state{}} |
+    {noreply, NewState :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term(), NewState :: #state{}}).
+handle_cast({stream_data, TaskId, Type, Stream}, State = #state{listeners = Listeners}) ->
+    case maps:find(TaskId, Listeners) of
+        error ->
+            ok;
+        {ok, ListenerPid} ->
+            is_process_alive(ListenerPid) andalso ListenerPid ! {stream_data, TaskId, Type, Stream}
+    end,
+    {noreply, State};
+handle_cast({stream_close, TaskId}, State = #state{listeners = Listeners}) ->
+    case maps:find(TaskId, Listeners) of
+        error ->
+            ok;
+        {ok, ListenerPid} ->
+            is_process_alive(ListenerPid) andalso ListenerPid ! {stream_close, TaskId}
+    end,
+    {noreply, State}.
+
+%% @private
+%% @doc Handling all non call/cast messages
+-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
+    {noreply, NewState :: #state{}} |
+    {noreply, NewState :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term(), NewState :: #state{}}).
+handle_info({'DOWN', _Ref, process, Pid, _Reason}, State = #state{listeners = Listeners}) ->
+    NListeners = maps:filter(fun(_, ListenerPid) -> ListenerPid /= Pid end, Listeners),
+    {noreply, State#state{listeners = NListeners}}.
+
+%% @private
+%% @doc This function is called by a gen_server when it is about to
+%% terminate. It should be the opposite of Module:init/1 and do any
+%% necessary cleaning up. When it returns, the gen_server terminates
+%% with Reason. The return value is ignored.
+-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
+    State :: #state{}) -> term()).
+terminate(_Reason, _State = #state{}) ->
+    ok.
+
+%% @private
+%% @doc Convert process state when code is changed
+-spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
+    Extra :: term()) ->
+    {ok, NewState :: #state{}} | {error, Reason :: term()}).
+code_change(_OldVsn, State = #state{}, _Extra) ->
+    {ok, State}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================

apps/iot/src/iot_sup.erl

@@ -28,6 +28,15 @@ init([]) ->
     SupFlags = #{strategy => one_for_one, intensity => 10, period => 36},
 
     Specs = [
+        #{
+            id => 'iot_event_stream_observer',
+            start => {'iot_event_stream_observer', start_link, []},
+            restart => permanent,
+            shutdown => 2000,
+            type => worker,
+            modules => ['iot_event_stream_observer']
+        },
+
         #{
             id => 'iot_name_server',
             start => {'iot_name_server', start_link, []},

apps/iot/src/tcp/tcp_channel.erl

@@ -139,8 +139,11 @@ handle_info({tcp, Socket, <<?PACKET_CAST, CastBin/binary>>}, State = #state{sock
             iot_host:handle(HostPid, {data, Data});
         #event{} = Event ->
             iot_host:handle(HostPid, {event, Event});
+        #task_event_stream{task_id = TaskId, type = <<"close">>, stream = <<>>} ->
+            iot_event_stream_observer:stream_close(TaskId);
         #task_event_stream{task_id = TaskId, type = Type, stream = Stream} ->
-            lager:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream])
+            lager:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]),
+            iot_event_stream_observer:stream_data(TaskId, Type, Stream)
     end,
     {noreply, State};