step 1

proto/message.proto

@@ -0,0 +1,223 @@
+syntax = "proto3";
+
+// 整体封装的消息体
+message RequestFrame {
+  uint32 packet_id = 1;
+  oneof body {
+    AuthRequest auth_request = 2;
+    RpcRequest rpc_request = 3;
+    ContainerRequest container_request = 4;
+  }
+}
+
+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 ContainerRef {
+  string id = 1;
+  string name = 2;
+}
+
+message ContainerRequest {
+  message List {
+    bool all = 1;
+  }
+
+  message Deploy {
+    uint32 task_id = 1;
+    ContainerDeployParams params = 2;
+  }
+
+  message Start {
+    ContainerRef target = 1;
+  }
+
+  message Stop {
+    ContainerRef target = 1;
+    uint32 timeout_seconds = 2;
+  }
+
+  message Kill {
+    ContainerRef target = 1;
+    string signal = 2;
+  }
+
+  message Remove {
+    ContainerRef target = 1;
+    bool force = 2;
+    bool remove_volumes = 3;
+  }
+
+  message Config {
+    ContainerRef target = 1;
+    bytes config = 2;
+  }
+
+  oneof action {
+    List list = 10;
+    Deploy deploy = 11;
+    Start start = 12;
+    Stop stop = 13;
+    Kill kill = 14;
+    Remove remove = 15;
+    Config config = 16;
+  }
+}
+
+message ContainerDeployParams {
+  string container_name = 1;
+  string container_dir = 2;
+  ContainerSpec spec = 3;
+}
+
+message ContainerSpec {
+  string image = 1;
+  repeated string command = 2;
+  repeated string entrypoint = 3;
+
+  repeated string env = 4;
+  map<string, string> labels = 5;
+
+  repeated VolumeBind volumes = 6;
+  string user = 7;
+  string working_dir = 8;
+  string hostname = 9;
+
+  repeated PortExpose expose = 10;
+  repeated string networks = 11;
+  string network_mode = 12;
+
+  Healthcheck healthcheck = 13;
+  RestartPolicy restart = 14;
+
+  bool privileged = 15;
+  repeated string cap_add = 16;
+  repeated string cap_drop = 17;
+  repeated DeviceMapping devices = 18;
+
+  ResourceLimits resources = 19;
+  repeated Ulimit ulimits = 20;
+  repeated TmpfsMount tmpfs = 21;
+  map<string, string> sysctls = 22;
+  repeated string extra_hosts = 23;
+}
+
+message VolumeBind {
+  string host_path = 1;
+  string container_path = 2;
+  bool read_only = 3;
+}
+
+message PortExpose {
+  uint32 container_port = 1;
+  string protocol = 2; // tcp / udp
+}
+
+message Healthcheck {
+  repeated string test = 1;
+  uint64 interval_ns = 2;
+  uint64 timeout_ns = 3;
+  uint32 retries = 4;
+}
+
+message RestartPolicy {
+  string name = 1; // no, always, unless-stopped, on-failure
+  uint32 maximum_retry_count = 2;
+}
+
+message DeviceMapping {
+  string host_path = 1;
+  string container_path = 2;
+  string cgroup_permissions = 3; // 默认可填 rwm
+}
+
+message ResourceLimits {
+  uint64 memory_bytes = 1;
+  uint64 memory_reservation_bytes = 2;
+  uint64 nano_cpus = 3;
+  uint64 cpu_shares = 4;
+}
+
+message Ulimit {
+  string name = 1;
+  uint64 soft = 2;
+  uint64 hard = 3;
+}
+
+message TmpfsMount {
+  string path = 1;
+  string options = 2;
+}
+
+
+message AuthRequest {
+  bytes uuid = 1;
+  bytes username = 2;
+  bytes salt = 3;
+  bytes token = 4;
+  int32 timestamp = 5;
+}
+
+message AuthReply {
+  int32 code = 1;
+  bytes payload = 2;
+}
+
+message Pub {
+  bytes topic = 1;
+  int32 qos = 2;
+  bytes content = 3;
+}
+
+message Command {
+  int32 command_type = 1;
+  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;
+}
+
+message TaskEventStream {
+  int32 task_id = 1;
+  bytes type = 2;
+  bytes stream = 3;
+}

src/efka_remote_agent.erl

@@ -9,6 +9,7 @@
 -module(efka_remote_agent).
 -author("anlicheng").
 -include("message.hrl").
+-include("message_pb.hrl").
 -include("efka_tables.hrl").
 
 -behaviour(gen_statem).
@@ -87,7 +88,7 @@ callback_mode() ->
 
 %% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia
 handle_event(cast, {metric_data, RouteKey, Metric}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    Packet = message_codec:encode(?MESSAGE_DATA, #data{
+    Packet = message_codec:encode(?MESSAGE_DATA, #'Data'{
         route_key = RouteKey,
         metric = Metric
     }),
@@ -95,7 +96,7 @@ handle_event(cast, {metric_data, RouteKey, Metric}, ?STATE_ACTIVATED, State = #s
     {keep_state, State};
 
 handle_event(cast, {metric_data, RouteKey, Metric}, _, State) ->
-    Packet = message_codec:encode(?MESSAGE_DATA, #data{
+    Packet = message_codec:encode(?MESSAGE_DATA, #'Data'{
         route_key = RouteKey,
         metric = Metric
     }),
@@ -104,7 +105,7 @@ handle_event(cast, {metric_data, RouteKey, Metric}, _, State) ->
 
 handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
     logger:debug("[efka_remote_agent] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
-    EventPacket = message_codec:encode(?MESSAGE_EVENT_STREAM, #task_event_stream{
+    EventPacket = message_codec:encode(?MESSAGE_EVENT_STREAM, #'TaskEventStream'{
         task_id = TaskId,
         type = Type,
         stream = Stream
@@ -113,7 +114,7 @@ handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED,
     {keep_state, State};
 
 handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    EventPacket = message_codec:encode(?MESSAGE_EVENT_STREAM, #task_event_stream{
+    EventPacket = message_codec:encode(?MESSAGE_EVENT_STREAM, #'TaskEventStream'{
         task_id = TaskId,
         type = <<"close">>,
         stream = Reason
@@ -159,8 +160,8 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DENIED, State) ->
 handle_event(info, {connect_reply, Reply}, ?STATE_CONNECTING, State = #state{transport_pid = TransportPid}) ->
     case Reply of
         ok ->
-            AuthBin = auth_request(),
+            AuthRequest = auth_request(),
-            efka_transport:auth_request(TransportPid, AuthBin),
+            efka_transport:auth_request(TransportPid, AuthRequest),
             {next_state, ?STATE_AUTH, State};
         {error, Reason} ->
             logger:debug("[efka_remote_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]),
@@ -170,7 +171,7 @@ handle_event(info, {connect_reply, Reply}, ?STATE_CONNECTING, State = #state{tra
 
 handle_event(info, {auth_reply, Reply}, ?STATE_AUTH, State = #state{transport_pid = TransportPid}) ->
     case Reply of
-        {ok, #auth_reply{code = Code, payload = Message}} ->
+        {ok, #'AuthReply'{code = Code, payload = Message}} ->
             case Code of
                 0 ->
                     logger:debug("[efka_remote_agent] auth success, message: ~p", [Message]),
@@ -214,7 +215,7 @@ handle_event(info, flush_cache, _, State) ->
 %% 激活消息
 
 %% 微服务部署
-handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"get_containers">>}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
+handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"get_containers">>}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
     %% 短暂的等待，efka_inetd收到消息后就立即返回了
     case docker_manager:get_containers() of
         {ok, Containers} ->
@@ -225,8 +226,8 @@ handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"get_conta
     {keep_state, State};
 
 %% 微服务部署
-handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"deploy">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
+handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"deploy">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    #{<<"task_id">> := TaskId, <<"config">> := Config} = Params,
+    #{<<"task_id">> := TaskId, <<"config">> := Config} = decode_rpc_payload(ParamsBin),
     %% 短暂的等待，efka_inetd收到消息后就立即返回了
     case docker_manager:deploy(TaskId, Config) of
         ok ->
@@ -237,8 +238,8 @@ handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"deploy">>
     {keep_state, State};
 
 %% 启动微服务
-handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"start_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
+handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"start_container">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    #{<<"container_name">> := ContainerName} = Params,
+    #{<<"container_name">> := ContainerName} = decode_rpc_payload(ParamsBin),
     %% 短暂的等待，efka_inetd收到消息后就立即返回了
     case docker_manager:start_container(ContainerName) of
         ok ->
@@ -249,8 +250,8 @@ handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"start_con
     {keep_state, State};
 
 %% 停止微服务
-handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"stop_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
+handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"stop_container">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    #{<<"container_name">> := ContainerName} = Params,
+    #{<<"container_name">> := ContainerName} = decode_rpc_payload(ParamsBin),
     %% 短暂的等待，efka_inetd收到消息后就立即返回了
     case docker_manager:stop_container(ContainerName) of
         ok ->
@@ -260,8 +261,8 @@ handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"stop_cont
     end,
     {keep_state, State};
 
-handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"kill_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
+handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"kill_container">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    #{<<"container_name">> := ContainerName} = Params,
+    #{<<"container_name">> := ContainerName} = decode_rpc_payload(ParamsBin),
     %% 短暂的等待，efka_inetd收到消息后就立即返回了
     case docker_manager:kill_container(ContainerName) of
         ok ->
@@ -271,8 +272,8 @@ handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"kill_cont
     end,
     {keep_state, State};
 
-handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"remove_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
+handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"remove_container">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    #{<<"container_name">> := ContainerName} = Params,
+    #{<<"container_name">> := ContainerName} = decode_rpc_payload(ParamsBin),
     %% 短暂的等待，efka_inetd收到消息后就立即返回了
     case docker_manager:remove_container(ContainerName) of
         ok ->
@@ -283,8 +284,8 @@ handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"remove_co
     {keep_state, State};
 
 %% config.json配置信息
-handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"config_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
+handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"config_container">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    #{<<"container_name">> := ContainerName, <<"config">> := Config} = Params,
+    #{<<"container_name">> := ContainerName, <<"config">> := Config} = decode_rpc_payload(ParamsBin),
     case docker_manager:config_container(ContainerName, Config) of
         ok ->
             efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
@@ -310,15 +311,15 @@ handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"config_co
 %    {keep_state, State};
 
 %% 处理命令
-handle_event(info, {server_cast, #command{command_type = ?COMMAND_AUTH, command = Auth0}}, StateName, State = #state{transport_pid = TransportPid}) ->
+handle_event(info, {server_cast, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}}, StateName, State = #state{transport_pid = TransportPid}) ->
     Auth = binary_to_integer(Auth0),
     case {Auth, StateName} of
         {1, ?STATE_ACTIVATED} ->
             {keep_state, State};
         {1, ?STATE_DENIED} ->
             %% 重新激活, 需要重新校验
-            AuthRequestBin = auth_request(),
+            AuthRequest = auth_request(),
-            efka_transport:auth_request(TransportPid, AuthRequestBin),
+            efka_transport:auth_request(TransportPid, AuthRequest),
             {next_state, ?STATE_AUTH, State};
         {0, _} ->
             %% 这个时候的主机应该是受限制的状态，不允许发送消息；但是能够接受服务器推送的消息
@@ -326,7 +327,7 @@ handle_event(info, {server_cast, #command{command_type = ?COMMAND_AUTH, command
     end;
 
 %% 处理Pub/Sub机制
-handle_event(info, {server_cast, #pub{topic = Topic, qos = Qos, content = Content}}, ?STATE_ACTIVATED, State) ->
+handle_event(info, {server_cast, #'Pub'{topic = Topic, qos = Qos, content = Content}}, ?STATE_ACTIVATED, State) ->
     logger:debug("[efka_remote_agent] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
     %% 消息发送到订阅系统
     efka_subscription:publish(Topic, Qos, Content),
@@ -361,7 +362,7 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
--spec auth_request() -> binary().
+-spec auth_request() -> message_pb:'AuthRequest'().
 auth_request() ->
     {ok, AuthInfo} = application:get_env(efka, auth),
     UUID = proplists:get_value(uuid, AuthInfo),
@@ -369,22 +370,32 @@ auth_request() ->
     Salt = proplists:get_value(salt, AuthInfo),
     Token = proplists:get_value(token, AuthInfo),
 
-    message_codec:encode(?MESSAGE_AUTH_REQUEST, #auth_request{
+    #'AuthRequest'{
         uuid = unicode:characters_to_binary(UUID),
         username = unicode:characters_to_binary(Username),
         salt = unicode:characters_to_binary(Salt),
         token = unicode:characters_to_binary(Token),
         timestamp = efka_util:timestamp()
-    }).
+    }.
 
--spec reply_success(Result :: any()) -> binary().
+-spec reply_success(Result :: any()) -> message_pb:'RpcReply'().
 reply_success(Result) ->
-    message_codec:encode(?MESSAGE_JSONRPC_REPLY, #jsonrpc_reply{result = Result}).
+    #'RpcReply'{
+        reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(Result)}}
+    }.
 
--spec reply_error(Code :: integer(), Message :: binary()) -> binary().
+-spec reply_error(Code :: integer(), Message :: binary()) -> message_pb:'RpcReply'().
 reply_error(Code, Message) when is_integer(Code), is_binary(Message) ->
-    Error = #{
+    #'RpcReply'{
-        <<"code">> => Code,
+        reply = {error, #'RpcReply.RpcError'{code = Code, message = Message}}
-        <<"message">> => Message
+    }.
-    },
+
-    message_codec:encode(?MESSAGE_JSONRPC_REPLY, #jsonrpc_reply{error = Error}).
+-spec decode_rpc_payload(binary()) -> any().
+decode_rpc_payload(<<>>) ->
+    #{};
+decode_rpc_payload(Bin) when is_binary(Bin) ->
+    jiffy:decode(Bin, [return_maps]).
+
+-spec encode_rpc_payload(any()) -> binary().
+encode_rpc_payload(Payload) ->
+    jiffy:encode(Payload, [force_utf8]).

src/efka_transport.erl

@@ -32,9 +32,9 @@
 %%% API
 %%%===================================================================
 
--spec auth_request(Pid :: pid(), AuthBin :: binary()) -> no_return().
+-spec auth_request(Pid :: pid(), AuthRequest :: any()) -> no_return().
-auth_request(Pid, AuthBin) when is_pid(Pid), is_binary(AuthBin) ->
+auth_request(Pid, AuthRequest) when is_pid(Pid) ->
-    gen_server:cast(Pid, {auth_request, AuthBin}).
+    gen_server:cast(Pid, {auth_request, AuthRequest}).
 
 -spec connect(Pid :: pid()) -> no_return().
 connect(Pid) when is_pid(Pid) ->
@@ -44,10 +44,10 @@ connect(Pid) when is_pid(Pid) ->
 send(Pid, Packet) when is_pid(Pid), is_binary(Packet) ->
     gen_server:cast(Pid, {send, Packet}).
 
--spec rpc_reply(Pid :: pid() | undefined, PacketId :: integer(), Response :: binary()) -> no_return().
+-spec rpc_reply(Pid :: pid() | undefined, PacketId :: integer(), Response :: any()) -> no_return().
-rpc_reply(undefined, PacketId, Response) when is_integer(PacketId), is_binary(Response) ->
+rpc_reply(undefined, PacketId, _Response) when is_integer(PacketId) ->
     ok;
-rpc_reply(Pid, PacketId, Reply) when is_pid(Pid), is_integer(PacketId), is_binary(Reply) ->
+rpc_reply(Pid, PacketId, Reply) when is_pid(Pid), is_integer(PacketId) ->
     gen_server:cast(Pid, {rpc_reply, PacketId, Reply}).
 
 %% 关闭的时候不一定能成功，可能关闭的时候；transport进程已经退出了
@@ -112,18 +112,23 @@ handle_cast(connect, State = #state{host = Host, port = Port, parent_pid = Paren
     end;
 
 %% auth校验
-handle_cast({auth_request, AuthRequestBin}, State = #state{parent_pid = ParentPid, socket = Socket}) ->
+handle_cast({auth_request, AuthRequest}, State = #state{parent_pid = ParentPid, socket = Socket}) ->
     PacketId = 1,
-    ok = ssl:send(Socket, <<?PACKET_REQUEST, PacketId:32, AuthRequestBin/binary>>),
+    Packet = message_codec:encode_request(PacketId, ?MESSAGE_AUTH_REQUEST, AuthRequest),
+    ok = ssl:send(Socket, Packet),
     %% 需要等待auth返回的结果
     receive
-        {ssl, Socket, <<?PACKET_RESPONSE, PacketId:32, ReplyBin/binary>>} ->
+        {ssl, Socket, ReplyBin} ->
-            {ok, #auth_reply{} = Reply} = message_codec:decode(ReplyBin),
+            case message_codec:decode_response(ReplyBin) of
-            ParentPid ! {auth_reply, {ok, Reply}},
+                {ok, PacketId, Reply} ->
-            {noreply, State};
+                    ParentPid ! {auth_reply, {ok, Reply}};
-        {ssl, Socket, Info} ->
+                {ok, ReplyPacketId, _Reply} ->
-            logger:warning("[efka_transport] get invalid auth_reply: ~p", [Info]),
+                    logger:warning("[efka_transport] get unexpected auth_reply packet_id: ~p", [ReplyPacketId]),
-            ParentPid ! {auth_reply, {error, invalid_auth_reply}},
+                    ParentPid ! {auth_reply, {error, invalid_auth_reply}};
+                error ->
+                    logger:warning("[efka_transport] get invalid auth_reply: ~p", [ReplyBin]),
+                    ParentPid ! {auth_reply, {error, invalid_auth_reply}}
+            end,
             {noreply, State}
     after 5000 ->
         ParentPid ! {auth_reply, {error, timeout}},
@@ -131,12 +136,13 @@ handle_cast({auth_request, AuthRequestBin}, State = #state{parent_pid = ParentPi
     end;
 
 handle_cast({send, Packet}, State = #state{socket = Socket}) ->
-    ok = ssl:send(Socket, <<?PACKET_CAST, Packet/binary>>),
+    ok = ssl:send(Socket, Packet),
     {noreply, State};
 
 %% 服务push的消息的回复
 handle_cast({rpc_reply, PacketId, Reply}, State = #state{socket = Socket}) ->
-    ok = ssl:send(Socket, <<?PACKET_RESPONSE, PacketId:32, Reply/binary>>),
+    Packet = message_codec:encode_response(PacketId, ?MESSAGE_JSONRPC_REPLY, Reply),
+    ok = ssl:send(Socket, Packet),
     {noreply, State}.
 
 %% @private
@@ -145,15 +151,19 @@ handle_cast({rpc_reply, PacketId, Reply}, State = #state{socket = Socket}) ->
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
-%% 服务器主动推送的数据，有packetId的是要求返回的；为0的表示不需要返回值
+%% 服务器主动推送的数据
-handle_info({ssl, Socket, <<?PACKET_CAST, CastBin/binary>>}, State = #state{socket = Socket, parent_pid = ParentPid}) ->
+handle_info({ssl, Socket, PacketBin}, State = #state{socket = Socket, parent_pid = ParentPid}) ->
-    {ok, CastRequest} = message_codec:decode(CastBin),
+    case message_codec:decode_request(PacketBin) of
-    ParentPid ! {server_cast, CastRequest},
+        {ok, PacketId, Request} ->
-    {noreply, State};
+            ParentPid ! {server_rpc, PacketId, Request};
-
+        error ->
-handle_info({ssl, Socket, <<?PACKET_REQUEST, PacketId:32, RPCRequestBin/binary>>}, State = #state{socket = Socket, parent_pid = ParentPid}) ->
+            case message_codec:decode_cast(PacketBin) of
-    {ok, RPCRequest} = message_codec:decode(RPCRequestBin),
+                {ok, CastRequest} ->
-    ParentPid ! {server_rpc, PacketId, RPCRequest},
+                    ParentPid ! {server_cast, CastRequest};
+                error ->
+                    logger:warning("[efka_transport] get invalid packet: ~p", [PacketBin])
+            end
+    end,
     {noreply, State};
 
 handle_info({ssl_error, Socket, Reason}, State = #state{socket = Socket}) ->

src/message/message_codec.erl

@@ -9,133 +9,122 @@
 -module(message_codec).
 -author("anlicheng").
 -include("message.hrl").
-
+-include("message_pb.hrl").
--define(I8, 1).
--define(I16, 2).
--define(I32, 3).
--define(Bytes, 4).
 
 %% API
--export([encode/2, decode/1]).
+-export([encode/2, encode_request/3, encode_response/3]).
+-export([decode_request/1, decode_response/1, decode_cast/1]).
 
 -spec encode(MessageType :: integer(), Message :: any()) -> binary().
 encode(MessageType, Message) when is_integer(MessageType) ->
-    Bin = encode0(Message),
+    case frame_for_type(MessageType, Message, 0) of
-    <<MessageType, Bin/binary>>.
+        {request, Frame} ->
-encode0(#auth_request{uuid = UUID, username = Username, salt = Salt, token = Token, timestamp = Timestamp}) ->
+            message_pb:encode_msg(Frame);
-    iolist_to_binary([
+        {response, Frame} ->
-        marshal(?Bytes, UUID),
+            message_pb:encode_msg(Frame);
-        marshal(?Bytes, Username),
+        {cast, Frame} ->
-        marshal(?Bytes, Salt),
+            message_pb:encode_msg(Frame)
-        marshal(?Bytes, Token),
+    end.
-        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}) ->
+-spec encode_request(PacketId :: integer(), MessageType :: integer(), Message :: any()) -> binary().
-    ReqBody = erlang:term_to_binary(#{<<"method">> => Method, <<"params">> => Params}),
+encode_request(PacketId, MessageType, Message) when is_integer(PacketId), is_integer(MessageType) ->
-    iolist_to_binary([
+    case frame_for_type(MessageType, Message, PacketId) of
-        marshal(?Bytes, ReqBody)
+        {request, Frame} ->
-    ]);
+            message_pb:encode_msg(Frame);
-encode0(#data{route_key = RouteKey, metric = Metric}) ->
+        _ ->
-    iolist_to_binary([
+            erlang:error({unsupported_request_type, MessageType})
-        marshal(?Bytes, RouteKey),
+    end.
-        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.
+-spec encode_response(PacketId :: integer(), MessageType :: integer(), Message :: any()) -> binary().
-decode(<<PacketType:8, Packet/binary>>) ->
+encode_response(PacketId, MessageType, Message) when is_integer(PacketId), is_integer(MessageType) ->
-    case unmarshal(Packet) of
+    case frame_for_type(MessageType, Message, PacketId) of
-        {ok, Fields} ->
+        {response, Frame} ->
-            decode0(PacketType, Fields);
+            message_pb:encode_msg(Frame);
-        error ->
+        _ ->
+            erlang:error({unsupported_response_type, MessageType})
+    end.
+
+-spec decode_request(Bin :: binary()) -> {ok, PacketId :: integer(), Message :: any()} | error.
+decode_request(Bin) when is_binary(Bin) ->
+    try
+        Frame = message_pb:decode_msg(Bin, 'RequestFrame'),
+        decode_request_frame(Frame)
+    catch
+        _:_ ->
             error
     end.
-decode0(?MESSAGE_AUTH_REQUEST, [UUID, Username, Salt, Token, Timestamp]) ->
+
-    {ok, #auth_request{uuid = UUID, username = Username, salt = Salt, token = Token, timestamp = Timestamp}};
+-spec decode_response(Bin :: binary()) -> {ok, PacketId :: integer(), Message :: any()} | error.
-decode0(?MESSAGE_JSONRPC_REPLY, [ReplyBin]) ->
+decode_response(Bin) when is_binary(Bin) ->
-    case erlang:binary_to_term(ReplyBin) of
+    try
-        #{<<"result">> := Result} ->
+        Frame = message_pb:decode_msg(Bin, 'ResponseFrame'),
-            {ok, #jsonrpc_reply{result = Result}};
+        decode_response_frame(Frame)
-        #{<<"error">> := Error} ->
+    catch
-            {ok, #jsonrpc_reply{error = Error}};
+        _:_ ->
-        _ ->
             error
-    end;
+    end.
-decode0(?MESSAGE_PUB, [Topic, Qos, Content]) ->
+
-    {ok, #pub{topic = Topic, qos = Qos, content = Content}};
+-spec decode_cast(Bin :: binary()) -> {ok, Message :: any()} | error.
-decode0(?MESSAGE_COMMAND, [CommandType, Command]) ->
+decode_cast(Bin) when is_binary(Bin) ->
-    {ok, #command{command_type = CommandType, command = Command}};
+    try
-decode0(?MESSAGE_AUTH_REPLY, [Code, Payload]) ->
+        Frame = message_pb:decode_msg(Bin, 'CastFrame'),
-    {ok, #auth_reply{code = Code, payload = Payload}};
+        decode_cast_frame(Frame)
-decode0(?MESSAGE_JSONRPC_REQUEST, [ReqBody]) ->
+    catch
-    #{<<"method">> := Method, <<"params">> := Params} = erlang:binary_to_term(ReqBody),
+        _:_ ->
-    {ok, #jsonrpc_request{method = Method, params = Params}};
+            error
-decode0(?MESSAGE_DATA, [RouteKey, Metric]) ->
+    end.
-    {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 :: integer(), Field :: any()) -> binary().
+frame_for_type(?MESSAGE_AUTH_REQUEST, Message = #'AuthRequest'{}, PacketId) ->
-marshal(?I8, Field) when is_integer(Field) ->
+    {request, #'RequestFrame'{packet_id = PacketId, body = {auth_request, Message}}};
-    <<?I8, Field:8>>;
+frame_for_type(?MESSAGE_JSONRPC_REQUEST, Message = #'RpcRequest'{}, PacketId) ->
-marshal(?I16, Field) when is_integer(Field) ->
+    {request, #'RequestFrame'{packet_id = PacketId, body = {rpc_request, Message}}};
-    <<?I16, Field:16>>;
+frame_for_type(?MESSAGE_AUTH_REPLY, Message = #'AuthReply'{}, PacketId) ->
-marshal(?I32, Field) when is_integer(Field) ->
+    {response, #'ResponseFrame'{packet_id = PacketId, body = {auth_reply, Message}}};
-    <<?I32, Field:32>>;
+frame_for_type(?MESSAGE_JSONRPC_REPLY, Message = #'RpcReply'{}, PacketId) ->
-marshal(?Bytes, Field) when is_binary(Field) ->
+    {response, #'ResponseFrame'{packet_id = PacketId, body = {rpc_reply, Message}}};
-    Len = byte_size(Field),
+frame_for_type(?MESSAGE_PUB, Message = #'Pub'{}, _PacketId) ->
-    <<?Bytes, Len:16, Field/binary>>.
+    {cast, #'CastFrame'{body = {pub, Message}}};
+frame_for_type(?MESSAGE_COMMAND, Message = #'Command'{}, _PacketId) ->
+    {cast, #'CastFrame'{body = {command, Message}}};
+frame_for_type(?MESSAGE_DATA, Message = #'Data'{}, _PacketId) ->
+    {cast, #'CastFrame'{body = {data, Message}}};
+frame_for_type(?MESSAGE_EVENT_STREAM, Message = #'TaskEventStream'{}, _PacketId) ->
+    {cast, #'CastFrame'{body = {event_stream, Message}}};
+frame_for_type(MessageType, Message, _PacketId) ->
+    erlang:error({unsupported_message_type, MessageType, Message}).
 
--spec unmarshal(Bin :: binary()) -> {ok, Components :: [any()]} | error.
+decode_request_frame(#'RequestFrame'{packet_id = PacketId, body = {auth_request, Message}})
-unmarshal(Bin) when is_binary(Bin) ->
+  when is_integer(PacketId), PacketId > 0 ->
-    unmarshal(Bin, []).
+    {ok, PacketId, Message};
-unmarshal(<<>>, Acc) ->
+decode_request_frame(#'RequestFrame'{packet_id = PacketId, body = {rpc_request, Message}})
-    {ok, lists:reverse(Acc)};
+  when is_integer(PacketId), PacketId > 0 ->
-unmarshal(<<?I8, F:8, Rest/binary>>, Acc) ->
+    {ok, PacketId, Message};
-    unmarshal(Rest, [F|Acc]);
+decode_request_frame(#'RequestFrame'{packet_id = PacketId, body = {container_request, Message}})
-unmarshal(<<?I16, F:16, Rest/binary>>, Acc) ->
+  when is_integer(PacketId), PacketId > 0 ->
-    unmarshal(Rest, [F|Acc]);
+    {ok, PacketId, Message};
-unmarshal(<<?I32, F:32, Rest/binary>>, Acc) ->
+decode_request_frame(_) ->
-    unmarshal(Rest, [F|Acc]);
+    error.
-unmarshal(<<?Bytes, Len:16, F:Len/binary, Rest/binary>>, Acc) ->
+
-    unmarshal(Rest, [F|Acc]);
+decode_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {auth_reply, Message}})
-unmarshal(_, _) ->
+  when is_integer(PacketId), PacketId > 0 ->
+    {ok, PacketId, Message};
+decode_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {rpc_reply, Message}})
+  when is_integer(PacketId), PacketId > 0 ->
+    {ok, PacketId, Message};
+decode_response_frame(_) ->
+    error.
+
+decode_cast_frame(#'CastFrame'{body = {pub, Message}}) ->
+    {ok, Message};
+decode_cast_frame(#'CastFrame'{body = {command, Message}}) ->
+    {ok, Message};
+decode_cast_frame(#'CastFrame'{body = {data, Message}}) ->
+    {ok, Message};
+decode_cast_frame(#'CastFrame'{body = {event_stream, Message}}) ->
+    {ok, Message};
+decode_cast_frame(_) ->
     error.