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

simple efka

Browse Source
This commit is contained in:
anlicheng 2025-05-08 11:08:26 +08:00
parent f789e48d51
commit 4eaaed1904
2 changed files with 5 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
apps/efka/include/efka.hrl
View File

@ -34,8 +34,7 @@
%% 部署微服务 %% 部署微服务
-define(METHOD_DEPLOY, 16#10). -define(METHOD_DEPLOY, 16#10).
-define(METHOD_PRAMAS, 16#11). -define(METHOD_CONFIG, 16#11).
-define(METHOD_METRICS, 16#12).
%% 事件类型 %% 事件类型
-define(EVENT_DEVICE, 16#01). -define(EVENT_DEVICE, 16#01).

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

@ -259,9 +259,9 @@ handle_info({server_push_message, PacketId, <<?METHOD_DEPLOY:8, DeployBin/binary
{noreply, State}; {noreply, State};
%% 参数项目 %% config.json配置信息
handle_info({server_push_message, PacketId, <<?METHOD_PRAMAS:8, ParamsBin/binary>>}, State = #state{transport_pid = TransportPid, inflight = Inflight}) -> handle_info({server_push_message, PacketId, <<?METHOD_CONFIG:8, ParamsBin/binary>>}, State = #state{transport_pid = TransportPid, inflight = Inflight}) ->
#service_params{service_id = ServiceId, params = Params, timeout = Timeout} = message_pb:decode_msg(ParamsBin, service_params), #service_config{service_id = ServiceId, config_json = ConfigJson, timeout = Timeout} = message_pb:decode_msg(ParamsBin, service_params),
case efka_micro_service:get_pid(ServiceId) of case efka_micro_service:get_pid(ServiceId) of
undefined -> undefined ->
@ -273,7 +273,7 @@ handle_info({server_push_message, PacketId, <<?METHOD_PRAMAS:8, ParamsBin/binary
{noreply, State}; {noreply, State};
ServicePid when is_pid(ServicePid) -> ServicePid when is_pid(ServicePid) ->
Ref = make_ref(), Ref = make_ref(),
efka_micro_service:push_params(ServicePid, Ref, Params), efka_micro_service:push_config(ServicePid, Ref, ConfigJson),
%% 处理超时逻辑 %% 处理超时逻辑
erlang:start_timer(Timeout * 1000, self(), {request_timeout, Ref}), erlang:start_timer(Timeout * 1000, self(), {request_timeout, Ref}),
@ -282,29 +282,6 @@ handle_info({server_push_message, PacketId, <<?METHOD_PRAMAS:8, ParamsBin/binary
end, end,
{noreply, State}; {noreply, State};
%% 采集项目
handle_info({server_push_message, PacketId, <<?METHOD_METRICS:8, MetricsBin/binary>>}, State = #state{inflight = Inflight}) ->
#service_metrics{service_id = ServiceId, metrics = Metrics, timeout = Timeout} = message_pb:decode_msg(MetricsBin, service_metrics),
case efka_micro_service:get_pid(ServiceId) of
undefined ->
Reply = #efka_response{
code = 0,
message = <<"service not run">>
},
safe_response(PacketId, message_pb:encode_msg(Reply), State),
{noreply, State};
ServicePid when is_pid(ServicePid) ->
Ref = make_ref(),
efka_micro_service:push_metrics(ServicePid, Ref, Metrics),
%% 处理超时逻辑
erlang:start_timer(Timeout * 1000, self(), {request_timeout, Ref}),
{noreply, State#state{inflight = maps:put(Ref, PacketId, Inflight)}}
end;
%% 收到来自efka_micro_service的回复 %% 收到来自efka_micro_service的回复
handle_info({ems_reply, Ref, EmsReply}, State = #state{inflight = Inflight}) -> handle_info({ems_reply, Ref, EmsReply}, State = #state{inflight = Inflight}) ->
case maps:take(Ref, Inflight) of case maps:take(Ref, Inflight) of