fix proto

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.
 
@@ -221,14 +236,6 @@
         }).
 -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',
@@ -246,36 +253,6 @@
         }).
 -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',

proto/message.proto

@@ -5,19 +5,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;
@@ -165,7 +173,6 @@ message TmpfsMount {
   string options = 2;
 }
 
-
 message AuthRequest {
   bytes uuid = 1;
   bytes username = 2;
@@ -174,11 +181,6 @@ message AuthRequest {
   int32 timestamp = 5;
 }
 
-message AuthReply {
-  int32 code = 1;
-  bytes payload = 2;
-}
-
 message Pub {
   bytes topic = 1;
   int32 qos = 2;
@@ -190,27 +192,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;
@@ -220,4 +201,4 @@ message TaskEventStream {
   int32 task_id = 1;
   bytes type = 2;
   bytes stream = 3;
-}
+}

src/efka_client.erl

@@ -140,8 +140,8 @@ handle_event(info, flush_cache, _, State) ->
 
 %% 处理收到的ssl消息
 handle_event(info, {ssl, Socket, <<?FRAME_RESPONSE, PacketBin/binary>>}, _, State = #state{socket = Socket}) ->
-    ResponseFrame = message_pb:decode_msg(PacketBin, 'ResponseFrame'),
+    ReplyFrame = message_pb:decode_msg(PacketBin, 'ReplyFrame'),
-    {keep_state, State, [{next_event, internal, {decoded_response, ResponseFrame}}]};
+    {keep_state, State, [{next_event, internal, {decoded_reply, ReplyFrame}}]};
 handle_event(info, {ssl, Socket, <<?FRAME_REQUEST, PacketBin/binary>>}, _, State = #state{socket = Socket}) ->
     RequestFrame = message_pb:decode_msg(PacketBin, 'RequestFrame'),
     {keep_state, State, [{next_event, internal, {decoded_request, RequestFrame}}]};
@@ -164,44 +164,33 @@ handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, b
   ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     case handle_container_request(Request) of
         ok ->
-            send_rpc_result_reply(Socket, PacketId, <<"ok">>);
+            send_result_reply(Socket, PacketId, <<"ok">>);
         {ok, Reply} ->
-            send_rpc_result_reply(Socket, PacketId, Reply);
+            send_result_reply(Socket, PacketId, Reply);
         {error, Reason} ->
-            send_rpc_error_reply(Socket, PacketId, Reason)
+            send_error_reply(Socket, PacketId, Reason)
     end,
     {keep_state, State};
 handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = _Body}},
   ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
-    send_rpc_error_reply(Socket, PacketId, <<"agent restricted">>),
+    send_error_reply(Socket, PacketId, <<"agent restricted">>),
     {keep_state, State};
 handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = _Body}},
   _StateName, State = #state{socket = Socket}) ->
-    send_rpc_error_reply(Socket, PacketId, <<"agent state invalid">>),
+    send_error_reply(Socket, PacketId, <<"agent state invalid">>),
     {keep_state, State};
 
-handle_event(internal, {decoded_response, #'ResponseFrame'{
+handle_event(internal, {decoded_reply, #'ReplyFrame'{packet_id = 1, reply = {result, #'ReplyResult'{data = Message}}}}, ?STATE_AUTH, State) ->
-    packet_id = 1,
+    logger:debug("[efka_client] auth success, message: ~p", [Message]),
-    body = {auth_reply, #'AuthReply'{code = Code, payload = Message}}
+    {next_state, ?STATE_ACTIVATED, State, [{next_event, info, flush_cache}]};
-}}, ?STATE_AUTH, State = #state{socket = Socket}) ->
+handle_event(internal, {decoded_reply, #'ReplyFrame'{packet_id = 1, reply = {error, #'ReplyError'{code = 1, message = Message}}}}, ?STATE_AUTH, State) ->
-    case Code of
+    logger:debug("[efka_client] auth denied, message: ~p", [Message]),
-        0 ->
+    {next_state, ?STATE_RESTRICTED, State};
-            logger:debug("[efka_client] auth success, message: ~p", [Message]),
+handle_event(internal, {decoded_reply, #'ReplyFrame'{packet_id = 1, reply = {error, #'ReplyError'{message = Message}}}}, ?STATE_AUTH, State = #state{socket = Socket}) ->
-            {next_state, ?STATE_ACTIVATED, State, [{next_event, info, flush_cache}]};
+    logger:debug("[efka_client] auth failed, message: ~p", [Message]),
-        1 ->
+    disconnect(Socket),
-            logger:debug("[efka_client] auth denied, message: ~p", [Message]),
+    schedule_reconnect(),
-            {next_state, ?STATE_RESTRICTED, State};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}};
-        2 ->
-            logger:debug("[efka_client] auth failed, message: ~p", [Message]),
-            disconnect(Socket),
-            schedule_reconnect(),
-            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}};
-        _ ->
-            logger:debug("[efka_client] auth failed, invalid message"),
-            disconnect(Socket),
-            schedule_reconnect(),
-            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}}
-    end;
 
 %% 处理命令
 handle_event(internal, {decoded_cast, #'CastFrame'{
@@ -288,10 +277,6 @@ disconnect(Socket) ->
 schedule_reconnect() ->
     erlang:start_timer(5000, self(), create_transport).
 
--spec encode_rpc_payload(any()) -> binary().
-encode_rpc_payload(Payload) ->
-    jiffy:encode(Payload, [force_utf8]).
-
 -spec handle_container_request(message_pb:'ContainerRequest'()) -> ok.
 handle_container_request(#'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = _All}}}) ->
     case docker_manager:get_containers() of
@@ -318,23 +303,19 @@ handle_container_request(#'ContainerRequest'{action = {config, #'ContainerReques
     ContainerTarget = container_target(Target),
     docker_manager:config_container(ContainerTarget, iolist_to_binary(Config)).
 
--spec send_rpc_result_reply(ssl:sslsocket(), integer(), any()) -> ok.
+-spec send_result_reply(ssl:sslsocket(), integer(), binary()) -> ok.
-send_rpc_result_reply(Socket, PacketId, Payload) ->
+send_result_reply(Socket, PacketId, Payload) when is_binary(Payload) ->
-    Packet = message_pb:encode_msg(#'ResponseFrame'{
+    Packet = message_pb:encode_msg(#'ReplyFrame'{
         packet_id = PacketId,
-        body = {rpc_reply, #'RpcReply'{
+        reply = {result, #'ReplyResult'{data = Payload}}
-            reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(Payload)}}
-        }}
     }),
     send_packet(Socket, Packet).
 
--spec send_rpc_error_reply(ssl:sslsocket(), integer(), binary()) -> ok.
+-spec send_error_reply(ssl:sslsocket(), integer(), binary()) -> ok.
-send_rpc_error_reply(Socket, PacketId, Reason) ->
+send_error_reply(Socket, PacketId, Reason) when is_binary(Reason) ->
-    Packet = message_pb:encode_msg(#'ResponseFrame'{
+    Packet = message_pb:encode_msg(#'ReplyFrame'{
         packet_id = PacketId,
-        body = {rpc_reply, #'RpcReply'{
+        reply = {error, #'ReplyError'{code = -1, message = Reason}}
-            reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
-        }}
     }),
     send_packet(Socket, Packet).