ekfa/apps/efka/src/efka_tcp_channel.erl

%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2025, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 30. 4月 2025 09:22
%%%-------------------------------------------------------------------
-module(efka_tcp_channel).
-author("anlicheng").

-behaviour(gen_server).

%% API
-export([start_link/1]).
-export([push_metric/2, push_param/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(PACKET_REGISTER, 16).
%% 上传数据
-define(PACKET_METRIC_DATA, 3).
%% 消息响应
-define(PACKET_RESPONSE, 7).
%% efka下发给微服务参数
-define(PACKET_PUSH_PARAM, 5).
%% efka下发给微服务采集项
-define(PACKET_PUSH_METRIC, 6).

%% 微服务给efka发送log消息
-define(PACKET_LOG, 9).

%% 微服务从efka获取自身的采集项
-define(PACKET_REQUEST_METRIC, 10).
%% 微服务从efka获取自身的参数
-define(PACKET_REQUEST_PARAM, 12).
%% 微服务事件上报
-define(PACKET_EVENT, 15).
-define(PACKET_AI_EVENT, 16).

-record(state, {
    packet_id = 1,
    socket :: gen_tcp:socket(),
    service_id :: undefined | binary(),
    service_pid :: undefined | pid(),
    is_registered = false :: boolean(),

    %% 请求响应的对应关系, #{packet_id => {ReceiverPid, Ref}}
    inflight = #{}
}).

%%%===================================================================
%%% API
%%%===================================================================

-spec push_param(ChannelPid :: pid(), Params :: binary()) -> {ok, Ref :: reference()}.
push_param(ChannelPid, Params) when is_pid(ChannelPid), is_binary(Params) ->
    Ref = make_ref(),
    gen_server:cast(ChannelPid, {push_param, Ref, self(), Params}),
    {ok, Ref}.

-spec push_metric(ChannelPid :: pid(), Metrics :: binary()) -> {ok, Ref :: reference()}.
push_metric(ChannelPid, Metrics) when is_pid(ChannelPid), is_binary(Metrics) ->
    Ref = make_ref(),
    gen_server:cast(ChannelPid, {push_metric, Ref, self(), Metrics}),
    {ok, Ref}.

%% @doc Spawns the server and registers the local name (unique)
-spec(start_link(Socket :: gen_tcp:socket()) ->
    {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link(Socket) ->
    gen_server:start_link(?MODULE, [Socket], []).

%%%===================================================================
%%% 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([Socket]) ->
    ok = inet:setopts(Socket, [{active, true}]),
    lager:debug("[tcp_channel] get new socket: ~p", [Socket]),
    {ok, #state{socket = Socket}}.

%% @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(_Request, _From, State = #state{}) ->
    {reply, ok, State}.

%% @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({push_param, Ref, ReceiverPid, Params}, State = #state{socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
    ok = gen_tcp:send(Socket, <<PacketId:32, ?PACKET_PUSH_PARAM:8, Params/binary>>),
    {noreply, State#state{inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};

%% 推送采集项目
handle_cast({push_metric, Ref, ReceiverPid, Metrics}, State = #state{socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
    ok = gen_tcp:send(Socket, <<PacketId:32, ?PACKET_PUSH_METRIC:8, Metrics/binary>>),
    {noreply, State#state{inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};

handle_cast(_Request, State = #state{}) ->
    {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({tcp, Socket, <<PacketId:32, ?PACKET_REGISTER:8, ServiceId/binary>>}, State = #state{socket = Socket}) ->
    case efka_micro_service:get_pid(ServiceId) of
        undefined ->
            ok = gen_tcp:send(Socket, <<PacketId:32, ?PACKET_RESPONSE:8, 0:8, "service not running">>);
        Pid when is_pid(Pid) ->
            case efka_micro_service:attach_channel(Pid, self()) of
                ok ->
                    ok = gen_tcp:send(Socket, <<PacketId:32, ?PACKET_RESPONSE:8, 1:8>>),
                    {noreply, State#state{service_id = ServiceId, service_pid = Pid, is_registered = true}};
                {error, Reason} ->
                    ok = gen_tcp:send(Socket, <<PacketId:32, ?PACKET_RESPONSE:8, 10:8, Reason/binary>>),
                    {stop, normal, State}
            end
    end;

%% 请求参数
handle_info({tcp, Socket, <<PacketId:32, ?PACKET_REQUEST_PARAM:8>>}, State = #state{socket = Socket, service_pid = ServicePid, is_registered = true}) ->
    {ok, Args} = efka_micro_service:request_arguments(ServicePid),
    ok = gen_tcp:send(Socket, <<PacketId:32, ?PACKET_RESPONSE:8, Args/binary>>),
    {noreply, State};

%% 请求采集项目
handle_info({tcp, Socket, <<PacketId:32, ?PACKET_REQUEST_METRIC:8>>}, State = #state{socket = Socket, service_pid = ServicePid, is_registered = true}) ->
    {ok, Metrics} = efka_micro_service:request_metrics(ServicePid),
    ok = gen_tcp:send(Socket, <<PacketId:32, ?PACKET_RESPONSE:8, Metrics/binary>>),
    {noreply, State};

%% 数据项
handle_info({tcp, Socket, <<0:32, ?PACKET_METRIC_DATA:8, Data/binary>>}, State = #state{socket = Socket, service_pid = ServicePid, is_registered = true}) ->
    case catch jiffy:decode(Data, [return_maps]) of
        #{<<"fields">> := Fields, <<"tag">> := Tags} ->
            efka_micro_service:metric_data(ServicePid, Fields, Tags);
        _ ->
            ok
    end,
    {noreply, State};

%% 远程日志
handle_info({tcp, Socket, <<0:32, ?PACKET_LOG:8, Log/binary>>}, State = #state{socket = Socket, service_pid = ServicePid, is_registered = true}) ->
    efka_micro_service:send_log(ServicePid, Log),
    {noreply, State};

%% Event事件
handle_info({tcp, Socket, <<0:32, ?PACKET_EVENT:8, EventData/binary>>}, State = #state{socket = Socket, service_pid = ServicePid, is_registered = true}) ->
    case catch jiffy:decode(EventData, [return_maps]) of
        #{<<"event_type">> := EventType, <<"params">> := Params} ->
            efka_micro_service:send_event(ServicePid, EventType, Params);
        _ ->
            ok
    end,
    {noreply, State};

%% AIEvent事件
handle_info({tcp, Socket, <<0:32, ?PACKET_AI_EVENT:8, EventData/binary>>}, State = #state{socket = Socket, service_pid = ServicePid, is_registered = true}) ->
    case catch jiffy:decode(EventData, [return_maps]) of
        #{<<"event_type">> := EventType, <<"params">> := Params} ->
            efka_micro_service:send_ai_event(ServicePid, EventType, Params);
        _ ->
            ok
    end,
    {noreply, State};

%% 收到端上的响应
handle_info({tcp, Socket, <<PacketId:32, ?PACKET_RESPONSE:8, Response/binary>>}, State = #state{socket = Socket, inflight = Inflight}) ->
    case maps:take(PacketId, Inflight) of
        error ->
            lager:warning("[tcp_channel] get unknown publish response message: ~p, packet_id: ~p", [Response, PacketId]),
            {noreply, State};
        {{ReceiverPid, Ref}, NInflight} ->
            case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
                true ->
                    case Response of
                        <<1:8, Result/binary>> ->
                            ReceiverPid ! {channel_reply, Ref, {ok, Result}};
                        <<0:8, Error/binary>> ->
                            ReceiverPid ! {channel_reply, Ref, {error, Error}}
                    end;
                false ->
                    lager:warning("[tcp_channel] get publish response message: ~p, packet_id: ~p, but receiver_pid is deaded", [Response, PacketId])
            end,
            {noreply, State#state{inflight = NInflight}}
    end;

handle_info(Info, State = #state{}) ->
    lager:debug("[tcp_channel] get info: ~p", [Info]),
    {noreply, State}.

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