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

fix efka_client

Browse Source
This commit is contained in:
anlicheng 2026-04-20 17:24:37 +08:00
parent db5a5e9071
commit 597fdbd958
1 changed files with 9 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/efka_client.erl
View File

@ -78,21 +78,20 @@ callback_mode() ->
handle_event_function. handle_event_function.
%% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia %% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia
handle_event(cast, {metric_data, RouteKey, Metric}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) ->
CastFrame = message_pb:encode_msg(#'CastFrame'{ CastFrame = message_pb:encode_msg(#'CastFrame'{
body = {data, #'Data'{route_key = RouteKey, metric = Metric}} body = {data, #'Data'{route_key = RouteKey, metric = Metric}}
}), }),
Packet = <<?FRAME_CAST, CastFrame/binary>>, Packet = <<?FRAME_CAST, CastFrame/binary>>,
send_packet(Socket, Packet), case StateName of
{keep_state, State}; ?STATE_ACTIVATED ->
send_packet(Socket, Packet);
handle_event(cast, {metric_data, RouteKey, Metric}, _, State) -> _ ->
Packet = message_pb:encode_msg(#'CastFrame'{ ok = cache_model:insert(Packet)
body = {data, #'Data'{route_key = RouteKey, metric = Metric}} end,
}),
ok = cache_model:insert(Packet),
{keep_state, State}; {keep_state, State};
%% Task的stream流，只做实时的
handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
logger:debug("[efka_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]), logger:debug("[efka_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
EventPacket = message_pb:encode_msg(#'CastFrame'{ EventPacket = message_pb:encode_msg(#'CastFrame'{
@ -109,9 +108,7 @@ handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED,
{keep_state, State}; {keep_state, State};
%% 其他情况下直接忽略 %% 其他情况下直接忽略
handle_event(cast, {task_event_stream, _TaskId, _Stream}, _, State = #state{}) -> handle_event(cast, _, _, State = #state{}) ->
{keep_state, State};
handle_event(cast, {close_task_event_stream, _TaskId, _Reason}, _, State = #state{}) ->
{keep_state, State}; {keep_state, State};
handle_event({call, From}, is_activated, ?STATE_ACTIVATED, State = #state{}) -> handle_event({call, From}, is_activated, ?STATE_ACTIVATED, State = #state{}) ->