From 544e5086a1e5764958516c21bde0d4c6478f2d1f Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Wed, 13 May 2026 15:11:10 +0800 Subject: [PATCH] fix --- apps/light/src/efka_client.erl | 34 +++++++++++++++++++++++++-------- apps/light/src/light_device.erl | 25 +++++++++++++++--------- 2 files changed, 42 insertions(+), 17 deletions(-) diff --git a/apps/light/src/efka_client.erl b/apps/light/src/efka_client.erl index cb49751..1b2d235 100644 --- a/apps/light/src/efka_client.erl +++ b/apps/light/src/efka_client.erl @@ -71,9 +71,9 @@ send_metric_data(Fields, Tags) when is_list(Fields), is_map(Tags) -> -spec invoke_service(ToService :: binary(), Message :: map(), Timeout :: integer()) -> {ok, Result :: any()} | {error, Reason :: any()}. -invoke_service(ToService, Message, Timeout) when is_binary(ToService), is_map(Message), is_integer(Timeout) -> +invoke_service(ToService, Message, Timeout) when is_binary(ToService), is_map(Message), is_integer(Timeout), Timeout > 0 -> {ok, Ref} = gen_server:call(?MODULE, {invoke_service, self(), ToService, Message, Timeout}), - await_reply(Ref, ?EFKA_REQUEST_TIMEOUT). + await_reply(Ref, Timeout). -spec send_log(Message :: binary() | map()) -> no_return(). send_log(Message) when is_binary(Message); is_map(Message) -> @@ -191,7 +191,8 @@ handle_call({send_metric_data, ReceiverPid, Fields, Tags}, _From, State = #state Packet = pack(PacketId, ?PACKET_TYPE_METRIC_DATA, Body), ok = gen_tcp:send(Socket, Packet), Ref = make_ref(), - {reply, {ok, Ref}, State#state{packet_id = next_packet_id(PacketId), inflight = maps:put(PacketId, {Ref, ReceiverPid}, Inflight)}}; + NInflight = put_inflight(PacketId, Ref, ReceiverPid, ?EFKA_REQUEST_TIMEOUT, Inflight), + {reply, {ok, Ref}, State#state{packet_id = next_packet_id(PacketId), inflight = NInflight}}; handle_call({invoke_service, ReceiverPid, ToService, Message, Timeout}, _From, State = #state{socket = Socket, packet_id = PacketId, inflight = Inflight}) -> Body = #{ @@ -202,19 +203,22 @@ handle_call({invoke_service, ReceiverPid, ToService, Message, Timeout}, _From, S Packet = pack(PacketId, ?PACKET_TYPE_INVOKE, Body), ok = gen_tcp:send(Socket, Packet), Ref = make_ref(), - {reply, {ok, Ref}, State#state{packet_id = next_packet_id(PacketId), inflight = maps:put(PacketId, {Ref, ReceiverPid}, Inflight)}}; + NInflight = put_inflight(PacketId, Ref, ReceiverPid, Timeout, Inflight), + {reply, {ok, Ref}, State#state{packet_id = next_packet_id(PacketId), inflight = NInflight}}; handle_call({request_metric, ReceiverPid}, _From, State = #state{socket = Socket, packet_id = PacketId, inflight = Inflight}) -> Packet = pack(PacketId, ?PACKET_TYPE_REQUEST_METRIC), ok = gen_tcp:send(Socket, Packet), Ref = make_ref(), - {reply, {ok, Ref}, State#state{packet_id = next_packet_id(PacketId), inflight = maps:put(PacketId, {Ref, ReceiverPid}, Inflight)}}; + NInflight = put_inflight(PacketId, Ref, ReceiverPid, ?EFKA_REQUEST_TIMEOUT, Inflight), + {reply, {ok, Ref}, State#state{packet_id = next_packet_id(PacketId), inflight = NInflight}}; handle_call({request_param, ReceiverPid}, _From, State = #state{socket = Socket, packet_id = PacketId, inflight = Inflight}) -> Packet = pack(PacketId, ?PACKET_TYPE_REQUEST_PARAM), ok = gen_tcp:send(Socket, Packet), Ref = make_ref(), - {reply, {ok, Ref}, State#state{packet_id = next_packet_id(PacketId), inflight = maps:put(PacketId, {Ref, ReceiverPid}, Inflight)}}. + NInflight = put_inflight(PacketId, Ref, ReceiverPid, ?EFKA_REQUEST_TIMEOUT, Inflight), + {reply, {ok, Ref}, State#state{packet_id = next_packet_id(PacketId), inflight = NInflight}}. %% @private %% @doc Handling cast messages @@ -259,7 +263,8 @@ handle_info({handle_packet, #efka_packet{packet_id = PacketId, type = ?PACKET_TY case maps:take(PacketId, Inflight) of error -> {noreply, State}; - {{Ref, ReceiverPid}, NInflight} -> + {{Ref, ReceiverPid, TimerRef}, NInflight} -> + erlang:cancel_timer(TimerRef), case Message of #{<<"c">> := 1, <<"r">> := Result} -> ReceiverPid ! {response, Ref, {ok, Result}}; @@ -269,6 +274,15 @@ handle_info({handle_packet, #efka_packet{packet_id = PacketId, type = ?PACKET_TY {noreply, State#state{inflight = NInflight}} end; +handle_info({request_timeout, PacketId, Ref}, State = #state{inflight = Inflight}) -> + case maps:take(PacketId, Inflight) of + {{Ref, ReceiverPid, _TimerRef}, NInflight} -> + ReceiverPid ! {response, Ref, {error, timeout}}, + {noreply, State#state{inflight = NInflight}}; + _ -> + {noreply, State} + end; + %% 收到efka推送的参数设置 handle_info({handle_packet, #efka_packet{packet_id = PacketId, type = ?PACKET_TYPE_PUSH_PARAM, message = Params}}, State = #state{socket = Socket}) when is_map(Params) -> @@ -370,6 +384,10 @@ next_packet_id(PacketId) when PacketId >= 65535 -> next_packet_id(PacketId) -> PacketId + 1. +put_inflight(PacketId, Ref, ReceiverPid, Timeout, Inflight) -> + TimerRef = erlang:send_after(Timeout, self(), {request_timeout, PacketId, Ref}), + maps:put(PacketId, {Ref, ReceiverPid, TimerRef}, Inflight). + -spec pack(PacketId :: integer(), Type :: integer()) -> binary(). pack(PacketId, Type) when is_integer(PacketId), is_integer(Type) -> <>. @@ -428,4 +446,4 @@ handle_metric(Metric) when is_list(Metric) -> | {break, NewServiceName :: binary(), NewFields :: list(), NewTag :: map()} | error. handle_stream_call(ServiceName, Fields, Tag) when is_binary(ServiceName), is_list(Fields), is_map(Tag) -> - {continue, ServiceName, Fields, Tag}. \ No newline at end of file + {continue, ServiceName, Fields, Tag}. diff --git a/apps/light/src/light_device.erl b/apps/light/src/light_device.erl index 4bbc9a2..e950603 100644 --- a/apps/light/src/light_device.erl +++ b/apps/light/src/light_device.erl @@ -99,19 +99,15 @@ handle_cast({metric_data, Message}, State = #state{device_uuid = DeviceUUID, dat #{<<"properties">> := Props0} -> Props = lists:map(fun(Fields) -> transform(Fields#{<<"device_uuid">> => DeviceUUID}) end, Props0), Info = iolist_to_binary(jiffy:encode(Props, [force_utf8])), - case catch efka_client:send_metric_data(Props, #{}) of - {ok, _} -> - light_logger:write([<<"OK">>, Info]); - _ -> - light_logger:write([<<"ERROR">>, Info]) - end, + send_metric_data(Props, Info), %% 如果设备当前是离线状态，则需要发送上线消息 Status == 0 andalso efka_client:device_online(DeviceUUID), {noreply, State#state{data_counter = DataCounter + 1, status = 1}}; M when is_map(M) -> - lager:notice("[light_device] invalid map: ~p", [M]); + lager:notice("[light_device] invalid map: ~p", [M]), + {noreply, State}; Error -> lager:notice("[light_device] jiffy decode error: ~p", [Error]), {noreply, State} @@ -127,7 +123,7 @@ handle_info({timeout, _, heartbeat_ticker}, State = #state{device_uuid = DeviceU erlang:start_timer(HeartbeatTicker, self(), heartbeat_ticker), case DataCounter > 0 of true -> - {noreply, State}; + {noreply, State#state{data_counter = 0, status = 1}}; false -> Status == 1 andalso efka_client:device_offline(DeviceUUID), {noreply, State#state{status = 0}} @@ -174,4 +170,15 @@ transform(Prop = #{<<"key">> := <<"light_change_time">>, <<"unit">> := Unit}) -> transform(Prop = #{<<"key">> := <<"light_status">>, <<"unit">> := Unit}) -> Prop#{<<"name">> => <<"是否损坏"/utf8>>, <<"value">> => Unit, <<"unit">> => ?UNIT0}; transform(Prop = #{<<"unit">> := Unit}) -> - Prop#{<<"value">> => Unit, <<"unit">> => ?UNIT0}. \ No newline at end of file + Prop#{<<"value">> => Unit, <<"unit">> => ?UNIT0}. + +send_metric_data(Props, Info) -> + _ = spawn(fun() -> + case catch efka_client:send_metric_data(Props, #{}) of + {ok, _} -> + light_logger:write([<<"OK">>, Info]); + _ -> + light_logger:write([<<"ERROR">>, Info]) + end + end), + ok.