fix

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).

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;

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}),
+    Encoded = message_pb:encode_msg(#'CastFrame'{
-    Transport:send(Socket, <<?FRAME_CAST, ?MESSAGE_PUB, Encoded/binary>>),
+        body = {pub, #'Pub'{topic = Topic, qos = Qos, content = Content}}
+    }),
+    Transport:send(Socket, <<?FRAME_CAST, Encoded/binary>>),
     {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}),
+    Encoded = message_pb:encode_msg(#'CastFrame'{
-    Transport:send(Socket, <<?FRAME_CAST, ?MESSAGE_COMMAND, Encoded/binary>>),
+        body = {command, #'Command'{command_type = CommandType, command = Command}}
+    }),
+    Transport:send(Socket, <<?FRAME_CAST, Encoded/binary>>),
     {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'{
+            Encoded = message_pb:encode_msg(#'RequestFrame'{
-                method = Method,
+                packet_id = NPacketId,
-                params = Params
+                body = {jsonrpc_request, #'JsonRpcRequest'{method = Method, params = Params}}
             }),
-
-            EncRequest = <<?MESSAGE_JSONRPC_REQUEST, Encoded/binary>>,
             TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}),
-            Transport:send(Socket, <<?FRAME_REQUEST, NPacketId:32, EncRequest/binary>>),
+            Transport:send(Socket, <<?FRAME_REQUEST, Encoded/binary>>),
 
             RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef},
             {noreply, State#state{packet_id = NextPacketId, inflight = maps:put(NPacketId, RequestInfo, Inflight)}};
@@ -129,9 +131,9 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{t
     end.
 
 %% auth验证
-handle_info({tcp, Socket, <<?FRAME_REQUEST, PacketId:32, ?MESSAGE_AUTH_REQUEST, RequestBin/binary>>}, State = #state{transport = Transport, socket = Socket}) ->
+handle_info({tcp, Socket, <<?FRAME_REQUEST, FrameBin/binary>>}, State = #state{transport = Transport, socket = Socket}) ->
-    #'AuthRequest'{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp} = message_pb:decode_msg(RequestBin, 'AuthRequest'),
+    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 ->
@@ -146,24 +148,30 @@ handle_info({tcp, Socket, <<?FRAME_REQUEST, PacketId:32, ?MESSAGE_AUTH_REQUEST,
                                 ok ->
                                     %% 建立到host的monitor
                                     erlang:monitor(process, HostPid),
-                            Encoded = message_pb:encode_msg(#'AuthReply'{code = 0, payload = <<"ok">>}, 'AuthReply'),
+                                    Encoded = message_pb:encode_msg(#'ResponseFrame'{
-                            AuthReplyBin = <<?MESSAGE_AUTH_REPLY, Encoded/binary>>,
+                                        packet_id = PacketId,
-                            Transport:send(Socket, <<?FRAME_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
+                                        body = {auth_reply, #'AuthReply'{code = 0, payload = <<"ok">>}}
+                                    }),
+                                    Transport:send(Socket, <<?FRAME_RESPONSE, Encoded/binary>>),
 
                                     {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'),
+                                    Encoded = message_pb:encode_msg(#'ResponseFrame'{
-                            AuthReplyBin = <<?MESSAGE_AUTH_REPLY, Encoded/binary>>,
+                                        packet_id = PacketId,
-                            Transport:send(Socket, <<?FRAME_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
+                                        body = {auth_reply, #'AuthReply'{code = 1, payload = Reason}}
+                                    }),
+                                    Transport:send(Socket, <<?FRAME_RESPONSE, Encoded/binary>>),
                                     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(#'AuthReply'{code = 2, payload = Reason}, 'AuthReply'),
+                                    Encoded = message_pb:encode_msg(#'ResponseFrame'{
-                            AuthReplyBin = <<?MESSAGE_AUTH_REPLY, Encoded/binary>>,
+                                        packet_id = PacketId,
-                            Transport:send(Socket, <<?FRAME_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
+                                        body = {auth_reply, #'AuthReply'{code = 2, payload = Reason}}
+                                    }),
+                                    Transport:send(Socket, <<?FRAME_RESPONSE, Encoded/binary>>),
                                     logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
 
                                     {stop, State}
@@ -173,17 +181,20 @@ handle_info({tcp, Socket, <<?FRAME_REQUEST, PacketId:32, ?MESSAGE_AUTH_REQUEST,
                     logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]),
                     {stop, State}
             end;
-handle_info({tcp, Socket, <<?FRAME_REQUEST, _PacketId:32, MsgType:8, _/binary>>}, State = #state{socket = Socket}) ->
+        #'RequestFrame'{packet_id = PacketId, body = {jsonrpc_request, RpcRequest}} ->
-    logger:warning("[ws_channel] unsupported request message type: ~p", [MsgType]),
+            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, <<?FRAME_CAST, ?MESSAGE_DATA, CastBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
+handle_info({tcp, Socket, <<?FRAME_CAST, FrameBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) ->
-    Data = message_pb:decode_msg(CastBin, 'Data'),
+    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) ->
-handle_info({tcp, Socket, <<?FRAME_CAST, ?MESSAGE_EVENT_STREAM, CastBin/binary>>}, 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));
@@ -192,19 +203,25 @@ handle_info({tcp, Socket, <<?FRAME_CAST, ?MESSAGE_EVENT_STREAM, CastBin/binary>>
                     iot_event_stream_observer:stream_data(TaskId, Type, Stream)
             end,
             {noreply, State};
-handle_info({tcp, Socket, <<?FRAME_CAST, MsgType:8, _/binary>>}, State = #state{socket = Socket}) ->
+        #'CastFrame'{body = {data, _Data}} ->
-    logger:warning("[tcp_channel] unsupported cast message type: ~p", [MsgType]),
             {noreply, State};
-
+        #'CastFrame'{body = {event_stream, _EventStream}} ->
-%handle_info({tcp, Socket, <<?PACKET_PING, PingData/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
+            {noreply, State};
-%    Ping = message_pb:decode_msg(PingData, ping),
+        #'CastFrame'{body = {pub, Pub}} ->
-%    iot_host:handle(HostPid, {ping, Ping}),
+            logger:warning("[tcp_channel] unsupported cast message type: pub, body: ~p", [Pub]),
-%    {noreply, State};
+            {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, <<?FRAME_RESPONSE, PacketId:32, ?MESSAGE_JSONRPC_REPLY, ReplyBin/binary>>}, State = #state{socket = Socket, inflight = Inflight}) when PacketId > 0 ->
+handle_info({tcp, Socket, <<?FRAME_RESPONSE, FrameBin/binary>>}, State = #state{socket = Socket, inflight = Inflight}) ->
-    RpcReply = message_pb:decode_msg(ReplyBin, 'JsonRpcReply'),
+    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};
@@ -218,6 +235,13 @@ handle_info({tcp, Socket, <<?FRAME_RESPONSE, PacketId:32, ?MESSAGE_JSONRPC_REPLY
                     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;
 
 handle_info({timeout, TimerRef, {jsonrpc_timeout, PacketId}}, State = #state{inflight = Inflight}) ->
     case maps:get(PacketId, Inflight, undefined) of