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 16:17:45 +08:00
parent 15ea812aab
commit 2893585ea4
1 changed files with 19 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/efka_client.erl
View File

@ -141,13 +141,13 @@ handle_event(info, flush_cache, _, State) ->
%% 处理收到的ssl消息 %% 处理收到的ssl消息
handle_event(info, {ssl, Socket, <<?FRAME_RESPONSE, PacketBin/binary>>}, _, State = #state{socket = Socket}) -> handle_event(info, {ssl, Socket, <<?FRAME_RESPONSE, PacketBin/binary>>}, _, State = #state{socket = Socket}) ->
ResponseFrame = message_pb:decode_msg(PacketBin, 'ResponseFrame'), ResponseFrame = message_pb:decode_msg(PacketBin, 'ResponseFrame'),
{keep_state, State, [{next_event, internal, ResponseFrame}]}; {keep_state, State, [{next_event, internal, {decoded_response, ResponseFrame}}]};
handle_event(info, {ssl, Socket, <<?FRAME_REQUEST, PacketBin/binary>>}, _, State = #state{socket = Socket}) -> handle_event(info, {ssl, Socket, <<?FRAME_REQUEST, PacketBin/binary>>}, _, State = #state{socket = Socket}) ->
RequestFrame = message_pb:decode_msg(PacketBin, 'RequestFrame'), RequestFrame = message_pb:decode_msg(PacketBin, 'RequestFrame'),
{keep_state, State, [{next_event, internal, RequestFrame}]}; {keep_state, State, [{next_event, internal, {decoded_request, RequestFrame}}]};
handle_event(info, {ssl, Socket, <<?FRAME_CAST, PacketBin/binary>>}, _, State = #state{socket = Socket}) -> handle_event(info, {ssl, Socket, <<?FRAME_CAST, PacketBin/binary>>}, _, State = #state{socket = Socket}) ->
CastFrame = message_pb:decode_msg(PacketBin, 'CastFrame'), CastFrame = message_pb:decode_msg(PacketBin, 'CastFrame'),
{keep_state, State, [{next_event, internal, CastFrame}]}; {keep_state, State, [{next_event, internal, {decoded_cast, CastFrame}}]};
handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) -> handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) ->
logger:debug("[efka_client] ssl error: ~p", [Reason]), logger:debug("[efka_client] ssl error: ~p", [Reason]),
disconnect(Socket), disconnect(Socket),
@ -160,7 +160,8 @@ handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
%%% 处理内部消息，ssl收到的消息会解析成protobuf的消息格式，并按照internal类型处理 %%% 处理内部消息，ssl收到的消息会解析成protobuf的消息格式，并按照internal类型处理
%% 微服务部署 %% 微服务部署
handle_event(internal, #'RequestFrame'{packet_id = PacketId, body = {container_request, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = {container_request, Request}}},
?STATE_ACTIVATED, State = #state{socket = Socket}) ->
case handle_container_request(Request) of case handle_container_request(Request) of
ok -> ok ->
send_rpc_result_reply(Socket, PacketId, <<"ok">>); send_rpc_result_reply(Socket, PacketId, <<"ok">>);
@ -170,14 +171,19 @@ handle_event(internal, #'RequestFrame'{packet_id = PacketId, body = {container_r
send_rpc_error_reply(Socket, PacketId, Reason) send_rpc_error_reply(Socket, PacketId, Reason)
end, end,
{keep_state, State}; {keep_state, State};
handle_event(internal, #'RequestFrame'{packet_id = PacketId, body = _Body}, ?STATE_RESTRICTED, State = #state{socket = Socket}) -> handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = _Body}},
?STATE_RESTRICTED, State = #state{socket = Socket}) ->
send_rpc_error_reply(Socket, PacketId, <<"agent restricted">>), send_rpc_error_reply(Socket, PacketId, <<"agent restricted">>),
{keep_state, State}; {keep_state, State};
handle_event(internal, #'RequestFrame'{packet_id = PacketId, body = _Body}, _StateName, State = #state{socket = Socket}) -> handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = _Body}},
_StateName, State = #state{socket = Socket}) ->
send_rpc_error_reply(Socket, PacketId, <<"agent state invalid">>), send_rpc_error_reply(Socket, PacketId, <<"agent state invalid">>),
{keep_state, State}; {keep_state, State};
handle_event(internal, #'ResponseFrame'{packet_id = 1, body = {auth_reply, #'AuthReply'{code = Code, payload = Message}}}, ?STATE_AUTH, State = #state{socket = Socket}) -> handle_event(internal, {decoded_response, #'ResponseFrame'{
packet_id = 1,
body = {auth_reply, #'AuthReply'{code = Code, payload = Message}}
}}, ?STATE_AUTH, State = #state{socket = Socket}) ->
case Code of case Code of
0 -> 0 ->
logger:debug("[efka_client] auth success, message: ~p", [Message]), logger:debug("[efka_client] auth success, message: ~p", [Message]),
@ -198,7 +204,9 @@ handle_event(internal, #'ResponseFrame'{packet_id = 1, body = {auth_reply, #'Aut
end; end;
%% 处理命令 %% 处理命令
handle_event(internal, #'CastFrame'{body = {command, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}}}, StateName, State = #state{socket = Socket}) -> handle_event(internal, {decoded_cast, #'CastFrame'{
body = {command, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}}
}}, StateName, State = #state{socket = Socket}) ->
Auth = binary_to_integer(Auth0), Auth = binary_to_integer(Auth0),
case {Auth, StateName} of case {Auth, StateName} of
{1, ?STATE_ACTIVATED} -> {1, ?STATE_ACTIVATED} ->
@ -212,7 +220,9 @@ handle_event(internal, #'CastFrame'{body = {command, #'Command'{command_type = ?
end; end;
%% 处理Pub/Sub机制 %% 处理Pub/Sub机制
handle_event(internal, #'CastFrame'{body = {pub, #'Pub'{topic = Topic, qos = Qos, content = Content}}}, ?STATE_ACTIVATED, State) -> handle_event(internal, {decoded_cast, #'CastFrame'{
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};