fix rpc

include/message_pb.hrl

@@ -8,6 +8,29 @@
 -define(message_pb_gpb_version, "4.21.7").
 
 
+-ifndef('REQUESTFRAME_PB_H').
+-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'()} | undefined % oneof
+        }).
+-endif.
+
+-ifndef('RESPONSEFRAME_PB_H').
+-define('RESPONSEFRAME_PB_H', true).
+-record('ResponseFrame',
+        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
+         body                   :: {auth_reply, message_pb:'AuthReply'()} | {rpc_reply, message_pb:'RpcReply'()} | undefined % oneof
+        }).
+-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',
@@ -44,42 +67,18 @@
         }).
 -endif.
 
--ifndef('JSONRPCREQUEST_PB_H').
+-ifndef('RPCREQUEST_PB_H').
--define('JSONRPCREQUEST_PB_H', true).
+-define('RPCREQUEST_PB_H', true).
--record('JsonRpcRequest',
+-record('RpcRequest',
         {method = <<>>          :: iodata() | undefined, % = 1, optional
          params = <<>>          :: iodata() | undefined % = 2, optional
         }).
 -endif.
 
--ifndef('JSONRPCREPLY_PB_H').
+-ifndef('RPCREPLY_PB_H').
--define('JSONRPCREPLY_PB_H', true).
+-define('RPCREPLY_PB_H', true).
--record('JsonRpcReply',
+-record('RpcReply',
-        {result = <<>>          :: iodata() | undefined, % = 1, optional
+        {body                   :: {result, iodata()} | {error, iodata()} | undefined % oneof
-         error = <<>>           :: iodata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('REQUESTFRAME_PB_H').
--define('REQUESTFRAME_PB_H', true).
--record('RequestFrame',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         body                   :: {auth_request, message_pb:'AuthRequest'()} | {jsonrpc_request, message_pb:'JsonRpcRequest'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('RESPONSEFRAME_PB_H').
--define('RESPONSEFRAME_PB_H', true).
--record('ResponseFrame',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         body                   :: {auth_reply, message_pb:'AuthReply'()} | {jsonrpc_reply, message_pb:'JsonRpcReply'()} | undefined % oneof
-        }).
--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.
 

proto/message.proto

@@ -1,5 +1,33 @@
 syntax = "proto3";
 
+// 整体封装的消息体
+message RequestFrame {
+  uint32 packet_id = 1;
+  oneof body {
+    AuthRequest auth_request = 2;
+    RpcRequest rpc_request = 3;
+  }
+}
+
+message ResponseFrame {
+  uint32 packet_id = 1;
+  oneof body {
+    AuthReply auth_reply = 2;
+    RpcReply rpc_reply = 3;
+  }
+}
+
+message CastFrame {
+  oneof body {
+    Pub pub = 1;
+    Command command = 2;
+    Data data = 3;
+    TaskEventStream event_stream = 4;
+  }
+}
+
+// message定义
+
 message AuthRequest {
   bytes uuid = 1;
   bytes username = 2;
@@ -24,38 +52,15 @@ message Command {
   bytes command = 2;
 }
 
-message JsonRpcRequest {
+message RpcRequest {
   bytes method = 1;
   bytes params = 2;
 }
 
-message JsonRpcReply {
+message RpcReply {
-  bytes result = 1;
-  bytes error = 2;
-}
-
-message RequestFrame {
-  uint32 packet_id = 1;
   oneof body {
-    AuthRequest auth_request = 2;
+    bytes result = 1;
-    JsonRpcRequest jsonrpc_request = 3;
+    bytes error = 2;
-  }
-}
-
-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;
   }
 }
 

src/host/iot_host.erl

@@ -132,9 +132,9 @@ remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName)
 -spec await_reply(Pid :: pid(), Ref :: reference(), Timeout :: integer()) -> {ok, Result :: binary()} | {error, Reason :: binary()}.
 await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_reference(Ref), is_integer(Timeout) ->
     receive
-        {jsonrpc_reply, Ref, #'JsonRpcReply'{result = ResultBin, error = <<>>}} ->
+        {rpc_reply, Ref, #'RpcReply'{body = {result, ResultBin}}} ->
             {ok, iolist_to_binary(ResultBin)};
-        {jsonrpc_reply, Ref, #'JsonRpcReply'{result = <<>>, error = ErrorBin}} ->
+        {rpc_reply, Ref, #'RpcReply'{body = {error, ErrorBin}}} ->
             {error, iolist_to_binary(ErrorBin)}
     after Timeout ->
         ok = gen_statem:call(Pid, {cancel_jsonrpc_call, Ref}),
@@ -399,7 +399,7 @@ state_map(#state{host_id = HostId, uuid = UUID, has_session = HasSession, heartb
 
 flush_reply(Ref) ->
     receive
-        {jsonrpc_reply, Ref, _Reply} ->
+        {rpc_reply, Ref, _Reply} ->
             ok
     after 0 ->
         ok

src/transport/tcp/tcp_channel.erl

@@ -118,7 +118,7 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{t
         {ok, NPacketId, NextPacketId} ->
             Encoded = message_pb:encode_msg(#'RequestFrame'{
                 packet_id = NPacketId,
-                body = {jsonrpc_request, #'JsonRpcRequest'{method = Method, params = Params}}
+                body = {rpc_request, #'RpcRequest'{method = Method, params = Params}}
             }),
             TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}),
             Transport:send(Socket, <<?FRAME_REQUEST, Encoded/binary>>),
@@ -267,8 +267,8 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId,
             logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]),
             {stop, State}
     end;
-handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {jsonrpc_request, RpcRequest}}, _Transport, _Socket, State) ->
+handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {rpc_request, RpcRequest}}, _Transport, _Socket, State) ->
-    logger:warning("[ws_channel] unsupported request message type: jsonrpc_request, packet_id: ~p, request: ~p", [PacketId, RpcRequest]),
+    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 = undefined}, _Transport, _Socket, State) ->
     logger:warning("[ws_channel] empty request frame, packet_id: ~p", [PacketId]),
@@ -295,19 +295,19 @@ handle_event_stream_frame(#'TaskEventStream'{task_id = TaskId, type = Type, stre
 
 -spec handle_response_frame(message_pb:'ResponseFrame'(), map(), #state{}) ->
     {noreply, #state{}}.
-handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {jsonrpc_reply, RpcReply}}, Inflight, State)
+handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {rpc_reply, RpcReply}}, Inflight, State)
     when PacketId > 0 ->
     case maps:take(PacketId, Inflight) of
         error ->
             {noreply, State};
-        {#inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, NInflight} ->
+                {#inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, NInflight} ->
-            erlang:cancel_timer(TimerRef),
+                    erlang:cancel_timer(TimerRef),
-            case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
+                    case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
-                true ->
+                        true ->
-                    ReceiverPid ! {jsonrpc_reply, Ref, RpcReply};
+                            ReceiverPid ! {rpc_reply, Ref, RpcReply};
-                false ->
+                        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 async_call_reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [RpcReply, PacketId])
-            end,
+                    end,
             {noreply, State#state{inflight = NInflight}}
     end;
 handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {auth_reply, AuthReply}}, _Inflight, State) ->