fix

apps/efka/src/efka_agent.erl

@@ -106,7 +106,7 @@ handle_event(info, {auth_reply, {ok, #auth_reply{code = -1, message = Message}}}
     %% socket的连接状态需要维持
     efka_logger:debug("[efka_agent] auth denied, message: ~p", [Message]),
     {next_state, ?STATE_RESTRICTED, State};
-handle_event(info, {auth_reply,  {ok, #auth_reply{code = -2, message = Message}}}, ?STATE_AUTH, State = #state{transport_pid = TransportPid}) ->
+handle_event(info, {auth_reply, {ok, #auth_reply{code = -2, message = Message}}}, ?STATE_AUTH, State = #state{transport_pid = TransportPid}) ->
     %% 其他类型的错误，需要间隔时间重试
     efka_logger:debug("[efka_agent] auth failed, message: ~p", [Message]),
     efka_transport:stop(TransportPid),

apps/efka/src/efka_agent1.erl

@@ -0,0 +1,190 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 06. 5月 2025 00:01
+%%%-------------------------------------------------------------------
+-module(efka_agent1).
+-author("anlicheng").
+-include("message_pb.hrl").
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+-define(SERVER, ?MODULE).
+
+%% 标记当前agent的状态，只有在 activated 状态下才可以正常的发送数据
+-define(STATE_DENIED, denied).
+-define(STATE_CONNECTING, connecting).
+-define(STATE_AUTH, auth).
+%% 不能推送消息到服务，但是可以接受服务器的部分指令
+-define(STATE_RESTRICTED, restricted).
+%% 激活状态下
+-define(STATE_ACTIVATED, activated).
+
+-record(state, {
+    transport_pid :: undefined | pid(),
+    status = ?STATE_DENIED
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+%% @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([]) ->
+    erlang:process_flag(trap_exit, true),
+    erlang:start_timer(0, self(), create_transport),
+    {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(_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(_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({timeout, _, create_transport}, State = #state{status = ?STATE_DENIED}) ->
+    {ok, Props} = application:get_env(efka, tls_server),
+    Host = proplists:get_value(host, Props),
+    Port = proplists:get_value(port, Props),
+    {ok, TransportPid} = efka_transport:start_link(self(), Host, Port),
+    efka_transport:connect(TransportPid),
+
+    {noreply, State#state{status = ?STATE_CONNECTING, transport_pid = TransportPid}};
+
+handle_info({connect_reply, Reply}, State = #state{status = ?STATE_CONNECTING, transport_pid = TransportPid}) ->
+    case Reply of
+        ok ->
+            efka_transport:auth_request(TransportPid, 5000),
+            {noreply, State#state{status = ?STATE_AUTH}};
+        {error, Reason} ->
+            efka_logger:debug("[efka_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]),
+            efka_transport:stop(TransportPid),
+            {noreply, ?STATE_DENIED, State#state{status = ?STATE_DENIED}}
+    end;
+
+handle_info({auth_reply, {ok, #auth_reply{code = 1, message = Message, repository_url = RepositoryUrl}}}, State = #state{status = ?STATE_AUTH}) ->
+    efka_logger:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]),
+    {noreply, State#state{status = ?STATE_ACTIVATED}};
+handle_info({auth_reply, {ok, #auth_reply{code = -1, message = Message}}}, State = #state{status = ?STATE_AUTH}) ->
+    %% 主机在后台的授权未通过；此时agent不能推送数据给云端服务器，但是云端服务器可以推送命令给agent
+    %% socket的连接状态需要维持
+    efka_logger:debug("[efka_agent] auth denied, message: ~p", [Message]),
+    {noreply, State#state{status = ?STATE_RESTRICTED}};
+handle_info({auth_reply, {ok, #auth_reply{code = -2, message = Message}}}, State = #state{transport_pid = TransportPid, status = ?STATE_AUTH}) ->
+    %% 其他类型的错误，需要间隔时间重试
+    efka_logger:debug("[efka_agent] auth failed, message: ~p", [Message]),
+    efka_transport:stop(TransportPid),
+    {noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}};
+handle_info({auth_reply, {error, Reason}}, State = #state{transport_pid = TransportPid, status = ?STATE_AUTH}) ->
+    efka_logger:debug("[efka_agent] auth_request failed, error: ~p", [Reason]),
+    efka_transport:stop(TransportPid),
+    {noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}};
+
+
+%% 云端服务器推送了消息
+%% 激活消息
+
+%% TODO
+handle_info({server_push_message, <<8:8, ActivatePush>>}, State = #state{transport_pid = TransportPid, status = Status}) ->
+    #activate_push{auth = Auth} = message_pb:decode_msg(ActivatePush, activate_push),
+    case {Auth, Status} of
+        {true, ?STATE_ACTIVATED} ->
+            {noreply, State};
+        {true, ?STATE_DENIED} ->
+            %% 重新激活, 需要重新校验
+            efka_transport:auth_request(TransportPid, 5000),
+            {noreply, State#state{status = ?STATE_AUTH}};
+        {false, _} ->
+            %% 这个时候的主机应该是受限制的状态，不允许发送消息；但是能够接受服务器推送的消息
+            {noreply, State#state{status = ?STATE_RESTRICTED}}
+    end;
+
+%% 收到需要回复的指令
+handle_info({server_push_message, PacketId, <<16:8, Directive>>}, State = #state{transport_pid = TransportPid, status = ?STATE_ACTIVATED}) ->
+    #topic_message{topic = Topic, content = Content} = message_pb:decode_msg(Directive, directive),
+    efka_logger:debug("[efka_agent] get directive with packet_id: ~p, to device_uuid: ~p, content: ~p", [PacketId, Topic, Content]),
+    %% 消息发送到订阅系统
+    case PacketId > 0 of
+        true ->
+            CallbackFun = fun(Response) -> is_process_alive(TransportPid) andalso efka_transport:response(TransportPid, PacketId, Response) end,
+            efka_subscription:publish(PacketId, Topic, Content, CallbackFun);
+        false ->
+            efka_subscription:publish(Topic, Content)
+    end,
+
+    {noreply, State};
+
+%% transport进程退出
+handle_info({'EXIT', TransportPid, Reason}, State = #state{transport_pid = TransportPid}) ->
+    efka_logger:debug("[efka_agent] transport pid: ~p, exit with reason: ~p", [TransportPid, Reason]),
+    erlang:start_timer(500000, self(), create_transport),
+    {noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}};
+
+handle_info(_Info, State = #state{}) ->
+    {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
+%%%===================================================================

apps/efka/src/efka_client.erl

@@ -46,7 +46,7 @@
 %% API
 -export([start_link/3]).
 -export([device_offline/1, device_online/1]).
--export([send_metric_data/2, invoke_service/3, send_log/1, request_metric/0, request_param/0, send_event/2]).
+-export([send_metric_data/2, invoke_service/3, send_log/1, request_metric/0, request_param/0, send_event/2, controller_process/1]).
 
 -export([test/0]).
 
@@ -66,6 +66,7 @@
 test() ->
     start_link(<<"test">>, "localhost", 18080).
 
+-spec controller_process(ControllerPid :: pid()) -> ok.
 controller_process(ControllerPid) when is_pid(ControllerPid) ->
     gen_server:call(?MODULE, {controller_process, ControllerPid}).
 
@@ -256,7 +257,7 @@ handle_info({tcp, Socket, <<PacketId:32, ?PACKET_TYPE_RESPONSE:8, Message/binary
     end;
 
 %% 收到efka推送的参数设置
-handle_info({tcp, Socket, <<PacketId:32, ?PACKET_TYPE_PUSH_PARAM:8, Params/binary>>}, State = #state{socket = Socket, controller_process = ControllerPid}) when is_map(Params) ->
+handle_info({tcp, Socket, <<PacketId:32, ?PACKET_TYPE_PUSH_PARAM:8, Params/binary>>}, State = #state{socket = Socket, controller_process = ControllerPid}) ->
     Message = case is_pid(ControllerPid) andalso is_process_alive(ControllerPid) of
                   true ->
                       Ref = make_ref(),
@@ -278,7 +279,7 @@ handle_info({tcp, Socket, <<PacketId:32, ?PACKET_TYPE_PUSH_PARAM:8, Params/binar
     {noreply, State};
 
 %% 收到efka推送的采集项消息
-handle_info({tcp, <<PacketId:32, ?PACKET_TYPE_PUSH_METRIC:8, Metrics/binary>>}, State = #state{socket = Socket, controller_process = ControllerPid}) when is_list(Metrics) ->
+handle_info({tcp, <<PacketId:32, ?PACKET_TYPE_PUSH_METRIC:8, Metrics/binary>>}, State = #state{socket = Socket, controller_process = ControllerPid}) ->
     Message = case is_pid(ControllerPid) andalso is_process_alive(ControllerPid) of
                   true ->
                       Ref = make_ref(),