From e303cd29c0c0f49c238e6bb579278cac364144b0 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sat, 18 Apr 2026 23:17:59 +0800 Subject: [PATCH] fix --- include/protocol.hrl | 13 +++++++------ src/transport/tcp/tcp_channel.erl | 24 ++++++++++++------------ 2 files changed, 19 insertions(+), 18 deletions(-) diff --git a/include/protocol.hrl b/include/protocol.hrl index 35cbd22..3580715 100644 --- a/include/protocol.hrl +++ b/include/protocol.hrl @@ -7,12 +7,13 @@ %%%------------------------------------------------------------------- -author("anlicheng"). -%% efka主动发起的消息体类型, 消息大类 --define(PACKET_REQUEST, 16#01). --define(PACKET_RESPONSE, 16#02). - -%% efka主动发起不需要返回的数据 --define(PACKET_CAST, 16#03). +%% 一级帧类型 +%% REQUEST: 需要响应的请求帧 +%% RESPONSE: 对 REQUEST 的响应帧 +%% CAST: 单向消息帧，不需要响应 +-define(FRAME_REQUEST, 16#01). +-define(FRAME_RESPONSE, 16#02). +-define(FRAME_CAST, 16#03). %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %%%% 二级分类定义 diff --git a/src/transport/tcp/tcp_channel.erl b/src/transport/tcp/tcp_channel.erl index c77dd1c..ac4ecfc 100644 --- a/src/transport/tcp/tcp_channel.erl +++ b/src/transport/tcp/tcp_channel.erl @@ -99,13 +99,13 @@ 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, <>), + 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, <>), + Transport:send(Socket, <>), {noreply, State}; %% 推送消息 @@ -119,7 +119,7 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{t 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,7 +129,7 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{t end. %% auth验证 -handle_info({tcp, Socket, <>}, State = #state{transport = Transport, socket = Socket}) -> +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'), logger:debug("[ws_channel] auth uuid: ~p", [UUID]), @@ -148,14 +148,14 @@ handle_info({tcp, Socket, <>}, 'AuthReply'), AuthReplyBin = <>, - Transport:send(Socket, <>), + 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(#'AuthReply'{code = 1, payload = Reason}, 'AuthReply'), AuthReplyBin = <>, - Transport:send(Socket, <>), + 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}}; @@ -163,7 +163,7 @@ handle_info({tcp, Socket, < Encoded = message_pb:encode_msg(#'AuthReply'{code = 2, payload = Reason}, 'AuthReply'), AuthReplyBin = <>, - Transport:send(Socket, <>), + Transport:send(Socket, <>), logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), {stop, State} @@ -173,16 +173,16 @@ handle_info({tcp, Socket, <>}, State = #state{socket = Socket}) -> +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) -> +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) -> +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">> -> @@ -192,7 +192,7 @@ handle_info({tcp, Socket, < iot_event_stream_observer:stream_data(TaskId, Type, Stream) end, {noreply, State}; -handle_info({tcp, Socket, <>}, State = #state{socket = Socket}) -> +handle_info({tcp, Socket, <>}, State = #state{socket = Socket}) -> logger:warning("[tcp_channel] unsupported cast message type: ~p", [MsgType]), {noreply, State}; @@ -202,7 +202,7 @@ handle_info({tcp, Socket, <>}, State = #state % {noreply, State}; %% 主机端的消息响应 -handle_info({tcp, Socket, <>}, State = #state{socket = Socket, inflight = Inflight}) when PacketId > 0 -> +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