fix

config/sys-dev.config

@@ -25,7 +25,7 @@
       {port, 18080}
     ]},
 
-    {api_url, "http://100.123.0.4/api/v1"},
+    {api_url, "http://127.0.0.1:18090/simulator"},
     %% 支持的协议
     {endpoints, [
       {support_protocols, [

include/message_pb.hrl

@@ -16,36 +16,6 @@
         }).
 -endif.
 
--ifndef('REPLYFRAME_PB_H').
--define('REPLYFRAME_PB_H', true).
--record('ReplyFrame',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         reply                  :: {result, message_pb:'ReplyResult'()} | {error, message_pb:'ReplyError'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('REPLYRESULT_PB_H').
--define('REPLYRESULT_PB_H', true).
--record('ReplyResult',
-        {data = <<>>            :: iodata() | undefined % = 1, optional
-        }).
--endif.
-
--ifndef('REPLYERROR_PB_H').
--define('REPLYERROR_PB_H', true).
--record('ReplyError',
-        {code = 0               :: integer() | undefined, % = 1, optional, 32 bits
-         message = <<>>         :: unicode:chardata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('CASTFRAME_PB_H').
--define('CASTFRAME_PB_H', true).
--record('CastFrame',
-        {body                   :: {pub, message_pb:'Pub'()} | {command, message_pb:'Command'()} | {data, message_pb:'Data'()} | {event_stream, message_pb:'TaskEventStream'()} | undefined % oneof
-        }).
--endif.
-
 -ifndef('AUTHREQUEST_PB_H').
 -define('AUTHREQUEST_PB_H', true).
 -record('AuthRequest',
@@ -57,40 +27,63 @@
         }).
 -endif.
 
--ifndef('PUB_PB_H').
+-ifndef('REPLYFRAME.ERROR_PB_H').
--define('PUB_PB_H', true).
+-define('REPLYFRAME.ERROR_PB_H', true).
--record('Pub',
+-record('ReplyFrame.Error',
+        {code = 0               :: integer() | undefined, % = 1, optional, 32 bits
+         message = <<>>         :: unicode:chardata() | undefined % = 2, optional
+        }).
+-endif.
+
+-ifndef('REPLYFRAME_PB_H').
+-define('REPLYFRAME_PB_H', true).
+-record('ReplyFrame',
+        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
+         reply                  :: {result, iodata()} | {error, message_pb:'ReplyFrame.Error'()} | undefined % oneof
+        }).
+-endif.
+
+-ifndef('CASTFRAME.PUB_PB_H').
+-define('CASTFRAME.PUB_PB_H', true).
+-record('CastFrame.Pub',
         {topic = <<>>           :: iodata() | undefined, % = 1, optional
          qos = 0                :: integer() | undefined, % = 2, optional, 32 bits
          content = <<>>         :: iodata() | undefined % = 3, optional
         }).
 -endif.
 
--ifndef('COMMAND_PB_H').
+-ifndef('CASTFRAME.COMMAND_PB_H').
--define('COMMAND_PB_H', true).
+-define('CASTFRAME.COMMAND_PB_H', true).
--record('Command',
+-record('CastFrame.Command',
         {command_type = 0       :: integer() | undefined, % = 1, optional, 32 bits
          command = <<>>         :: iodata() | undefined % = 2, optional
         }).
 -endif.
 
--ifndef('DATA_PB_H').
+-ifndef('CASTFRAME.DATA_PB_H').
--define('DATA_PB_H', true).
+-define('CASTFRAME.DATA_PB_H', true).
--record('Data',
+-record('CastFrame.Data',
         {route_key = <<>>       :: iodata() | undefined, % = 1, optional
          metric = <<>>          :: iodata() | undefined % = 2, optional
         }).
 -endif.
 
--ifndef('TASKEVENTSTREAM_PB_H').
+-ifndef('CASTFRAME.TASKEVENT_PB_H').
--define('TASKEVENTSTREAM_PB_H', true).
+-define('CASTFRAME.TASKEVENT_PB_H', true).
--record('TaskEventStream',
+-record('CastFrame.TaskEvent',
         {task_id = 0            :: integer() | undefined, % = 1, optional, 32 bits
          type = <<>>            :: iodata() | undefined, % = 2, optional
          stream = <<>>          :: iodata() | undefined % = 3, optional
         }).
 -endif.
 
+-ifndef('CASTFRAME_PB_H').
+-define('CASTFRAME_PB_H', true).
+-record('CastFrame',
+        {body                   :: {pub, message_pb:'CastFrame.Pub'()} | {command, message_pb:'CastFrame.Command'()} | {data, message_pb:'CastFrame.Data'()} | {task_event, message_pb:'CastFrame.TaskEvent'()} | undefined % oneof
+        }).
+-endif.
+
 -ifndef('CONTAINERREF_PB_H').
 -define('CONTAINERREF_PB_H', true).
 -record('ContainerRef',

src/host/iot_host.erl

@@ -130,12 +130,12 @@ remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName)
     {ok, Result :: binary()} | {error, Code :: integer(), Reason :: binary()}.
 await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_reference(Ref), is_integer(Timeout) ->
     receive
-        {rpc_reply, Ref, #'ReplyFrame'{reply = {result, #'ReplyResult'{data = ResultBin}}}} ->
+        {request_reply, Ref, #'ReplyFrame'{reply = {result, ResultBin}}} ->
             {ok, iolist_to_binary(ResultBin)};
-        {rpc_reply, Ref, #'ReplyFrame'{reply = {error, #'ReplyError'{code = Code, message = Message}}}} ->
+        {request_reply, Ref, #'ReplyFrame'{reply = {error, #'ReplyFrame.Error'{code = Code, message = Message}}}} ->
             {error, Code, unicode:characters_to_binary(Message)}
     after Timeout ->
-        ok = gen_statem:call(Pid, {cancel_rpc_call, Ref}),
+        ok = gen_statem:call(Pid, {cancel_request_call, Ref}),
         flush_reply(Ref),
         {error, -1, <<"timeout">>}
     end.
@@ -214,18 +214,6 @@ handle_event({call, From}, get_status, _, State = #state{channel_pid = ChannelPi
     },
     {keep_state, State, [{reply, From, {ok, Reply}}]};
 
-%% 只要channel存在，就负责将消息推送到边缘端主机
-handle_event({call, From}, {rpc_call, ReceiverPid, RpcCall}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) ->
-    case HasSession andalso is_pid(ChannelPid) of
-        true ->
-            %% 通过websocket发送请求
-            Ref = ssl_channel:rpc_call(ChannelPid, ReceiverPid, RpcCall),
-            {keep_state, State, [{reply, From, {ok, Ref}}]};
-        false ->
-            logger:debug("[iot_host] uuid: ~p, invalid state: ~p", [UUID, state_map(State)]),
-            {keep_state, State, [{reply, From, {error, <<"主机离线，发送请求失败"/utf8>>}}]}
-    end;
-
 handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) ->
     case HasSession andalso is_pid(ChannelPid) of
         true ->
@@ -236,10 +224,10 @@ handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #s
             {keep_state, State, [{reply, From, {error, <<"主机离线，发送请求失败"/utf8>>}}]}
     end;
 
-handle_event({call, From}, {cancel_rpc_call, Ref}, _, State = #state{channel_pid = ChannelPid}) ->
+handle_event({call, From}, {cancel_request_call, Ref}, _, State = #state{channel_pid = ChannelPid}) ->
     case is_pid(ChannelPid) of
         true ->
-            ok = ssl_channel:cancel_rpc_call(ChannelPid, Ref),
+            ok = ssl_channel:cancel_request_call(ChannelPid, Ref),
             {keep_state, State, [{reply, From, ok}]};
         false ->
             {keep_state, State, [{reply, From, ok}]}
@@ -316,7 +304,7 @@ 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, #'CastFrame.Data'{route_key = RouteKey0, metric = Metric}}}, ?STATE_ACTIVATED,
     State = #state{uuid = UUID, has_session = true}) ->
     RouteKey = iolist_to_binary(RouteKey0),
     MetricBin = iolist_to_binary(Metric),
@@ -412,7 +400,7 @@ state_map(#state{host_id = HostId, uuid = UUID, has_session = HasSession, heartb
 
 flush_reply(Ref) ->
     receive
-        {rpc_reply, Ref, _Reply} ->
+        {request_reply, Ref, _Reply} ->
             ok
     after 0 ->
         ok

src/iot_app.erl

@@ -51,6 +51,7 @@ start_http_server() ->
             {"/host/[...]", http_protocol, [host_handler]},
             {"/container/[...]", http_protocol, [container_handler]},
             {"/endpoint/[...]", http_protocol, [endpoint_handler]},
+            {"/simulator/[...]", http_protocol, [simulator_api_handler]},
             {"/event_stream", event_stream_handler, []}
         ]}
     ]),

src/simulator/simulator_api_handler.erl

@@ -0,0 +1,156 @@
+%%%-------------------------------------------------------------------
+%%% @author Codex
+%%% @doc
+%%% 本地模拟 control api，用于替代 iot_api_client 依赖的远端接口。
+%%% @end
+%%%-------------------------------------------------------------------
+-module(simulator_api_handler).
+
+-include("endpoint.hrl").
+
+-export([handle_request/4]).
+
+-define(HOST_UUID, <<"sim-host-001">>).
+-define(HOST_ID, 1).
+-define(DEVICE_UUID, <<"sim-device-001">>).
+-define(DEVICE_ID, 1).
+-define(ENDPOINT_ID, 1).
+-define(TABLE, ?MODULE).
+
+-spec handle_request(string(), string(), map(), map()) ->
+    {ok, non_neg_integer(), iodata()}.
+handle_request("GET", "/simulator/get_all_hosts", _, _) ->
+    {ok, 200, iot_util:json_data([?HOST_UUID])};
+handle_request("GET", "/simulator/get_host_by_uuid", #{<<"uuid">> := UUID}, _) when is_binary(UUID) ->
+    reply_host_by_uuid(UUID);
+handle_request("GET", "/simulator/get_host_by_id", #{<<"host_id">> := HostIdBin}, _) when is_binary(HostIdBin) ->
+    case binary_to_integer(HostIdBin) of
+        ?HOST_ID ->
+            {ok, 200, iot_util:json_data(host_info())};
+        _ ->
+            {ok, 200, iot_util:json_error(404, <<"host not found">>)}
+    end;
+handle_request("POST", "/simulator/change_host_status", _, #{<<"uuid">> := UUID, <<"new_status">> := Status})
+        when is_binary(UUID), is_integer(Status) ->
+    reply_change_host_status(UUID, Status);
+handle_request("GET", "/simulator/get_host_devices", #{<<"host_id">> := HostIdBin}, _) when is_binary(HostIdBin) ->
+    case binary_to_integer(HostIdBin) of
+        ?HOST_ID ->
+            {ok, 200, iot_util:json_data([device_info()])};
+        _ ->
+            {ok, 200, iot_util:json_error(404, <<"host not found">>)}
+    end;
+handle_request("GET", "/simulator/get_device_by_uuid", #{<<"device_uuid">> := DeviceUUID}, _) when is_binary(DeviceUUID) ->
+    case DeviceUUID of
+        ?DEVICE_UUID ->
+            {ok, 200, iot_util:json_data(device_info())};
+        _ ->
+            {ok, 200, iot_util:json_error(404, <<"device not found">>)}
+    end;
+handle_request("POST", "/simulator/change_device_status", _, #{<<"device_uuid">> := DeviceUUID, <<"new_status">> := Status})
+        when is_binary(DeviceUUID), is_integer(Status) ->
+    reply_change_device_status(DeviceUUID, Status);
+handle_request("GET", "/simulator/get_all_endpoints", _, _) ->
+    {ok, 200, iot_util:json_data([endpoint_info()])};
+handle_request("GET", "/simulator/get_endpoint", #{<<"id">> := IdBin}, _) when is_binary(IdBin) ->
+    case binary_to_integer(IdBin) of
+        ?ENDPOINT_ID ->
+            {ok, 200, iot_util:json_data(endpoint_info())};
+        _ ->
+            {ok, 200, iot_util:json_error(404, <<"endpoint not found">>)}
+    end;
+handle_request("POST", "/simulator/endpoint_sink", _, Body) ->
+    logger:debug("[simulator_api_handler] receive endpoint sink body: ~p", [Body]),
+    {ok, 200, iot_util:json_data(<<"ok">>)};
+handle_request(_, Path, _, _) ->
+    PathBin = list_to_binary(Path),
+    {ok, 200, iot_util:json_error(-1, <<"url: ", PathBin/binary, " not found">>)}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+-spec reply_host_by_uuid(binary()) -> {ok, non_neg_integer(), iodata()}.
+reply_host_by_uuid(?HOST_UUID) ->
+    {ok, 200, iot_util:json_data(host_info())};
+reply_host_by_uuid(_) ->
+    {ok, 200, iot_util:json_error(404, <<"host not found">>)}.
+
+-spec reply_change_host_status(binary(), integer()) -> {ok, non_neg_integer(), iodata()}.
+reply_change_host_status(?HOST_UUID, Status) ->
+    ok = put_state({host_status, ?HOST_UUID}, Status),
+    {ok, 200, iot_util:json_data(1)};
+reply_change_host_status(_, _) ->
+    {ok, 200, iot_util:json_error(404, <<"host not found">>)}.
+
+-spec reply_change_device_status(binary(), integer()) -> {ok, non_neg_integer(), iodata()}.
+reply_change_device_status(?DEVICE_UUID, Status) ->
+    ok = put_state({device_status, ?DEVICE_UUID}, Status),
+    {ok, 200, iot_util:json_data(1)};
+reply_change_device_status(_, _) ->
+    {ok, 200, iot_util:json_error(404, <<"device not found">>)}.
+
+-spec host_info() -> map().
+host_info() ->
+    #{
+        <<"id">> => ?HOST_ID,
+        <<"uuid">> => ?HOST_UUID,
+        <<"authorize_status">> => 1,
+        <<"status">> => get_state({host_status, ?HOST_UUID}, 1)
+    }.
+
+-spec device_info() -> map().
+device_info() ->
+    #{
+        <<"id">> => ?DEVICE_ID,
+        <<"host_id">> => ?HOST_ID,
+        <<"device_uuid">> => ?DEVICE_UUID,
+        <<"status">> => get_state({device_status, ?DEVICE_UUID}, 1)
+    }.
+
+-spec endpoint_info() -> map().
+endpoint_info() ->
+    #{
+        <<"id">> => ?ENDPOINT_ID,
+        <<"matcher">> => <<"simulator.metric">>,
+        <<"title">> => <<"Simulator HTTP Endpoint">>,
+        <<"type">> => <<"http">>,
+        <<"config">> => #{
+            <<"url">> => <<"http://127.0.0.1:18090/simulator/endpoint_sink">>,
+            <<"pool_size">> => 4
+        },
+        <<"status">> => 1,
+        <<"updated_at">> => iot_util:timestamp(),
+        <<"created_at">> => iot_util:timestamp()
+    }.
+
+-spec ensure_table() -> ets:tid().
+ensure_table() ->
+    case ets:info(?TABLE) of
+        undefined ->
+            try ets:new(?TABLE, [named_table, public, set]) of
+                Tid ->
+                    Tid
+            catch
+                error:badarg ->
+                    ?TABLE
+            end;
+        _ ->
+            ?TABLE
+    end.
+
+-spec get_state(term(), term()) -> term().
+get_state(Key, Default) ->
+    _ = ensure_table(),
+    case ets:lookup(?TABLE, Key) of
+        [{_, Value}] ->
+            Value;
+        [] ->
+            Default
+    end.
+
+-spec put_state(term(), term()) -> ok.
+put_state(Key, Value) ->
+    _ = ensure_table(),
+    true = ets:insert(?TABLE, {Key, Value}),
+    ok.

src/transport/http/container_handler.erl

@@ -25,9 +25,9 @@ handle_request("GET", "/container/get_all", #{<<"uuid">> := UUID}, _) when is_bi
                 {ok, Ref} ->
                     case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of
                         {ok, Result} ->
-                            {ok, 200, rpc_success_response(Result)};
+                            {ok, 200, request_success_response(Result)};
                         {error, Code, Reason} ->
-                            rpc_error_http_response(Code, Reason)
+                            request_error_http_response(Code, Reason)
                     end;
                 {error, Reason} when is_binary(Reason) ->
                     {ok, 200, iot_util:json_error(-1, Reason)}
@@ -50,9 +50,9 @@ handle_request("POST", "/container/push_config", _,
                 {ok, Ref} ->
                     case iot_host:await_reply(Pid, Ref, Timeout) of
                         {ok, Result} ->
-                            {ok, 200, rpc_success_response(Result)};
+                            {ok, 200, request_success_response(Result)};
                         {error, Code, Reason} ->
-                            rpc_error_http_response(Code, Reason)
+                            request_error_http_response(Code, Reason)
                     end;
                 {error, Reason} when is_binary(Reason) ->
                     {ok, 200, iot_util:json_error(-1, Reason)}
@@ -70,9 +70,9 @@ handle_request("POST", "/container/deploy", _, #{<<"uuid">> := UUID, <<"task_id"
                 {ok, Ref} ->
                     case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of
                         {ok, Result} ->
-                            {ok, 200, rpc_success_response(Result)};
+                            {ok, 200, request_success_response(Result)};
                         {error, Code, Reason} ->
-                            rpc_error_http_response(Code, Reason)
+                            request_error_http_response(Code, Reason)
                     end;
                 {error, Reason} when is_binary(Reason) ->
                     {ok, 200, iot_util:json_error(400, Reason)}
@@ -89,9 +89,9 @@ handle_request("POST", "/container/start", _, #{<<"uuid">> := UUID, <<"container
                 {ok, Ref} ->
                     case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of
                         {ok, Result} ->
-                            {ok, 200, rpc_success_response(Result)};
+                            {ok, 200, request_success_response(Result)};
                         {error, Code, Reason} ->
-                            rpc_error_http_response(Code, Reason)
+                            request_error_http_response(Code, Reason)
                     end;
                 {error, Reason} when is_binary(Reason) ->
                     {ok, 200, iot_util:json_error(400, Reason)}
@@ -108,9 +108,9 @@ handle_request("POST", "/container/stop", _, #{<<"uuid">> := UUID, <<"container_
                 {ok, Ref} ->
                     case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of
                         {ok, Result} ->
-                            {ok, 200, rpc_success_response(Result)};
+                            {ok, 200, request_success_response(Result)};
                         {error, Code, Reason} ->
-                            rpc_error_http_response(Code, Reason)
+                            request_error_http_response(Code, Reason)
                     end;
                 {error, Reason} when is_binary(Reason) ->
                     {ok, 200, iot_util:json_error(400, Reason)}
@@ -126,9 +126,9 @@ handle_request("POST", "/container/kill", _, #{<<"uuid">> := UUID, <<"container_
                 {ok, Ref} ->
                     case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of
                         {ok, Result} ->
-                            {ok, 200, rpc_success_response(Result)};
+                            {ok, 200, request_success_response(Result)};
                         {error, Code, Reason} ->
-                            rpc_error_http_response(Code, Reason)
+                            request_error_http_response(Code, Reason)
                     end;
                 {error, Reason} when is_binary(Reason) ->
                     {ok, 200, iot_util:json_error(400, Reason)}
@@ -145,9 +145,9 @@ handle_request("POST", "/container/remove", _, #{<<"uuid">> := UUID, <<"containe
                 {ok, Ref} ->
                     case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of
                         {ok, Result} ->
-                            {ok, 200, rpc_success_response(Result)};
+                            {ok, 200, request_success_response(Result)};
                         {error, Code, Reason} ->
-                            rpc_error_http_response(Code, Reason)
+                            request_error_http_response(Code, Reason)
                     end;
                 {error, Reason} when is_binary(Reason) ->
                     {ok, 200, iot_util:json_error(400, Reason)}
@@ -158,7 +158,7 @@ handle_request(_, Path, _, _) ->
     Path1 = list_to_binary(Path),
     {ok, 200, iot_util:json_error(-1, <<"url: ", Path1/binary, " not found">>)}.
 
-rpc_success_response(Result) when is_binary(Result) ->
+request_success_response(Result) when is_binary(Result) ->
     case decode_json_bytes(Result) of
         {ok, Data} ->
             iot_util:json_data(Data);
@@ -166,7 +166,7 @@ rpc_success_response(Result) when is_binary(Result) ->
             iot_util:json_data(Result)
     end.
 
-rpc_error_response(Code, Reason) when is_integer(Code), is_binary(Reason) ->
+request_error_response(Code, Reason) when is_integer(Code), is_binary(Reason) ->
     case decode_json_bytes(Reason) of
         {ok, #{<<"message">> := Message}} when is_binary(Message) ->
             iot_util:json_error(Code, Message);
@@ -176,17 +176,17 @@ rpc_error_response(Code, Reason) when is_integer(Code), is_binary(Reason) ->
             iot_util:json_error(Code, Reason)
     end.
 
--spec rpc_error_http_response(Code :: integer(), Reason :: binary()) ->
+-spec request_error_http_response(Code :: integer(), Reason :: binary()) ->
     {ok, HttpStatus :: integer(), Body :: iolist()}.
-rpc_error_http_response(Code, Reason) when is_integer(Code), is_binary(Reason) ->
+request_error_http_response(Code, Reason) when is_integer(Code), is_binary(Reason) ->
-    {ok, rpc_error_status(Code), rpc_error_response(Code, Reason)}.
+    {ok, request_error_status(Code), request_error_response(Code, Reason)}.
 
--spec rpc_error_status(Code :: integer()) -> integer().
+-spec request_error_status(Code :: integer()) -> integer().
-rpc_error_status(Code) when is_integer(Code), Code >= 400, Code < 600 ->
+request_error_status(Code) when is_integer(Code), Code >= 400, Code < 600 ->
     Code;
-rpc_error_status(-1) ->
+request_error_status(-1) ->
     504;
-rpc_error_status(_) ->
+request_error_status(_) ->
     400.
 
 decode_json_bytes(Data) when is_binary(Data) ->

src/transport/tcp/ssl_channel.erl

@@ -16,7 +16,7 @@
 -define(INFLIGHT_TIMEOUT, 60000).
 
 %% API
--export([pub/4, rpc_call/3, container_call/3, cancel_rpc_call/2, command/3]).
+-export([pub/4, container_call/3, cancel_request_call/2, command/3]).
 
 -export([start_link/3, stop/2]).
 %% gen_server callbacks
@@ -52,22 +52,15 @@ pub(Pid, Topic, Qos, Content) when is_pid(Pid), is_binary(Topic), is_integer(Qos
 command(Pid, CommandType, Command) when is_pid(Pid), is_integer(CommandType), is_binary(Command) ->
     gen_server:cast(Pid, {command, CommandType, Command}).
 
-%% 向通道中写入消息
--spec rpc_call(Pid :: pid(), ReceiverPid :: pid(), Request :: {Method :: binary(), Params :: binary()}) -> Ref :: reference().
-rpc_call(Pid, ReceiverPid, {Method, Params}) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Method), is_binary(Params) ->
-    Ref = make_ref(),
-    gen_server:cast(Pid, {unsupported_rpc_call, ReceiverPid, Ref, Method, Params}),
-    Ref.
-
 -spec container_call(Pid :: pid(), ReceiverPid :: pid(), Request :: message_pb:'ContainerRequest'()) -> Ref :: reference().
 container_call(Pid, ReceiverPid, Request) when is_pid(Pid), is_pid(ReceiverPid), is_record(Request, 'ContainerRequest') ->
     Ref = make_ref(),
     gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {container_request, Request}}),
     Ref.
 
--spec cancel_rpc_call(Pid :: pid(), Ref :: reference()) -> ok.
+-spec cancel_request_call(Pid :: pid(), Ref :: reference()) -> ok.
-cancel_rpc_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) ->
+cancel_request_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) ->
-    gen_server:call(Pid, {cancel_rpc_call, Ref}).
+    gen_server:call(Pid, {cancel_request_call, Ref}).
 
 %% 关闭方法
 -spec stop(Pid :: pid(), Reason :: any()) -> no_return().
@@ -91,7 +84,7 @@ init(Ref, Transport, _Opts = []) ->
     % erlang:start_timer(?PING_TICKER, self(), ping_ticker),
     gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}).
 
-handle_call({cancel_rpc_call, Ref}, _From, State = #state{inflight = Inflight}) ->
+handle_call({cancel_request_call, Ref}, _From, State = #state{inflight = Inflight}) ->
     case take_inflight_by_ref(Ref, Inflight) of
         {ok, #inflight_request{timer_ref = TimerRef}, NInflight} ->
             erlang:cancel_timer(TimerRef),
@@ -105,7 +98,7 @@ handle_call(_Request, _From, State) ->
 %% 发送消息, 基于pub/sub机制
 handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) ->
     Encoded = message_pb:encode_msg(#'CastFrame'{
-        body = {pub, #'Pub'{topic = Topic, qos = Qos, content = Content}}
+        body = {pub, #'CastFrame.Pub'{topic = Topic, qos = Qos, content = Content}}
     }),
     Transport:send(Socket, <<?FRAME_CAST, Encoded/binary>>),
     {noreply, State};
@@ -113,24 +106,11 @@ handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, so
 %% 发送Command消息
 handle_cast({command, CommandType, Command}, State = #state{transport = Transport, socket = Socket}) ->
     Encoded = message_pb:encode_msg(#'CastFrame'{
-        body = {command, #'Command'{command_type = CommandType, command = Command}}
+        body = {command, #'CastFrame.Command'{command_type = CommandType, command = Command}}
     }),
     Transport:send(Socket, <<?FRAME_CAST, Encoded/binary>>),
     {noreply, State};
 
-handle_cast({unsupported_rpc_call, ReceiverPid, Ref, Method, _Params}, State) ->
-    Reply = #'ReplyFrame'{
-        packet_id = 0,
-        reply = {error, #'ReplyError'{code = -1, message = <<"rpc_request is not supported by current message.proto">>}}
-    },
-    case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
-        true ->
-            ReceiverPid ! {rpc_reply, Ref, Reply};
-        false ->
-            logger:warning("[ssl_channel] drop unsupported rpc_call reply because receiver is dead, method: ~p", [Method])
-    end,
-    {noreply, State};
-
 %% 推送需要响应的请求
 handle_cast({request_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
     Encoded = message_pb:encode_msg(#'RequestFrame'{
@@ -243,27 +223,24 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId,
                     case iot_host:attach_channel(HostPid, self()) of
                         ok ->
                             erlang:monitor(process, HostPid),
-                            send_reply_frame(Transport, Socket, PacketId, {result, #'ReplyResult'{data = <<"ok">>}}),
+                            send_reply_frame(Transport, Socket, PacketId, {result, <<"ok">>}),
                             {noreply, State#state{uuid = UUID, host_pid = HostPid}};
                         {denied, Reason} when is_binary(Reason) ->
                             erlang:monitor(process, HostPid),
-                            send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyError'{code = 1, message = Reason}}),
+                            send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyFrame.Error'{code = 1, message = 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_reply_frame(Transport, Socket, PacketId, {error, #'ReplyError'{code = 2, message = Reason}}),
+                            send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyFrame.Error'{code = 2, message = Reason}}),
                             logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
                             {stop, State}
                     end
             end;
         false ->
-            send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyError'{code = 2, message = <<"auth failed">>}}),
+            send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyFrame.Error'{code = 2, message = <<"auth failed">>}}),
             logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]),
             {stop, State}
     end;
-handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {rpc_request, RpcRequest}}, _Transport, _Socket, State) ->
-    logger:warning("[ws_channel] unsupported request message type: rpc_request, packet_id: ~p, request: ~p", [PacketId, RpcRequest]),
-    {stop, State};
 handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {container_request, ContainerRequest}}, _Transport, _Socket, State) ->
     logger:warning("[ws_channel] unsupported request message type: container_request, packet_id: ~p, request: ~p", [PacketId, ContainerRequest]),
     {stop, State};
@@ -276,17 +253,17 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = undefined}, _T
 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_cast_frame(#'CastFrame'{body = {task_event, CastMessage}}, HostPid, State) when is_pid(HostPid) ->
     handle_event_stream_frame(CastMessage),
     {noreply, State};
 handle_cast_frame(#'CastFrame'{body = Body}, _HostPid, State) ->
     logger:warning("[ssl_channel] unsupported cast message type: command, body: ~p", [Body]),
     {noreply, State}.
 
--spec handle_event_stream_frame(message_pb:'TaskEventStream'()) -> any().
+-spec handle_event_stream_frame(message_pb:'CastFrame.TaskEvent'()) -> any().
-handle_event_stream_frame(#'TaskEventStream'{task_id = TaskId, type = Type0, stream = Reason0}) when Type0 =:= <<"close">> ->
+handle_event_stream_frame(#'CastFrame.TaskEvent'{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}) ->
+handle_event_stream_frame(#'CastFrame.TaskEvent'{task_id = TaskId, type = Type, stream = Stream}) ->
     logger:debug("[ssl_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]),
     iot_event_stream_observer:stream_data(TaskId, Type, Stream).
 
@@ -301,7 +278,7 @@ handle_reply_frame(#'ReplyFrame'{packet_id = PacketId, reply = Reply}, Inflight,
             erlang:cancel_timer(TimerRef),
             case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
                 true ->
-                    ReceiverPid ! {rpc_reply, Ref, #'ReplyFrame'{packet_id = PacketId, reply = Reply}};
+                    ReceiverPid ! {request_reply, Ref, #'ReplyFrame'{packet_id = PacketId, reply = Reply}};
                 false ->
                     logger:warning("[ws_channel] get reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [Reply, PacketId])
             end,