122 lines
4.2 KiB
Erlang
122 lines
4.2 KiB
Erlang
%%%-------------------------------------------------------------------
|
|
%%% @author anlicheng
|
|
%%% @copyright (C) 2026, <COMPANY>
|
|
%%% @doc
|
|
%%%
|
|
%%% @end
|
|
%%% Created : 20. 4月 2026
|
|
%%%-------------------------------------------------------------------
|
|
-module(docker_task_reporter).
|
|
-author("anlicheng").
|
|
|
|
-behaviour(gen_server).
|
|
|
|
%% API
|
|
-export([start_link/0]).
|
|
-export([stream/3, close/2]).
|
|
|
|
%% gen_server callbacks
|
|
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
|
|
|
|
-define(SERVER, ?MODULE).
|
|
-define(FLUSH_INTERVAL, 1000).
|
|
|
|
-record(state, {
|
|
pending = queue:new(),
|
|
flush_ref = undefined
|
|
}).
|
|
|
|
%%%===================================================================
|
|
%%% API
|
|
%%%===================================================================
|
|
|
|
-spec stream(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> ok.
|
|
stream(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
|
|
gen_server:cast(?SERVER, {stream, TaskId, Type, Stream}).
|
|
|
|
-spec close(TaskId :: integer(), Reason :: binary()) -> ok.
|
|
close(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) ->
|
|
gen_server:cast(?SERVER, {close, TaskId, Reason}).
|
|
|
|
-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
|
|
start_link() ->
|
|
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
|
|
|
|
%%%===================================================================
|
|
%%% gen_server callbacks
|
|
%%%===================================================================
|
|
|
|
-spec init(list()) -> {ok, #state{}}.
|
|
init([]) ->
|
|
{ok, #state{}}.
|
|
|
|
-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, ok, #state{}}.
|
|
handle_call(_Request, _From, State) ->
|
|
{reply, ok, State}.
|
|
|
|
-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
|
|
handle_cast({stream, TaskId, Type, Stream}, State0 = #state{pending = Pending0}) ->
|
|
Pending = queue:in({stream, TaskId, Type, Stream}, Pending0),
|
|
{noreply, ensure_flush_timer(State0#state{pending = Pending}, 0)};
|
|
handle_cast({close, TaskId, Reason}, State0 = #state{pending = Pending0}) ->
|
|
Pending = queue:in({close, TaskId, Reason}, Pending0),
|
|
{noreply, ensure_flush_timer(State0#state{pending = Pending}, 0)};
|
|
handle_cast(_Request, State) ->
|
|
{noreply, State}.
|
|
|
|
-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
|
|
handle_info(flush, State0 = #state{}) ->
|
|
State1 = State0#state{flush_ref = undefined},
|
|
{noreply, flush_pending(State1)};
|
|
handle_info(_Info, State) ->
|
|
{noreply, State}.
|
|
|
|
-spec terminate(term(), #state{}) -> ok.
|
|
terminate(_Reason, _State) ->
|
|
ok.
|
|
|
|
-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
|
|
code_change(_OldVsn, State, _Extra) ->
|
|
{ok, State}.
|
|
|
|
%%%===================================================================
|
|
%%% Internal functions
|
|
%%%===================================================================
|
|
|
|
-spec ensure_flush_timer(#state{}, non_neg_integer()) -> #state{}.
|
|
ensure_flush_timer(State = #state{flush_ref = undefined}, Delay) when is_integer(Delay), Delay >= 0 ->
|
|
Ref = erlang:send_after(Delay, self(), flush),
|
|
State#state{flush_ref = Ref};
|
|
ensure_flush_timer(State = #state{}, _Delay) ->
|
|
State.
|
|
|
|
-spec flush_pending(#state{}) -> #state{}.
|
|
flush_pending(State = #state{pending = Pending0}) ->
|
|
case queue:out(Pending0) of
|
|
{empty, _} ->
|
|
State;
|
|
{{value, Event}, Pending1} ->
|
|
case send_event(Event) of
|
|
ok ->
|
|
flush_pending(State#state{pending = Pending1});
|
|
not_ready ->
|
|
ensure_flush_timer(State#state{pending = Pending0}, ?FLUSH_INTERVAL)
|
|
end
|
|
end.
|
|
|
|
-spec send_event({stream, integer(), binary(), binary()} | {close, integer(), binary()}) -> ok | not_ready.
|
|
send_event({stream, TaskId, Type, Stream}) ->
|
|
maybe_send(fun() -> efka_client:task_event_stream(TaskId, Type, Stream) end);
|
|
send_event({close, TaskId, Reason}) ->
|
|
maybe_send(fun() -> efka_client:close_task_event_stream(TaskId, Reason) end).
|
|
|
|
-spec maybe_send(fun(() -> any())) -> ok | not_ready.
|
|
maybe_send(SendFun) ->
|
|
case catch efka_client:is_activated() of
|
|
true ->
|
|
_ = catch SendFun(),
|
|
ok;
|
|
_ ->
|
|
not_ready
|
|
end.
|