fix

config/sys-dev.config

@@ -8,7 +8,7 @@
     ]},
 
     {ssl_server, [
-      {port, 18092},
+      {port, 443},
       {acceptors, 500},
       {max_connections, 10240},
       {backlog, 10240}

include/message_pb.hrl

@@ -12,15 +12,30 @@
 -define('REQUESTFRAME_PB_H', true).
 -record('RequestFrame',
         {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         body                   :: {auth_request, message_pb:'AuthRequest'()} | {rpc_request, message_pb:'RpcRequest'()} | {container_request, message_pb:'ContainerRequest'()} | undefined % oneof
+         body                   :: {auth_request, message_pb:'AuthRequest'()} | {container_request, message_pb:'ContainerRequest'()} | undefined % oneof
         }).
 -endif.
 
--ifndef('RESPONSEFRAME_PB_H').
+-ifndef('REPLYFRAME_PB_H').
--define('RESPONSEFRAME_PB_H', true).
+-define('REPLYFRAME_PB_H', true).
--record('ResponseFrame',
+-record('ReplyFrame',
         {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         body                   :: {auth_reply, message_pb:'AuthReply'()} | {rpc_reply, message_pb:'RpcReply'()} | undefined % oneof
+         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.
 
@@ -31,6 +46,51 @@
         }).
 -endif.
 
+-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('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('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.
+
 -ifndef('CONTAINERREF_PB_H').
 -define('CONTAINERREF_PB_H', true).
 -record('ContainerRef',
@@ -215,87 +275,4 @@
         }).
 -endif.
 
--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('RPCREQUEST_PB_H').
--define('RPCREQUEST_PB_H', true).
--record('RpcRequest',
-        {method = <<>>          :: iodata() | undefined, % = 1, optional
-         params = <<>>          :: iodata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('RPCREPLY.RPCRESULT_PB_H').
--define('RPCREPLY.RPCRESULT_PB_H', true).
--record('RpcReply.RpcResult',
-        {data = <<>>            :: iodata() | undefined % = 1, optional
-        }).
--endif.
-
--ifndef('RPCREPLY.RPCERROR_PB_H').
--define('RPCREPLY.RPCERROR_PB_H', true).
--record('RpcReply.RpcError',
-        {code = 0               :: integer() | undefined, % = 1, optional, 32 bits
-         message = <<>>         :: unicode:chardata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('RPCREPLY_PB_H').
--define('RPCREPLY_PB_H', true).
--record('RpcReply',
-        {reply                  :: {result, message_pb:'RpcReply.RpcResult'()} | {error, message_pb:'RpcReply.RpcError'()} | undefined % oneof
-        }).
--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.

proto/message.proto

@@ -7,19 +7,27 @@ message RequestFrame {
   uint32 packet_id = 1;
   oneof body {
     AuthRequest auth_request = 2;
-    RpcRequest rpc_request = 3;
+    ContainerRequest container_request = 3;
-    ContainerRequest container_request = 4;
   }
 }
 
-message ResponseFrame {
+message ReplyFrame {
   uint32 packet_id = 1;
-  oneof body {
+  oneof reply {
-    AuthReply auth_reply = 2;
+    ReplyResult result = 2;
-    RpcReply rpc_reply = 3;
+    ReplyError error = 3;
   }
 }
 
+message ReplyResult {
+  bytes data = 1;
+}
+
+message ReplyError {
+  int32 code = 1;
+  string message = 2;
+}
+
 message CastFrame {
   oneof body {
     Pub pub = 1;
@@ -29,6 +37,7 @@ message CastFrame {
   }
 }
 
+// message定义
 message AuthRequest {
   bytes uuid = 1;
   bytes username = 2;
@@ -37,11 +46,6 @@ message AuthRequest {
   int32 timestamp = 5;
 }
 
-message AuthReply {
-  int32 code = 1;
-  bytes payload = 2;
-}
-
 message Pub {
   bytes topic = 1;
   int32 qos = 2;
@@ -53,27 +57,6 @@ message Command {
   bytes command = 2;
 }
 
-message RpcRequest {
-  bytes method = 1;
-  bytes params = 2;
-}
-
-message RpcReply {
-  message RpcResult {
-    bytes data = 1;
-  }
-
-  message RpcError {
-    int32 code = 1;
-    string message = 2;
-  }
-
-  oneof reply {
-    RpcResult result = 1;
-    RpcError error = 2;
-  }
-}
-
 message Data {
   bytes route_key = 1;
   bytes metric = 2;

rebar.config

@@ -10,6 +10,7 @@
 {gpb_opts, [
     {i, "proto"},                 % proto 文件路径
     {src_dirs, ["proto"]},        % 源码目录（必须）
+    {f, ["message.proto"]},
     recursive,                     % 递归查找 proto 文件
     {module_name_suffix, "_pb"},   % 生成模块后缀
     {o_erl, "src/protobuf"},                % .erl 输出目录

src/host/iot_host.erl

@@ -130,9 +130,9 @@ 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, #'RpcReply'{reply = {result, #'RpcReply.RpcResult'{data = ResultBin}}}} ->
+        {rpc_reply, Ref, #'ReplyFrame'{reply = {result, #'ReplyResult'{data = ResultBin}}}} ->
             {ok, iolist_to_binary(ResultBin)};
-        {rpc_reply, Ref, #'RpcReply'{reply = {error, #'RpcReply.RpcError'{code = Code, message = Message}}}} ->
+        {rpc_reply, Ref, #'ReplyFrame'{reply = {error, #'ReplyError'{code = Code, message = Message}}}} ->
             {error, Code, unicode:characters_to_binary(Message)}
     after Timeout ->
         ok = gen_statem:call(Pid, {cancel_rpc_call, Ref}),

src/transport/tcp/ssl_channel.erl

@@ -56,8 +56,7 @@ command(Pid, CommandType, Command) when is_pid(Pid), is_integer(CommandType), is
 -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(),
-    RpcRequest = #'RpcRequest'{method = Method, params = Params},
+    gen_server:cast(Pid, {unsupported_rpc_call, ReceiverPid, Ref, Method, Params}),
-    gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {rpc_request, RpcRequest}}),
     Ref.
 
 -spec container_call(Pid :: pid(), ReceiverPid :: pid(), Request :: message_pb:'ContainerRequest'()) -> Ref :: reference().
@@ -87,7 +86,7 @@ start_link(Ref, Transport, Opts) ->
 init(Ref, Transport, _Opts = []) ->
     ok = iot_log:set_metadata(),
     {ok, Socket} = ranch:handshake(Ref),
-    logger:debug("[sdlan_channel] get a new connection: ~p", [Socket]),
+    logger:debug("[ssl_channel] get a new connection: ~p", [Socket]),
     Transport:setopts(Socket, [binary, {active, true}, {packet, 4}]),
     % erlang:start_timer(?PING_TICKER, self(), ping_ticker),
     gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}).
@@ -119,6 +118,19 @@ handle_cast({command, CommandType, Command}, State = #state{transport = Transpor
     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'{
@@ -164,15 +176,15 @@ handle_info({ssl, Socket, <<?FRAME_CAST, FrameBin/binary>>},
 
 handle_info({ssl, Socket, <<?FRAME_RESPONSE, FrameBin/binary>>},
             State = #state{socket = Socket, inflight = Inflight}) ->
-    ResponseFrame = message_pb:decode_msg(FrameBin, 'ResponseFrame'),
+    ReplyFrame = message_pb:decode_msg(FrameBin, 'ReplyFrame'),
-    handle_response_frame(ResponseFrame, Inflight, State);
+    handle_reply_frame(ReplyFrame, Inflight, State);
 
 handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) ->
-    logger:notice("[sdlan_channel] ssl socket closed"),
+    logger:notice("[ssl_channel] ssl socket closed"),
     {stop, normal, State};
 
 handle_info({ssl_error, Socket, Reason}, State = #state{socket = Socket}) ->
-    logger:notice("[sdlan_channel] ssl socket error: ~p", [Reason]),
+    logger:notice("[ssl_channel] ssl socket error: ~p", [Reason]),
     {stop, normal, State};
 
 handle_info({ssl_passive, Socket}, State = #state{transport = Transport, socket = Socket}) ->
@@ -180,11 +192,11 @@ handle_info({ssl_passive, Socket}, State = #state{transport = Transport, socket
     {noreply, State};
 
 handle_info(Info, State) ->
-    logger:warning("[sdlan_channel] get a unknown message: ~p, state: ~p", [Info, State]),
+    logger:warning("[ssl_channel] get a unknown message: ~p, state: ~p", [Info, State]),
     {noreply, State}.
 
 terminate(Reason, #state{}) ->
-    logger:warning("[sdlan_channel] stop with reason: ~p", [Reason]),
+    logger:warning("[ssl_channel] stop with reason: ~p", [Reason]),
     ok.
 
 code_change(_OldVsn, State, _Extra) ->
@@ -231,23 +243,21 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId,
                     case iot_host:attach_channel(HostPid, self()) of
                         ok ->
                             erlang:monitor(process, HostPid),
-                            send_response_frame(Transport, Socket, PacketId,
+                            send_reply_frame(Transport, Socket, PacketId, {result, #'ReplyResult'{data = <<"ok">>}}),
-                                {auth_reply, #'AuthReply'{code = 0, payload = <<"ok">>}}),
                             {noreply, State#state{uuid = UUID, host_pid = HostPid}};
                         {denied, Reason} when is_binary(Reason) ->
                             erlang:monitor(process, HostPid),
-                            send_response_frame(Transport, Socket, PacketId,
+                            send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyError'{code = 1, message = Reason}}),
-                                {auth_reply, #'AuthReply'{code = 1, payload = 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_response_frame(Transport, Socket, PacketId,
+                            send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyError'{code = 2, message = Reason}}),
-                                {auth_reply, #'AuthReply'{code = 2, payload = 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">>}}),
             logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]),
             {stop, State}
     end;
@@ -280,9 +290,9 @@ handle_event_stream_frame(#'TaskEventStream'{task_id = TaskId, type = Type, stre
     logger:debug("[ssl_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]),
     iot_event_stream_observer:stream_data(TaskId, Type, Stream).
 
--spec handle_response_frame(message_pb:'ResponseFrame'(), map(), #state{}) ->
+-spec handle_reply_frame(message_pb:'ReplyFrame'(), map(), #state{}) ->
     {noreply, #state{}}.
-handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {rpc_reply, RpcReply}}, Inflight, State)
+handle_reply_frame(#'ReplyFrame'{packet_id = PacketId, reply = Reply}, Inflight, State)
     when PacketId > 0 ->
     case maps:take(PacketId, Inflight) of
         error ->
@@ -291,20 +301,20 @@ handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {rpc_reply,
             erlang:cancel_timer(TimerRef),
             case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
                 true ->
-                    ReceiverPid ! {rpc_reply, Ref, RpcReply};
+                    ReceiverPid ! {rpc_reply, Ref, #'ReplyFrame'{packet_id = PacketId, reply = Reply}};
                 false ->
-                    logger:warning("[ws_channel] get async_call_reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [RpcReply, PacketId])
+                    logger:warning("[ws_channel] get reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [Reply, PacketId])
             end,
             {noreply, State#state{inflight = NInflight}}
     end;
-handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {auth_reply, AuthReply}}, _Inflight, State) ->
+handle_reply_frame(#'ReplyFrame'{packet_id = PacketId, reply = undefined}, _Inflight, State) ->
-    logger:warning("[ws_channel] unexpected auth_reply response, packet_id: ~p, body: ~p", [PacketId, AuthReply]),
+    logger:warning("[ws_channel] empty reply frame, packet_id: ~p", [PacketId]),
     {noreply, State};
-handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = undefined}, _Inflight, State) ->
+handle_reply_frame(#'ReplyFrame'{packet_id = PacketId, reply = Reply}, _Inflight, State) ->
-    logger:warning("[ws_channel] empty response frame, packet_id: ~p", [PacketId]),
+    logger:warning("[ws_channel] unexpected reply frame, packet_id: ~p, reply: ~p", [PacketId, Reply]),
     {noreply, State}.
 
--spec send_response_frame(module(), any(), non_neg_integer(), tuple()) -> any().
+-spec send_reply_frame(module(), any(), non_neg_integer(), tuple()) -> any().
-send_response_frame(Transport, Socket, PacketId, Body) ->
+send_reply_frame(Transport, Socket, PacketId, Reply) ->
-    Encoded = message_pb:encode_msg(#'ResponseFrame'{packet_id = PacketId, body = Body}),
+    Encoded = message_pb:encode_msg(#'ReplyFrame'{packet_id = PacketId, reply = Reply}),
     Transport:send(Socket, <<?FRAME_RESPONSE, Encoded/binary>>).