diff --git a/apps/efka/src/efka_agent.erl b/apps/efka/src/efka_agent.erl index 3086500..a925b21 100644 --- a/apps/efka/src/efka_agent.erl +++ b/apps/efka/src/efka_agent.erl @@ -318,7 +318,7 @@ handle_info({server_push_message, PacketId, <>}, State = #state{transport_pid = TransportPid, inflight = Inflight}) -> +handle_info({server_push_message, PacketId, <>}, 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 @@ -327,11 +327,16 @@ handle_info({server_push_message, PacketId, <> }, - efka_transport:response(TransportPid, PacketId, message_pb:encode_msg(Reply)), + 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;