From 871993db032440ba1aeb8ac1a1b9b36373109e3c Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sat, 18 Apr 2026 23:43:34 +0800 Subject: [PATCH] fix --- include/protocol.hrl | 19 --- proto/message.proto | 25 ++++ src/transport/tcp/tcp_channel.erl | 198 +++++++++++++++++------------- 3 files changed, 136 insertions(+), 106 deletions(-) diff --git a/include/protocol.hrl b/include/protocol.hrl index 3580715..e3e6a5b 100644 --- a/include/protocol.hrl +++ b/include/protocol.hrl @@ -15,25 +15,6 @@ -define(FRAME_RESPONSE, 16#02). -define(FRAME_CAST, 16#03). -%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% -%%%% 二级分类定义 -%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% - -%% 主机端上报数据类型标识 --define(MESSAGE_AUTH_REQUEST, 16#01). --define(MESSAGE_AUTH_REPLY, 16#02). - --define(MESSAGE_COMMAND, 16#03). --define(MESSAGE_PUB, 16#05). - --define(MESSAGE_DATA, 16#06). - -%% efka主动上报的event-stream流, 单向消息，主要是: docker-create的实时处理逻辑上报 --define(MESSAGE_EVENT_STREAM, 16#08). - --define(MESSAGE_JSONRPC_REQUEST, 16#F0). --define(MESSAGE_JSONRPC_REPLY, 16#F1). - %%%% 命令类型子分类, 不需要返回值 %% 授权 -define(COMMAND_AUTH, 16#08). diff --git a/proto/message.proto b/proto/message.proto index 5beffee..e70273b 100644 --- a/proto/message.proto +++ b/proto/message.proto @@ -34,6 +34,31 @@ message JsonRpcReply { bytes error = 2; } +message RequestFrame { + uint32 packet_id = 1; + oneof body { + AuthRequest auth_request = 2; + JsonRpcRequest jsonrpc_request = 3; + } +} + +message ResponseFrame { + uint32 packet_id = 1; + oneof body { + AuthReply auth_reply = 2; + JsonRpcReply jsonrpc_reply = 3; + } +} + +message CastFrame { + oneof body { + Pub pub = 1; + Command command = 2; + Data data = 3; + TaskEventStream event_stream = 4; + } +} + message Data { bytes route_key = 1; bytes metric = 2; diff --git a/src/transport/tcp/tcp_channel.erl b/src/transport/tcp/tcp_channel.erl index ac4ecfc..2cf909e 100644 --- a/src/transport/tcp/tcp_channel.erl +++ b/src/transport/tcp/tcp_channel.erl @@ -98,28 +98,30 @@ handle_call(_Request, _From, State) -> %% 发送消息, 基于pub/sub机制 handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) -> - Encoded = message_pb:encode_msg(#'Pub'{topic = Topic, qos = Qos, content = Content}), - Transport:send(Socket, <>), + Encoded = message_pb:encode_msg(#'CastFrame'{ + body = {pub, #'Pub'{topic = Topic, qos = Qos, content = Content}} + }), + Transport:send(Socket, <>), {noreply, State}; %% 发送Command消息 handle_cast({command, CommandType, Command}, State = #state{transport = Transport, socket = Socket}) -> - Encoded = message_pb:encode_msg(#'Command'{command_type = CommandType, command = Command}), - Transport:send(Socket, <>), + Encoded = message_pb:encode_msg(#'CastFrame'{ + body = {command, #'Command'{command_type = CommandType, command = Command}} + }), + Transport:send(Socket, <>), {noreply, State}; %% 推送消息 handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) -> case next_packet_id(PacketId, Inflight) of {ok, NPacketId, NextPacketId} -> - Encoded = message_pb:encode_msg(#'JsonRpcRequest'{ - method = Method, - params = Params + Encoded = message_pb:encode_msg(#'RequestFrame'{ + packet_id = NPacketId, + body = {jsonrpc_request, #'JsonRpcRequest'{method = Method, params = Params}} }), - - EncRequest = <>, TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}), - Transport:send(Socket, <>), + Transport:send(Socket, <>), RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, {noreply, State#state{packet_id = NextPacketId, inflight = maps:put(NPacketId, RequestInfo, Inflight)}}; @@ -129,94 +131,116 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{t end. %% auth验证 -handle_info({tcp, Socket, <>}, State = #state{transport = Transport, socket = Socket}) -> - #'AuthRequest'{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp} = message_pb:decode_msg(RequestBin, 'AuthRequest'), +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, <>), - 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(#'AuthReply'{code = 0, payload = <<"ok">>}, 'AuthReply'), - AuthReplyBin = <>, - 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}}; - {denied, Reason} when is_binary(Reason) -> - erlang:monitor(process, HostPid), - Encoded = message_pb:encode_msg(#'AuthReply'{code = 1, payload = Reason}, 'AuthReply'), - AuthReplyBin = <>, - 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}}; - {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]), - {error, Reason} when is_binary(Reason) -> - Encoded = message_pb:encode_msg(#'AuthReply'{code = 2, payload = Reason}, 'AuthReply'), - AuthReplyBin = <>, - Transport:send(Socket, <>), - logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), - - {stop, State} - end + {stop, State} + end + end; + false -> + logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]), + {stop, State} end; - false -> - logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]), + #'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; -handle_info({tcp, Socket, <>}, State = #state{socket = Socket}) -> - logger:warning("[ws_channel] unsupported request message type: ~p", [MsgType]), - {stop, State}; -handle_info({tcp, Socket, <>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) -> - Data = message_pb:decode_msg(CastBin, 'Data'), - iot_host:handle(HostPid, {data, Data}), - {noreply, State}; - -handle_info({tcp, Socket, <>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) -> - CastMessage = message_pb:decode_msg(CastBin, 'TaskEventStream'), - 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}; -handle_info({tcp, Socket, <>}, State = #state{socket = Socket}) -> - logger:warning("[tcp_channel] unsupported cast message type: ~p", [MsgType]), - {noreply, State}; - -%handle_info({tcp, Socket, <>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) -> -% Ping = message_pb:decode_msg(PingData, ping), -% iot_host:handle(HostPid, {ping, Ping}), -% {noreply, 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; %% 主机端的消息响应 -handle_info({tcp, Socket, <>}, State = #state{socket = Socket, inflight = Inflight}) when PacketId > 0 -> - RpcReply = message_pb:decode_msg(ReplyBin, 'JsonRpcReply'), - - case maps:take(PacketId, Inflight) of - error -> +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}; - {#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}} + #'ResponseFrame'{packet_id = PacketId, body = undefined} -> + logger:warning("[ws_channel] empty response frame, packet_id: ~p", [PacketId]), + {noreply, State} end; handle_info({timeout, TimerRef, {jsonrpc_timeout, PacketId}}, State = #state{inflight = Inflight}) ->