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 2026-04-20 19:25:44 +08:00
parent 12f485f077
commit 2bda5f6d72
1 changed files with 9 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/transport/efka_client.erl
View File

@ -129,8 +129,6 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State =
schedule_reconnect(), schedule_reconnect(),
{keep_state, State#state{socket = undefined}} {keep_state, State#state{socket = undefined}}
end; end;
handle_event(info, {timeout, _, create_transport}, _, State) ->
{keep_state, State};
handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = Socket}) -> handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = Socket}) ->
logger:debug("[efka_client] auth request timeout"), logger:debug("[efka_client] auth request timeout"),
@ -174,9 +172,8 @@ handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
%% 微服务部署 %% 微服务部署
handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = {container_request, Request}}}, handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = {container_request, Request}}},
?STATE_ACTIVATED, State = #state{socket = Socket}) -> ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Result = docker_container_service:handle_request(Request), case docker_container_service:handle_request(Request) of
case Result of
ok -> ok ->
send_result_reply(Socket, PacketId, <<"ok">>); send_result_reply(Socket, PacketId, <<"ok">>);
{ok, Reply} -> {ok, Reply} ->
@ -186,11 +183,11 @@ handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, b
end, end,
{keep_state, State}; {keep_state, State};
handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = _Body}}, handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = _Body}},
?STATE_RESTRICTED, State = #state{socket = Socket}) -> ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
send_error_reply(Socket, PacketId, <<"agent restricted">>), send_error_reply(Socket, PacketId, <<"agent restricted">>),
{keep_state, State}; {keep_state, State};
handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = _Body}}, handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = _Body}},
_StateName, State = #state{socket = Socket}) -> _StateName, State = #state{socket = Socket}) ->
send_error_reply(Socket, PacketId, <<"agent state invalid">>), send_error_reply(Socket, PacketId, <<"agent state invalid">>),
{keep_state, State}; {keep_state, State};
@ -214,9 +211,9 @@ handle_event(internal, {decoded_reply, ReplyFrame}, StateName, State) ->
{keep_state, State}; {keep_state, State};
%% 处理命令 %% 处理命令
handle_event(internal, {decoded_cast, #'CastFrame'{ handle_event(internal, {decoded_cast, #'CastFrame'{body = {command, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}}}},
body = {command, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}} StateName, State = #state{socket = Socket, next_packet_id = PacketId}) ->
}}, StateName, State = #state{socket = Socket, next_packet_id = PacketId}) ->
Auth = binary_to_integer(Auth0), Auth = binary_to_integer(Auth0),
case {Auth, StateName} of case {Auth, StateName} of
{1, ?STATE_ACTIVATED} -> {1, ?STATE_ACTIVATED} ->
@ -231,9 +228,7 @@ handle_event(internal, {decoded_cast, #'CastFrame'{
end; end;
%% 处理Pub/Sub机制 %% 处理Pub/Sub机制
handle_event(internal, {decoded_cast, #'CastFrame'{ handle_event(internal, {decoded_cast, #'CastFrame'{body = {pub, #'Pub'{topic = Topic, qos = Qos, content = Content}}}}, ?STATE_ACTIVATED, State) ->
body = {pub, #'Pub'{topic = Topic, qos = Qos, content = Content}}
}}, ?STATE_ACTIVATED, State) ->
logger:debug("[efka_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]), logger:debug("[efka_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
efka_subscription:publish(Topic, Qos, Content), efka_subscription:publish(Topic, Qos, Content),
{keep_state, State}; {keep_state, State};