fix

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) ->
     <<PacketId:16, Type:8>>.

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
+            send_metric_data(Props, Info),
-                {ok, _} ->
-                    light_logger:write([<<"OK">>, Info]);
-                _ ->
-                    light_logger:write([<<"ERROR">>, Info])
-            end,
 
             %% 如果设备当前是离线状态，则需要发送上线消息
             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}}
@@ -175,3 +171,14 @@ transform(Prop = #{<<"key">> := <<"light_status">>, <<"unit">> := Unit}) ->
     Prop#{<<"name">> => <<"是否损坏"/utf8>>, <<"value">> => Unit, <<"unit">> => ?UNIT0};
 transform(Prop = #{<<"unit">> := Unit}) ->
     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.