fix

include/message_pb.hrl

@@ -0,0 +1,80 @@
+%% -*- coding: utf-8 -*-
+%% Automatically generated, do not edit
+%% Generated by gpb_compile version 4.21.7
+
+-ifndef(message_pb).
+-define(message_pb, true).
+
+-define(message_pb_gpb_version, "4.21.7").
+
+
+-ifndef('AUTHREQUEST_PB_H').
+-define('AUTHREQUEST_PB_H', true).
+-record('AuthRequest',
+        {uuid = <<>>            :: iodata() | undefined, % = 1, optional
+         username = <<>>        :: iodata() | undefined, % = 2, optional
+         salt = <<>>            :: iodata() | undefined, % = 3, optional
+         token = <<>>           :: iodata() | undefined, % = 4, optional
+         timestamp = 0          :: integer() | undefined % = 5, optional, 32 bits
+        }).
+-endif.
+
+-ifndef('AUTHREPLY_PB_H').
+-define('AUTHREPLY_PB_H', true).
+-record('AuthReply',
+        {code = 0               :: integer() | undefined, % = 1, optional, 32 bits
+         payload = <<>>         :: iodata() | undefined % = 2, optional
+        }).
+-endif.
+
+-ifndef('PUB_PB_H').
+-define('PUB_PB_H', true).
+-record('Pub',
+        {topic = <<>>           :: iodata() | undefined, % = 1, optional
+         qos = 0                :: integer() | undefined, % = 2, optional, 32 bits
+         content = <<>>         :: iodata() | undefined % = 3, optional
+        }).
+-endif.
+
+-ifndef('COMMAND_PB_H').
+-define('COMMAND_PB_H', true).
+-record('Command',
+        {command_type = 0       :: integer() | undefined, % = 1, optional, 32 bits
+         command = <<>>         :: iodata() | undefined % = 2, optional
+        }).
+-endif.
+
+-ifndef('JSONRPCREQUEST_PB_H').
+-define('JSONRPCREQUEST_PB_H', true).
+-record('JsonRpcRequest',
+        {method = <<>>          :: iodata() | undefined, % = 1, optional
+         params = <<>>          :: iodata() | undefined % = 2, optional
+        }).
+-endif.
+
+-ifndef('JSONRPCREPLY_PB_H').
+-define('JSONRPCREPLY_PB_H', true).
+-record('JsonRpcReply',
+        {result = <<>>          :: iodata() | undefined, % = 1, optional
+         error = <<>>           :: iodata() | undefined % = 2, optional
+        }).
+-endif.
+
+-ifndef('DATA_PB_H').
+-define('DATA_PB_H', true).
+-record('Data',
+        {route_key = <<>>       :: iodata() | undefined, % = 1, optional
+         metric = <<>>          :: iodata() | undefined % = 2, optional
+        }).
+-endif.
+
+-ifndef('TASKEVENTSTREAM_PB_H').
+-define('TASKEVENTSTREAM_PB_H', true).
+-record('TaskEventStream',
+        {task_id = 0            :: integer() | undefined, % = 1, optional, 32 bits
+         type = <<>>            :: iodata() | undefined, % = 2, optional
+         stream = <<>>          :: iodata() | undefined % = 3, optional
+        }).
+-endif.
+
+-endif.

rebar.config

@@ -12,7 +12,7 @@
     {src_dirs, ["proto"]},        % 源码目录（必须）
     recursive,                     % 递归查找 proto 文件
     {module_name_suffix, "_pb"},   % 生成模块后缀
-    {o_erl, "src"},                % .erl 输出目录
+    {o_erl, "src/protobuf"},                % .erl 输出目录
     {o_hrl, "include"},            % .hrl 输出目录
     include_as_lib,                % gpb.hrl 通过 -include_lib("gpb/include/gpb.hrl")
     {strings_as_binaries, true},   % proto string → Erlang binary
@@ -21,6 +21,13 @@
     verbose                        % 打印详细信息
 ]}.
 
+{provider_hooks, [
+    {pre, [
+        {compile, {protobuf, compile}},
+        {clean, {protobuf, clean}}
+    ]}
+]}.
+
 {deps, [
     {poolboy, ".*", {git, "https://github.com/devinus/poolboy.git", {tag, "1.5.1"}}},
     {hackney, ".*", {git, "https://github.com/benoitc/hackney.git", {tag, "1.25.0"}}},

src/codec/message_codec.erl

@@ -1,141 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @author anlicheng
-%%% @copyright (C) 2025, <COMPANY>
-%%% @doc
-%%%
-%%% @end
-%%% Created : 17. 9月 2025 16:05
-%%%-------------------------------------------------------------------
--module(message_codec).
--author("anlicheng").
--include("message.hrl").
-
--define(I8, 1).
--define(I16, 2).
--define(I32, 3).
--define(Bytes, 4).
-
-%% API
--export([encode/2, decode/1]).
-
--spec encode(MessageType :: integer(), Message :: any()) -> binary().
-encode(MessageType, Message) when is_integer(MessageType) ->
-    Bin = encode0(Message),
-    <<MessageType, Bin/binary>>.
-encode0(#auth_request{uuid = UUID, username = Username, salt = Salt, token = Token, timestamp = Timestamp}) ->
-    iolist_to_binary([
-        marshal(?Bytes, UUID),
-        marshal(?Bytes, Username),
-        marshal(?Bytes, Salt),
-        marshal(?Bytes, Token),
-        marshal(?I32, Timestamp)
-    ]);
-encode0(#auth_reply{code = Code, payload = Payload}) ->
-    iolist_to_binary([
-        marshal(?I32, Code),
-        marshal(?Bytes, Payload)
-    ]);
-encode0(#jsonrpc_reply{result = Result, error = undefined}) ->
-    ResultBin = erlang:term_to_binary(#{<<"result">> => Result}),
-    iolist_to_binary([
-        marshal(?Bytes, ResultBin)
-    ]);
-encode0(#jsonrpc_reply{result = undefined, error = Error}) ->
-    ResultBin = erlang:term_to_binary(#{<<"error">> => Error}),
-    iolist_to_binary([
-        marshal(?Bytes, ResultBin)
-    ]);
-encode0(#pub{topic = Topic, qos = Qos, content = Content}) ->
-    iolist_to_binary([
-        marshal(?Bytes, Topic),
-        marshal(?I8, Qos),
-        marshal(?Bytes, Content)
-    ]);
-encode0(#command{command_type = CommandType, command = Command}) ->
-    iolist_to_binary([
-        marshal(?I32, CommandType),
-        marshal(?Bytes, Command)
-    ]);
-
-encode0(#jsonrpc_request{method = Method, params = Params}) ->
-    ReqBody = erlang:term_to_binary(#{<<"method">> => Method, <<"params">> => Params}),
-    iolist_to_binary([
-        marshal(?Bytes, ReqBody)
-    ]);
-encode0(#data{route_key = RouteKey, metric = Metric}) ->
-    iolist_to_binary([
-        marshal(?Bytes, RouteKey),
-        marshal(?Bytes, Metric)
-    ]);
-encode0(#task_event_stream{task_id = TaskId, type = Type, stream = Stream}) ->
-    iolist_to_binary([
-        marshal(?I32, TaskId),
-        marshal(?Bytes, Type),
-        marshal(?Bytes, Stream)
-    ]).
-
--spec decode(Bin :: binary()) -> {ok, Message :: any()} | error.
-decode(<<PacketType:8, Packet/binary>>) ->
-    case unmarshal(Packet) of
-        {ok, Fields} ->
-            decode0(PacketType, Fields);
-        error ->
-            error
-    end.
-decode0(?MESSAGE_AUTH_REQUEST, [UUID, Username, Salt, Token, Timestamp]) ->
-    {ok, #auth_request{uuid = UUID, username = Username, salt = Salt, token = Token, timestamp = Timestamp}};
-decode0(?MESSAGE_JSONRPC_REPLY, [ReplyBin]) ->
-    case erlang:binary_to_term(ReplyBin) of
-        #{<<"result">> := Result} ->
-            {ok, #jsonrpc_reply{result = Result}};
-        #{<<"error">> := Error} ->
-            {ok, #jsonrpc_reply{error = Error}};
-        _ ->
-            error
-    end;
-decode0(?MESSAGE_PUB, [Topic, Qos, Content]) ->
-    {ok, #pub{topic = Topic, qos = Qos, content = Content}};
-decode0(?MESSAGE_COMMAND, [CommandType, Command]) ->
-    {ok, #command{command_type = CommandType, command = Command}};
-decode0(?MESSAGE_AUTH_REPLY, [Code, Payload]) ->
-    {ok, #auth_reply{code = Code, payload = Payload}};
-decode0(?MESSAGE_JSONRPC_REQUEST, [ReqBody]) ->
-    #{<<"method">> := Method, <<"params">> := Params} = erlang:binary_to_term(ReqBody),
-    {ok, #jsonrpc_request{method = Method, params = Params}};
-decode0(?MESSAGE_DATA, [RouteKey, Metric]) ->
-    {ok, #data{route_key = RouteKey, metric = Metric}};
-decode0(?MESSAGE_EVENT_STREAM, [TaskId, Type, Stream]) ->
-    {ok, #task_event_stream{task_id = TaskId, type = Type, stream = Stream}};
-decode0(_, _) ->
-    error.
-
-%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-%%% helper methods
-%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-
--spec marshal(Type :: ?I8 | ?I16 | ?I32 | ?Bytes, Field :: integer() | binary()) -> binary().
-marshal(?I8, Field) when is_integer(Field) ->
-    <<?I8, Field:8>>;
-marshal(?I16, Field) when is_integer(Field) ->
-    <<?I16, Field:16>>;
-marshal(?I32, Field) when is_integer(Field) ->
-    <<?I32, Field:32>>;
-marshal(?Bytes, Field) when is_binary(Field) ->
-    Len = byte_size(Field),
-    <<?Bytes, Len:16, Field/binary>>.
-
--spec unmarshal(Bin :: binary()) -> {ok, Components :: [any()]} | error.
-unmarshal(Bin) when is_binary(Bin) ->
-    unmarshal(Bin, []).
-unmarshal(<<>>, Acc) ->
-    {ok, lists:reverse(Acc)};
-unmarshal(<<?I8, F:8, Rest/binary>>, Acc) ->
-    unmarshal(Rest, [F|Acc]);
-unmarshal(<<?I16, F:16, Rest/binary>>, Acc) ->
-    unmarshal(Rest, [F|Acc]);
-unmarshal(<<?I32, F:32, Rest/binary>>, Acc) ->
-    unmarshal(Rest, [F|Acc]);
-unmarshal(<<?Bytes, Len:16, F:Len/binary, Rest/binary>>, Acc) ->
-    unmarshal(Rest, [F|Acc]);
-unmarshal(_, _) ->
-    error.

src/host/iot_host.erl

@@ -10,6 +10,7 @@
 -author("aresei").
 -include("iot.hrl").
 -include("message.hrl").
+-include("message_pb.hrl").
 
 -behaviour(gen_statem).
 
@@ -94,52 +95,52 @@ attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) ->
 
 -spec get_containers(Pid :: pid()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
 get_containers(Pid) when is_pid(Pid) ->
-    Request = #jsonrpc_request{method = <<"get_containers">>, params = #{}},
+    Request = #'JsonRpcRequest'{method = <<"get_containers">>, params = erlang:term_to_binary(#{} )},
-    EncConfigBin = message_codec:encode(?MESSAGE_JSONRPC_REQUEST, Request),
+    gen_statem:call(Pid, {jsonrpc_call, self(), Request}).
-    gen_statem:call(Pid, {jsonrpc_call, self(), EncConfigBin}).
 
 -spec config_container(Pid :: pid(), ContainerName :: binary(), ConfigJson :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
 config_container(Pid, ContainerName, ConfigJson) when is_pid(Pid), is_binary(ContainerName), is_binary(ConfigJson) ->
-    Request = #jsonrpc_request{method = <<"config_container">>, params = #{<<"container_name">> => ContainerName, <<"config">> => ConfigJson}},
+    Request = #'JsonRpcRequest'{method = <<"config_container">>,
-    EncConfigBin = message_codec:encode(?MESSAGE_JSONRPC_REQUEST, Request),
+        params = erlang:term_to_binary(#{<<"container_name">> => ContainerName, <<"config">> => ConfigJson})},
-    gen_statem:call(Pid, {jsonrpc_call, self(), EncConfigBin}).
+    gen_statem:call(Pid, {jsonrpc_call, self(), Request}).
 
 -spec deploy_container(Pid :: pid(), TaskId :: integer(), Config :: map()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
 deploy_container(Pid, TaskId, Config) when is_pid(Pid), is_integer(TaskId), is_map(Config) ->
-    Request = #jsonrpc_request{method = <<"deploy">>, params = #{<<"task_id">> => TaskId, <<"config">> => Config}},
+    Request = #'JsonRpcRequest'{method = <<"deploy">>,
-    EncDeployBin = message_codec:encode(?MESSAGE_JSONRPC_REQUEST, Request),
+        params = erlang:term_to_binary(#{<<"task_id">> => TaskId, <<"config">> => Config})},
-    gen_statem:call(Pid, {jsonrpc_call, self(), EncDeployBin}).
+    gen_statem:call(Pid, {jsonrpc_call, self(), Request}).
 
 -spec start_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
 start_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
-    Request = #jsonrpc_request{method = <<"start_container">>, params = #{<<"container_name">> => ContainerName}},
+    Request = #'JsonRpcRequest'{method = <<"start_container">>,
-    EncCallBin = message_codec:encode(?MESSAGE_JSONRPC_REQUEST, Request),
+        params = erlang:term_to_binary(#{<<"container_name">> => ContainerName})},
-    gen_statem:call(Pid, {jsonrpc_call, self(), EncCallBin}).
+    gen_statem:call(Pid, {jsonrpc_call, self(), Request}).
 
 -spec stop_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
 stop_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
-    Request = #jsonrpc_request{method = <<"stop_container">>, params = #{<<"container_name">> => ContainerName}},
+    Request = #'JsonRpcRequest'{method = <<"stop_container">>,
-    EncCallBin = message_codec:encode(?MESSAGE_JSONRPC_REQUEST, Request),
+        params = erlang:term_to_binary(#{<<"container_name">> => ContainerName})},
-    gen_statem:call(Pid, {jsonrpc_call, self(), EncCallBin}).
+    gen_statem:call(Pid, {jsonrpc_call, self(), Request}).
 
 -spec kill_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
 kill_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
-    Request = #jsonrpc_request{method = <<"kill_container">>, params = #{<<"container_name">> => ContainerName}},
+    Request = #'JsonRpcRequest'{method = <<"kill_container">>,
-    EncCallBin = message_codec:encode(?MESSAGE_JSONRPC_REQUEST, Request),
+        params = erlang:term_to_binary(#{<<"container_name">> => ContainerName})},
-    gen_statem:call(Pid, {jsonrpc_call, self(), EncCallBin}).
+    gen_statem:call(Pid, {jsonrpc_call, self(), Request}).
 
 -spec remove_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
 remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
-    Request = #jsonrpc_request{method = <<"remove_container">>, params = #{<<"container_name">> => ContainerName}},
+    Request = #'JsonRpcRequest'{method = <<"remove_container">>,
-    EncCallBin = message_codec:encode(?MESSAGE_JSONRPC_REQUEST, Request),
+        params = erlang:term_to_binary(#{<<"container_name">> => ContainerName})},
-    gen_statem:call(Pid, {jsonrpc_call, self(), EncCallBin}).
+    gen_statem:call(Pid, {jsonrpc_call, self(), Request}).
 
 -spec await_reply(Ref :: reference(), Timeout :: integer()) -> {ok, Result :: binary()} | {error, Reason :: binary()}.
 await_reply(Ref, Timeout) when is_reference(Ref), is_integer(Timeout) ->
     receive
-        {jsonrpc_reply, Ref, #jsonrpc_reply{result = Result, error = undefined}} ->
+        {jsonrpc_reply, Ref, #'JsonRpcReply'{result = ResultBin, error = <<>>}} ->
-            {ok, Result};
+            {ok, erlang:binary_to_term(iolist_to_binary(ResultBin))};
-        {jsonrpc_reply, Ref, #jsonrpc_reply{result = undefined, error = #{<<"message">> := Message}}} ->
+        {jsonrpc_reply, Ref, #'JsonRpcReply'{result = <<>>, error = ErrorBin}} ->
+            #{<<"message">> := Message} = erlang:binary_to_term(iolist_to_binary(ErrorBin)),
             {error, Message}
     after Timeout ->
         {error, <<"timeout">>}
@@ -302,11 +303,12 @@ handle_event({call, From}, {attach_channel, _}, _, State = #state{uuid = UUID, c
     {keep_state, State, [{reply, From, {error, <<"channel existed">>}}]};
 
 %% 数据分发
-handle_event(cast, {handle, {data, #data{route_key = RouteKey0, metric = Metric}}}, ?STATE_ACTIVATED,
+handle_event(cast, {handle, {data, #'Data'{route_key = RouteKey0, metric = Metric}}}, ?STATE_ACTIVATED,
     State = #state{uuid = UUID, has_session = true}) ->
-    logger:debug("[iot_host] metric_data host: ~p, route_key: ~p, metric: ~p", [UUID, RouteKey0, Metric]),
+    RouteKey = iolist_to_binary(RouteKey0),
-    RouteKey = get_route_key(RouteKey0),
+    MetricBin = iolist_to_binary(Metric),
-    endpoint_subscription:publish(RouteKey, Metric),
+    logger:debug("[iot_host] metric_data host: ~p, route_key: ~p, metric: ~p", [UUID, RouteKey, MetricBin]),
+    endpoint_subscription:publish(get_route_key(RouteKey), MetricBin),
     {keep_state, State};
 
 %% ping的数据是通过aes加密后的，因此需要在有会话的情况下才行

src/transport/tcp/tcp_channel.erl

@@ -9,6 +9,7 @@
 -module(tcp_channel).
 -author("licheng5").
 -include("message.hrl").
+-include("message_pb.hrl").
 -behaviour(ranch_protocol).
 
 %% API
@@ -43,10 +44,10 @@ command(Pid, CommandType, Command) when is_pid(Pid), is_integer(CommandType), is
     gen_server:cast(Pid, {command, CommandType, Command}).
 
 %% 向通道中写入消息
--spec jsonrpc_call(Pid :: pid(), ReceiverPid :: pid(), CallBin :: binary()) -> Ref :: reference().
+-spec jsonrpc_call(Pid :: pid(), ReceiverPid :: pid(), Request :: #'JsonRpcRequest'{}) -> Ref :: reference().
-jsonrpc_call(Pid, ReceiverPid, CallBin) when is_pid(Pid), is_pid(ReceiverPid), is_binary(CallBin) ->
+jsonrpc_call(Pid, ReceiverPid, Request = #'JsonRpcRequest'{}) when is_pid(Pid), is_pid(ReceiverPid) ->
     Ref = make_ref(),
-    gen_server:cast(Pid, {jsonrpc_call, ReceiverPid, Ref, CallBin}),
+    gen_server:cast(Pid, {jsonrpc_call, ReceiverPid, Ref, Request}),
     Ref.
 
 %% 关闭方法
@@ -76,24 +77,33 @@ handle_call(_Request, _From, State) ->
 
 %% 发送消息, 基于pub/sub机制
 handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) ->
-    EncPub = message_codec:encode(?MESSAGE_PUB, #pub{topic = Topic, qos = Qos, content = Content}),
+    Encoded = message_pb:encode_msg(#'Pub'{topic = Topic, qos = Qos, content = Content}, 'Pub'),
+    EncPub = <<?MESSAGE_PUB, Encoded/binary>>,
     Transport:send(Socket, <<?PACKET_CAST, EncPub/binary>>),
     {noreply, State};
 
 %% 发送Command消息
 handle_cast({command, CommandType, Command}, State = #state{transport = Transport, socket = Socket}) ->
-    EncCommand = message_codec:encode(?MESSAGE_COMMAND, #command{command_type = CommandType, command = Command}),
+    Encoded = message_pb:encode_msg(#'Command'{command_type = CommandType, command = Command}, 'Command'),
+    EncCommand = <<?MESSAGE_COMMAND, Encoded/binary>>,
     Transport:send(Socket, <<?PACKET_CAST, EncCommand/binary>>),
     {noreply, State};
 
 %% 推送消息
-handle_cast({jsonrpc_call, ReceiverPid, Ref, CallBin}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
+handle_cast({jsonrpc_call, ReceiverPid, Ref, Request = #'JsonRpcRequest'{}}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
-    Transport:send(Socket, <<?PACKET_REQUEST, PacketId:32, CallBin/binary>>),
+    Encoded = message_pb:encode_msg(Request, 'JsonRpcRequest'),
+    EncRequest = <<?MESSAGE_JSONRPC_REQUEST, Encoded/binary>>,
+    Transport:send(Socket, <<?PACKET_REQUEST, PacketId:32, EncRequest/binary>>),
     {noreply, State#state{packet_id = PacketId + 1, inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}}.
 
 %% auth验证
-handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, RequestBin/binary>>}, State = #state{transport = Transport, socket = Socket}) ->
+handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, ?MESSAGE_AUTH_REQUEST, RequestBin/binary>>}, State = #state{transport = Transport, socket = Socket}) ->
-    {ok, #auth_request{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp}} = message_codec:decode(RequestBin),
+    #'AuthRequest'{uuid = UUID0, username = Username0, token = Token0, salt = Salt0, timestamp = Timestamp} =
+        message_pb:decode_msg(RequestBin, 'AuthRequest'),
+    UUID = iolist_to_binary(UUID0),
+    Username = iolist_to_binary(Username0),
+    Token = iolist_to_binary(Token0),
+    Salt = iolist_to_binary(Salt0),
     logger:debug("[ws_channel] auth uuid: ~p", [UUID]),
     case iot_auth:check(Username, Token, UUID, Salt, Timestamp) of
         true ->
@@ -108,20 +118,23 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, RequestBin/binary>>},
                         ok ->
                             %% 建立到host的monitor
                             erlang:monitor(process, HostPid),
-                            AuthReplyBin = message_codec:encode(?MESSAGE_AUTH_REPLY, #auth_reply{code = 0, payload = <<"ok">>}),
+                            Encoded = message_pb:encode_msg(#'AuthReply'{code = 0, payload = <<"ok">>}, 'AuthReply'),
+                            AuthReplyBin = <<?MESSAGE_AUTH_REPLY, Encoded/binary>>,
                             Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
 
                             {noreply, State#state{uuid = UUID, host_pid = HostPid}};
                         {denied, Reason} when is_binary(Reason) ->
                             erlang:monitor(process, HostPid),
-                            AuthReplyBin = message_codec:encode(?MESSAGE_AUTH_REPLY, #auth_reply{code = 1, payload = Reason}),
+                            Encoded = message_pb:encode_msg(#'AuthReply'{code = 1, payload = Reason}, 'AuthReply'),
+                            AuthReplyBin = <<?MESSAGE_AUTH_REPLY, Encoded/binary>>,
                             Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, AuthReplyBin/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) ->
-                            AuthReplyBin = message_codec:encode(?MESSAGE_AUTH_REPLY, #auth_reply{code = 2, payload = Reason}),
+                            Encoded = message_pb:encode_msg(#'AuthReply'{code = 2, payload = Reason}, 'AuthReply'),
+                            AuthReplyBin = <<?MESSAGE_AUTH_REPLY, Encoded/binary>>,
                             Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
                             logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
 
@@ -132,19 +145,32 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, RequestBin/binary>>},
             logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]),
             {stop, State}
     end;
+handle_info({tcp, Socket, <<?PACKET_REQUEST, _PacketId:32, MsgType:8, _/binary>>}, State = #state{socket = Socket}) ->
+    logger:warning("[ws_channel] unsupported request message type: ~p", [MsgType]),
+    {stop, State};
 
-handle_info({tcp, Socket, <<?PACKET_CAST, CastBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
+handle_info({tcp, Socket, <<?PACKET_CAST, ?MESSAGE_DATA, CastBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
-    {ok, CastMessage} = message_codec:decode(CastBin),
+    CastMessage = message_pb:decode_msg(CastBin, 'Data'),
     case CastMessage of
-        #data{} = Data ->
+        #'Data'{} = Data ->
-            iot_host:handle(HostPid, {data, Data});
+            iot_host:handle(HostPid, {data, Data})
-        #task_event_stream{task_id = TaskId, type = <<"close">>, stream = Reason} ->
-            iot_event_stream_observer:stream_close(TaskId, Reason);
-        #task_event_stream{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, <<?PACKET_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">>; Type0 =:= [<<"close">>] ->
+            iot_event_stream_observer:stream_close(TaskId, iolist_to_binary(Reason0));
+        #'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream} ->
+            TypeBin = iolist_to_binary(Type),
+            StreamBin = iolist_to_binary(Stream),
+            logger:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, TypeBin, StreamBin]),
+            iot_event_stream_observer:stream_data(TaskId, TypeBin, StreamBin)
+    end,
+    {noreply, State};
+handle_info({tcp, Socket, <<?PACKET_CAST, MsgType:8, _/binary>>}, State = #state{socket = Socket}) ->
+    logger:warning("[tcp_channel] unsupported cast message type: ~p", [MsgType]),
+    {noreply, State};
 
 %handle_info({tcp, Socket, <<?PACKET_PING, PingData/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
 %    Ping = message_pb:decode_msg(PingData, ping),
@@ -153,7 +179,8 @@ handle_info({tcp, Socket, <<?PACKET_CAST, CastBin/binary>>}, State = #state{sock
 
 %% 主机端的消息响应
 handle_info({tcp, Socket, <<?PACKET_RESPONSE, PacketId:32, ResponseBin/binary>>}, State = #state{socket = Socket, inflight = Inflight}) when PacketId > 0 ->
-    {ok, RpcReply} = message_codec:decode(ResponseBin),
+    <<?MESSAGE_JSONRPC_REPLY, ReplyBin/binary>> = ResponseBin,
+    RpcReply = message_pb:decode_msg(ReplyBin, 'JsonRpcReply'),
     case maps:take(PacketId, Inflight) of
         error ->
             {noreply, State};