From c2ad7c5096ee813bb4aa03c287c0eef19960ecf0 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sat, 18 Apr 2026 23:54:48 +0800 Subject: [PATCH] fix tcp_channel --- src/transport/tcp/tcp_channel.erl | 216 +++++++++++++++--------------- 1 file changed, 111 insertions(+), 105 deletions(-) diff --git a/src/transport/tcp/tcp_channel.erl b/src/transport/tcp/tcp_channel.erl index 2cf909e..840a936 100644 --- a/src/transport/tcp/tcp_channel.erl +++ b/src/transport/tcp/tcp_channel.erl @@ -132,116 +132,17 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{t %% auth验证 handle_info({tcp, Socket, <>}, State = #state{transport = Transport, socket = Socket}) -> - case message_pb:decode_msg(FrameBin, 'RequestFrame') of - #'RequestFrame'{packet_id = PacketId, body = {auth_request, #'AuthRequest'{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp}}} -> - logger:debug("[ws_channel] auth uuid: ~p", [UUID]), - case iot_auth:check(Username, Token, UUID, Salt, Timestamp) of - true -> - case iot_api_client:get_host_by_uuid(UUID) of - undefined -> - logger:warning("[ws_channel] uuid: ~p, user: ~p, host not found", [UUID, Username]), - {stop, State}; - {ok, _} -> - %% 尝试启动主机的服务进程 - {ok, HostPid} = iot_host_sup:ensured_host_started(UUID), - case iot_host:attach_channel(HostPid, self()) of - ok -> - %% 建立到host的monitor - erlang:monitor(process, HostPid), - Encoded = message_pb:encode_msg(#'ResponseFrame'{ - packet_id = PacketId, - body = {auth_reply, #'AuthReply'{code = 0, payload = <<"ok">>}} - }), - Transport:send(Socket, <>), - - {noreply, State#state{uuid = UUID, host_pid = HostPid}}; - {denied, Reason} when is_binary(Reason) -> - erlang:monitor(process, HostPid), - Encoded = message_pb:encode_msg(#'ResponseFrame'{ - packet_id = PacketId, - body = {auth_reply, #'AuthReply'{code = 1, payload = Reason}} - }), - Transport:send(Socket, <>), - logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), - - {noreply, State#state{uuid = UUID, host_pid = HostPid}}; - - {error, Reason} when is_binary(Reason) -> - Encoded = message_pb:encode_msg(#'ResponseFrame'{ - packet_id = PacketId, - body = {auth_reply, #'AuthReply'{code = 2, payload = Reason}} - }), - Transport:send(Socket, <>), - logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), - - {stop, State} - end - end; - false -> - logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]), - {stop, State} - end; - #'RequestFrame'{packet_id = PacketId, body = {jsonrpc_request, RpcRequest}} -> - logger:warning("[ws_channel] unsupported request message type: jsonrpc_request, packet_id: ~p, request: ~p", [PacketId, RpcRequest]), - {stop, State}; - #'RequestFrame'{packet_id = PacketId, body = undefined} -> - logger:warning("[ws_channel] empty request frame, packet_id: ~p", [PacketId]), - {stop, State} - end; + RequestFrame = message_pb:decode_msg(FrameBin, 'RequestFrame'), + handle_request_frame(RequestFrame, Transport, Socket, State); handle_info({tcp, Socket, <>}, State = #state{socket = Socket, host_pid = HostPid}) -> - case message_pb:decode_msg(FrameBin, 'CastFrame') of - #'CastFrame'{body = {data, Data}} when is_pid(HostPid) -> - iot_host:handle(HostPid, {data, Data}), - {noreply, State}; - #'CastFrame'{body = {event_stream, CastMessage}} when is_pid(HostPid) -> - case CastMessage of - #'TaskEventStream'{task_id = TaskId, type = Type0, stream = Reason0} when Type0 =:= <<"close">> -> - iot_event_stream_observer:stream_close(TaskId, iolist_to_binary(Reason0)); - #'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream} -> - logger:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]), - iot_event_stream_observer:stream_data(TaskId, Type, Stream) - end, - {noreply, State}; - #'CastFrame'{body = {data, _Data}} -> - {noreply, State}; - #'CastFrame'{body = {event_stream, _EventStream}} -> - {noreply, State}; - #'CastFrame'{body = {pub, Pub}} -> - logger:warning("[tcp_channel] unsupported cast message type: pub, body: ~p", [Pub]), - {noreply, State}; - #'CastFrame'{body = {command, Cmd}} -> - logger:warning("[tcp_channel] unsupported cast message type: command, body: ~p", [Cmd]), - {noreply, State}; - #'CastFrame'{body = undefined} -> - logger:warning("[tcp_channel] empty cast frame"), - {noreply, State} - end; + CastFrame = message_pb:decode_msg(FrameBin, 'CastFrame'), + handle_cast_frame(CastFrame, HostPid, State); %% 主机端的消息响应 handle_info({tcp, Socket, <>}, State = #state{socket = Socket, inflight = Inflight}) -> - case message_pb:decode_msg(FrameBin, 'ResponseFrame') of - #'ResponseFrame'{packet_id = PacketId, body = {jsonrpc_reply, RpcReply}} when PacketId > 0 -> - case maps:take(PacketId, Inflight) of - error -> - {noreply, State}; - {#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 -> - ReceiverPid ! {jsonrpc_reply, Ref, RpcReply}; - false -> - logger:warning("[ws_channel] get async_call_reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [RpcReply, PacketId]) - end, - {noreply, State#state{inflight = NInflight}} - end; - #'ResponseFrame'{packet_id = PacketId, body = {auth_reply, AuthReply}} -> - logger:warning("[ws_channel] unexpected auth_reply response, packet_id: ~p, body: ~p", [PacketId, AuthReply]), - {noreply, State}; - #'ResponseFrame'{packet_id = PacketId, body = undefined} -> - logger:warning("[ws_channel] empty response frame, packet_id: ~p", [PacketId]), - {noreply, State} - end; + ResponseFrame = message_pb:decode_msg(FrameBin, 'ResponseFrame'), + handle_response_frame(ResponseFrame, Inflight, State); handle_info({timeout, TimerRef, {jsonrpc_timeout, PacketId}}, State = #state{inflight = Inflight}) -> case maps:get(PacketId, Inflight, undefined) of @@ -280,6 +181,13 @@ terminate(Reason, #state{}) -> code_change(_OldVsn, State, _Extra) -> {ok, State}. + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% +%%%% helper methods +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +-spec take_inflight_by_ref(reference(), map()) -> + error | {ok, #inflight_request{}, map()}. take_inflight_by_ref(Ref, Inflight) -> maps:fold(fun(PacketId, Request = #inflight_request{ref = PacketRef}, Acc) -> case Acc of @@ -290,11 +198,15 @@ take_inflight_by_ref(Ref, Inflight) -> end end, error, Inflight). +-spec next_packet_id(integer(), map()) -> + {ok, integer(), integer()} | {error, inflight_full}. next_packet_id(_PacketId, Inflight) when map_size(Inflight) >= ?MAX_PACKET_ID -> {error, inflight_full}; next_packet_id(PacketId, Inflight) -> next_packet_id(PacketId, Inflight, PacketId, 0). +-spec next_packet_id(integer(), map(), integer(), non_neg_integer()) -> + {ok, integer(), integer()} | {error, inflight_full}. next_packet_id(_PacketId, _Inflight, _StartPacketId, TryCount) when TryCount > ?MAX_PACKET_ID -> {error, inflight_full}; next_packet_id(PacketId, Inflight, StartPacketId, TryCount) -> @@ -311,7 +223,101 @@ next_packet_id(PacketId, Inflight, StartPacketId, TryCount) -> end end. +-spec inc_packet_id(integer()) -> integer(). inc_packet_id(?MAX_PACKET_ID) -> 1; inc_packet_id(PacketId) when PacketId > 0, PacketId < ?MAX_PACKET_ID -> PacketId + 1. + +-spec handle_request_frame(message_pb:'RequestFrame'(), module(), any(), #state{}) -> + {noreply, #state{}} | {stop, #state{}}. +handle_request_frame(#'RequestFrame'{packet_id = PacketId, + body = {auth_request, #'AuthRequest'{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp}}}, + Transport, Socket, State) -> + logger:debug("[ws_channel] auth uuid: ~p", [UUID]), + case iot_auth:check(Username, Token, UUID, Salt, Timestamp) of + true -> + case iot_api_client:get_host_by_uuid(UUID) of + undefined -> + logger:warning("[ws_channel] uuid: ~p, user: ~p, host not found", [UUID, Username]), + {stop, State}; + {ok, _} -> + %% 尝试启动主机的服务进程 + {ok, HostPid} = iot_host_sup:ensured_host_started(UUID), + case iot_host:attach_channel(HostPid, self()) of + ok -> + erlang:monitor(process, HostPid), + send_response_frame(Transport, Socket, PacketId, + {auth_reply, #'AuthReply'{code = 0, payload = <<"ok">>}}), + {noreply, State#state{uuid = UUID, host_pid = HostPid}}; + {denied, Reason} when is_binary(Reason) -> + erlang:monitor(process, HostPid), + send_response_frame(Transport, Socket, PacketId, + {auth_reply, #'AuthReply'{code = 1, payload = Reason}}), + logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), + {noreply, State#state{uuid = UUID, host_pid = HostPid}}; + {error, Reason} when is_binary(Reason) -> + send_response_frame(Transport, Socket, PacketId, + {auth_reply, #'AuthReply'{code = 2, payload = Reason}}), + logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), + {stop, State} + end + end; + false -> + logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]), + {stop, State} + end; +handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {jsonrpc_request, RpcRequest}}, _Transport, _Socket, State) -> + logger:warning("[ws_channel] unsupported request message type: jsonrpc_request, packet_id: ~p, request: ~p", [PacketId, RpcRequest]), + {stop, State}; +handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = undefined}, _Transport, _Socket, State) -> + logger:warning("[ws_channel] empty request frame, packet_id: ~p", [PacketId]), + {stop, State}. + +-spec handle_cast_frame(message_pb:'CastFrame'(), undefined | pid(), #state{}) -> + {noreply, #state{}}. +handle_cast_frame(#'CastFrame'{body = {data, Data}}, HostPid, State) when is_pid(HostPid) -> + iot_host:handle(HostPid, {data, Data}), + {noreply, State}; +handle_cast_frame(#'CastFrame'{body = {event_stream, CastMessage}}, HostPid, State) when is_pid(HostPid) -> + handle_event_stream_frame(CastMessage), + {noreply, State}; +handle_cast_frame(#'CastFrame'{body = Body}, _HostPid, State) -> + logger:warning("[tcp_channel] unsupported cast message type: command, body: ~p", [Body]), + {noreply, State}. + +-spec handle_event_stream_frame(message_pb:'TaskEventStream'()) -> any(). +handle_event_stream_frame(#'TaskEventStream'{task_id = TaskId, type = Type0, stream = Reason0}) when Type0 =:= <<"close">> -> + iot_event_stream_observer:stream_close(TaskId, iolist_to_binary(Reason0)); +handle_event_stream_frame(#'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream}) -> + logger:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]), + iot_event_stream_observer:stream_data(TaskId, Type, Stream). + +-spec handle_response_frame(message_pb:'ResponseFrame'(), map(), #state{}) -> + {noreply, #state{}}. +handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {jsonrpc_reply, RpcReply}}, Inflight, State) + when PacketId > 0 -> + case maps:take(PacketId, Inflight) of + error -> + {noreply, State}; + {#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 -> + ReceiverPid ! {jsonrpc_reply, Ref, RpcReply}; + false -> + logger:warning("[ws_channel] get async_call_reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [RpcReply, PacketId]) + end, + {noreply, State#state{inflight = NInflight}} + end; +handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {auth_reply, AuthReply}}, _Inflight, State) -> + logger:warning("[ws_channel] unexpected auth_reply response, packet_id: ~p, body: ~p", [PacketId, AuthReply]), + {noreply, State}; +handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = undefined}, _Inflight, State) -> + logger:warning("[ws_channel] empty response frame, packet_id: ~p", [PacketId]), + {noreply, State}. + +-spec send_response_frame(module(), any(), non_neg_integer(), tuple()) -> any(). +send_response_frame(Transport, Socket, PacketId, Body) -> + Encoded = message_pb:encode_msg(#'ResponseFrame'{packet_id = PacketId, body = Body}), + Transport:send(Socket, <>).