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 15:18:59 +08:00
parent 9bdf5fcac2
commit 0a775d7d65
1 changed files with 13 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
src/efka_client.erl
View File

@ -74,7 +74,7 @@ handle_event(cast, {metric_data, RouteKey, Metric}, ?STATE_ACTIVATED, State = #s
Packet = message_pb:encode_msg(#'CastFrame'{ Packet = message_pb:encode_msg(#'CastFrame'{
body = {data, #'Data'{route_key = RouteKey, metric = Metric}} body = {data, #'Data'{route_key = RouteKey, metric = Metric}}
}), }),
send_packet(Socket, Packet), send_packet(Socket, <<?FRAME_CAST, Packet/binary>>),
{keep_state, State}; {keep_state, State};
handle_event(cast, {metric_data, RouteKey, Metric}, _, State) -> handle_event(cast, {metric_data, RouteKey, Metric}, _, State) ->
@ -89,14 +89,14 @@ handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED,
EventPacket = message_pb:encode_msg(#'CastFrame'{ EventPacket = message_pb:encode_msg(#'CastFrame'{
body = {event_stream, #'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream}} body = {event_stream, #'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream}}
}), }),
send_packet(Socket, EventPacket), send_packet(Socket, <<?FRAME_CAST, EventPacket/binary>>),
{keep_state, State}; {keep_state, State};
handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
EventPacket = message_pb:encode_msg(#'CastFrame'{ EventPacket = message_pb:encode_msg(#'CastFrame'{
body = {event_stream, #'TaskEventStream'{task_id = TaskId, type = <<"close">>, stream = Reason}} body = {event_stream, #'TaskEventStream'{task_id = TaskId, type = <<"close">>, stream = Reason}}
}), }),
send_packet(Socket, EventPacket), send_packet(Socket, <<?FRAME_CAST, EventPacket/binary>>),
{keep_state, State}; {keep_state, State};
%% 其他情况下直接忽略 %% 其他情况下直接忽略
@ -139,9 +139,10 @@ handle_event(info, flush_cache, _, State) ->
%% 处理收到的ssl消息 %% 处理收到的ssl消息
handle_event(info, {ssl, Socket, PacketBin}, ?STATE_AUTH, State = #state{socket = Socket}) -> handle_event(info, {ssl, Socket, <<?FRAME_RESPONSE, PacketBin/binary>>}, ?STATE_AUTH, State = #state{socket = Socket}) ->
#'ResponseFrame'{packet_id = 1, body = {auth_reply, #'AuthReply'{code = Code, payload = Message}}} = #'ResponseFrame'{packet_id = 1, body = {auth_reply, #'AuthReply'{code = Code, payload = Message}}} =
message_pb:decode_msg(PacketBin, 'ResponseFrame'), message_pb:decode_msg(PacketBin, 'ResponseFrame'),
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]),
@ -161,28 +162,19 @@ handle_event(info, {ssl, Socket, PacketBin}, ?STATE_AUTH, State = #state{socket
{next_state, ?STATE_DENIED, State#state{socket = undefined}} {next_state, ?STATE_DENIED, State#state{socket = undefined}}
end; end;
handle_event(info, {ssl, Socket, <<8, _/binary>> = PacketBin}, StateName, State = #state{socket = Socket}) handle_event(info, {ssl, Socket, <<?FRAME_REQUEST, PacketBin/binary>>}, StateName, State = #state{socket = Socket})
when StateName =:= ?STATE_ACTIVATED; StateName =:= ?STATE_RESTRICTED -> when StateName =:= ?STATE_ACTIVATED; StateName =:= ?STATE_RESTRICTED ->
#'RequestFrame'{packet_id = PacketId, body = Body} = message_pb:decode_msg(PacketBin, 'RequestFrame'), #'RequestFrame'{packet_id = PacketId, body = Body} = message_pb:decode_msg(PacketBin, 'RequestFrame'),
true = is_integer(PacketId) andalso PacketId > 0,
case Body of case Body of
{rpc_request, Request} -> {rpc_request, Request} ->
{keep_state, State, [{next_event, info, {server_rpc, PacketId, Request}}]}; {keep_state, State, [{next_event, info, {server_rpc, PacketId, Request}}]};
{container_request, Request} -> {container_request, Request} ->
{keep_state, State, [{next_event, info, {container_request, PacketId, Request}}]} {keep_state, State, [{next_event, info, {container_request, PacketId, Request}}]}
end; end;
handle_event(info, {ssl, Socket, <<10, _/binary>> = PacketBin}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
#'CastFrame'{body = {pub, Pub}} = message_pb:decode_msg(PacketBin, 'CastFrame'), handle_event(info, {ssl, Socket, <<?FRAME_CAST, PacketBin/binary>>}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
{keep_state, State, [{next_event, info, {server_cast, Pub}}]}; CastFrame = message_pb:decode_msg(PacketBin, 'CastFrame'),
handle_event(info, {ssl, Socket, <<18, _/binary>> = PacketBin}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> {keep_state, State, [{next_event, info, {server_cast, CastFrame}}]};
#'CastFrame'{body = {command, Command}} = message_pb:decode_msg(PacketBin, 'CastFrame'),
{keep_state, State, [{next_event, info, {server_cast, Command}}]};
handle_event(info, {ssl, Socket, <<10, _/binary>> = PacketBin}, ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
#'CastFrame'{body = {pub, Pub}} = message_pb:decode_msg(PacketBin, 'CastFrame'),
{keep_state, State, [{next_event, info, {server_cast, Pub}}]};
handle_event(info, {ssl, Socket, <<18, _/binary>> = PacketBin}, ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
#'CastFrame'{body = {command, Command}} = message_pb:decode_msg(PacketBin, 'CastFrame'),
{keep_state, State, [{next_event, info, {server_cast, Command}}]};
%% 微服务部署 %% 微服务部署
handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = _All}}}}, handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = _All}}}},
@ -396,7 +388,7 @@ handle_event(info, {server_rpc, PacketId, _Request}, ?STATE_RESTRICTED, State =
{keep_state, State}; {keep_state, State};
%% 处理命令 %% 处理命令
handle_event(info, {server_cast, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}}, StateName, handle_event(info, {server_cast, {command, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}}}, StateName,
State = #state{socket = Socket}) -> State = #state{socket = Socket}) ->
Auth = binary_to_integer(Auth0), Auth = binary_to_integer(Auth0),
case {Auth, StateName} of case {Auth, StateName} of
@ -411,7 +403,7 @@ handle_event(info, {server_cast, #'Command'{command_type = ?COMMAND_AUTH, comman
end; end;
%% 处理Pub/Sub机制 %% 处理Pub/Sub机制
handle_event(info, {server_cast, #'Pub'{topic = Topic, qos = Qos, content = Content}}, ?STATE_ACTIVATED, State) -> handle_event(info, {server_cast, {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};