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

fix

Browse Source
This commit is contained in:
anlicheng 2026-04-18 17:34:50 +08:00
parent 5be57b1ae3
commit 5d91fef111
1 changed files with 13 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/transport/tcp/tcp_channel.erl
View File

@ -36,6 +36,12 @@
inflight = #{}
}).
-record(inflight_request, {
receiver_pid :: pid(),
ref :: reference(),
timer_ref :: reference()
}).
%% 向通道中写入消息
-spec pub(Pid :: pid(), Topic :: binary(), Qos :: integer(), Content :: binary()) -> no_return().
pub(Pid, Topic, Qos, Content) when is_pid(Pid), is_binary(Topic), is_integer(Qos), is_binary(Content) ->
@ -81,7 +87,7 @@ init(Ref, Transport, _Opts = []) ->
handle_call({cancel_jsonrpc_call, Ref}, _From, State = #state{inflight = Inflight}) ->
case take_inflight_by_ref(Ref, Inflight) of
{ok, _PacketId, {_ReceiverPid, _Ref, TimerRef}, NInflight} ->
{ok, #inflight_request{timer_ref = TimerRef}, NInflight} ->
erlang:cancel_timer(TimerRef),
{reply, ok, State#state{inflight = NInflight}};
error ->
@ -115,7 +121,8 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{t
TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}),
Transport:send(Socket, <<?PACKET_REQUEST, NPacketId:32, EncRequest/binary>>),
{noreply, State#state{packet_id = NextPacketId, inflight = maps:put(NPacketId, {ReceiverPid, Ref, TimerRef}, Inflight)}};
RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef},
{noreply, State#state{packet_id = NextPacketId, inflight = maps:put(NPacketId, RequestInfo, Inflight)}};
{error, inflight_full} ->
logger:warning("[ws_channel] uuid: ~p, inflight requests exhausted", [State#state.uuid]),
{noreply, State}
@ -201,7 +208,7 @@ handle_info({tcp, Socket, <<?PACKET_RESPONSE, PacketId:32, ?MESSAGE_JSONRPC_REPL
case maps:take(PacketId, Inflight) of
error ->
{noreply, State};
{{ReceiverPid, Ref, TimerRef}, NInflight} ->
{#inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, NInflight} ->
erlang:cancel_timer(TimerRef),
case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
true ->
@ -214,7 +221,7 @@ handle_info({tcp, Socket, <<?PACKET_RESPONSE, PacketId:32, ?MESSAGE_JSONRPC_REPL
handle_info({timeout, TimerRef, {jsonrpc_timeout, PacketId}}, State = #state{inflight = Inflight}) ->
case maps:get(PacketId, Inflight, undefined) of
{_ReceiverPid, Ref, TimerRef} ->
#inflight_request{ref = Ref, timer_ref = TimerRef} ->
logger:warning("[ws_channel] jsonrpc request timeout, packet_id: ~p, ref: ~p", [PacketId, Ref]),
{noreply, State#state{inflight = maps:remove(PacketId, Inflight)}};
_ ->
@ -250,10 +257,10 @@ code_change(_OldVsn, State, _Extra) ->
{ok, State}.
take_inflight_by_ref(Ref, Inflight) ->
maps:fold(fun(PacketId, Value = {_ReceiverPid, PacketRef, _TimerRef}, Acc) ->
maps:fold(fun(PacketId, Request = #inflight_request{ref = PacketRef}, Acc) ->
case Acc of
error when PacketRef =:= Ref ->
{ok, PacketId, Value, maps:remove(PacketId, Inflight)};
{ok, Request, maps:remove(PacketId, Inflight)};
_ ->
Acc
end