anlicheng/ekfa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
anlicheng 2025-05-09 14:46:56 +08:00
parent c2eb509706
commit aca0dbc7d4
1 changed files with 33 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
apps/efka/src/efka_agent.erl
View File

@ -27,6 +27,7 @@
-define(STATE_DENIED, denied). -define(STATE_DENIED, denied).
-define(STATE_CONNECTING, connecting). -define(STATE_CONNECTING, connecting).
-define(STATE_AUTH, auth). -define(STATE_AUTH, auth).
%% 不能推送消息到服务，但是可以接受服务器的部分指令 %% 不能推送消息到服务，但是可以接受服务器的部分指令
-define(STATE_RESTRICTED, restricted). -define(STATE_RESTRICTED, restricted).
%% 激活状态下 %% 激活状态下
@ -117,7 +118,6 @@ handle_cast({event, ServiceId, EventType, Params}, State) ->
event_type = EventType, event_type = EventType,
params = Params params = Params
}), }),
safe_send(?METHOD_EVENT, EventPacket, State), safe_send(?METHOD_EVENT, EventPacket, State),
{noreply, State}; {noreply, State};
@ -191,42 +191,44 @@ handle_info({connect_reply, Reply}, State = #state{status = ?STATE_CONNECTING, t
end; end;
%% 收到auth回复 %% 收到auth回复
handle_info({auth_reply, {ok, ReplyBin}}, State = #state{status = ?STATE_AUTH, transport_pid = TransportPid}) when is_pid(TransportPid) -> handle_info({auth_reply, Reply}, State = #state{status = ?STATE_AUTH, transport_pid = TransportPid}) when is_pid(TransportPid) ->
#auth_reply{code = Code, message = Message} = message_pb:decode_msg(ReplyBin, auth_reply), case Reply of
case Code of {ok, ReplyBin} ->
0 -> #auth_reply{code = Code, message = Message} = message_pb:decode_msg(ReplyBin, auth_reply),
lager:debug("[efka_agent] auth failed, message: ~p", [Message]), case Code of
0 ->
lager:debug("[efka_agent] auth failed, message: ~p", [Message]),
%% 上传缓冲区里面的所有数据 %% 上传缓冲区里面的所有数据
CacheItems = cache_model:get_all_cache(), CacheItems = cache_model:get_all_cache(),
lists:foreach(fun(#cache{id = Id, method = Method, data = Packet}) -> lists:foreach(fun(#cache{id = Id, method = Method, data = Packet}) ->
efka_transport:send(TransportPid, Method, Packet), efka_transport:send(TransportPid, Method, Packet),
cache_model:delete(Id) cache_model:delete(Id)
end, CacheItems), end, CacheItems),
{noreply, State#state{status = ?STATE_ACTIVATED}}; {noreply, State#state{status = ?STATE_ACTIVATED}};
1 -> 1 ->
%% 主机在后台的授权未通过；此时agent不能推送数据给云端服务器，但是云端服务器可以推送命令给agent %% 主机在后台的授权未通过；此时agent不能推送数据给云端服务器，但是云端服务器可以推送命令给agent
%% socket的连接状态需要维持 %% socket的连接状态需要维持
lager:debug("[efka_agent] auth denied, message: ~p", [Message]), lager:debug("[efka_agent] auth denied, message: ~p", [Message]),
{noreply, State#state{status = ?STATE_RESTRICTED}}; {noreply, State#state{status = ?STATE_RESTRICTED}};
2 -> 2 ->
% 其他类型的错误，需要间隔时间重试 % 其他类型的错误，需要间隔时间重试
efka_logger:debug("[efka_agent] auth failed, message: ~p", [Message]), efka_logger:debug("[efka_agent] auth failed, message: ~p", [Message]),
efka_transport:stop(TransportPid), efka_transport:stop(TransportPid),
{noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}}; {noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}};
_ -> _ ->
% 其他类型的错误，需要间隔时间重试 % 其他类型的错误，需要间隔时间重试
lager:debug("[efka_agent] auth failed, invalid message"), lager:debug("[efka_agent] auth failed, invalid message"),
efka_transport:stop(TransportPid),
{noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}}
end;
{error, Reason} ->
lager:debug("[efka_agent] auth_request failed, error: ~p", [Reason]),
efka_transport:stop(TransportPid), efka_transport:stop(TransportPid),
{noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}} {noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}}
end; end;
handle_info({auth_reply, {error, Reason}}, State = #state{transport_pid = TransportPid, status = ?STATE_AUTH}) ->
lager:debug("[efka_agent] auth_request failed, error: ~p", [Reason]),
efka_transport:stop(TransportPid),
{noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}};
%% 云端服务器推送了消息 %% 云端服务器推送了消息
%% 激活消息 %% 激活消息