diff --git a/proto/message.proto b/proto/message.proto new file mode 100644 index 0000000..9513346 --- /dev/null +++ b/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 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 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; +} diff --git a/src/efka_remote_agent.erl b/src/efka_remote_agent.erl index 4a35e8e..a8f5406 100644 --- a/src/efka_remote_agent.erl +++ b/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(), - efka_transport:auth_request(TransportPid, AuthBin), + AuthRequest = auth_request(), + 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}) -> - #{<<"task_id">> := TaskId, <<"config">> := Config} = Params, +handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"deploy">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) -> + #{<<"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}) -> - #{<<"container_name">> := ContainerName} = Params, +handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"start_container">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) -> + #{<<"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}) -> - #{<<"container_name">> := ContainerName} = Params, +handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"stop_container">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) -> + #{<<"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}) -> - #{<<"container_name">> := ContainerName} = Params, +handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"kill_container">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) -> + #{<<"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}) -> - #{<<"container_name">> := ContainerName} = Params, +handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"remove_container">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) -> + #{<<"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}) -> - #{<<"container_name">> := ContainerName, <<"config">> := Config} = Params, +handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"config_container">>, params = ParamsBin}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) -> + #{<<"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(), - efka_transport:auth_request(TransportPid, AuthRequestBin), + AuthRequest = auth_request(), + 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 = #{ - <<"code">> => Code, - <<"message">> => Message - }, - message_codec:encode(?MESSAGE_JSONRPC_REPLY, #jsonrpc_reply{error = Error}). + #'RpcReply'{ + reply = {error, #'RpcReply.RpcError'{code = Code, message = Message}} + }. + +-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]). diff --git a/src/efka_transport.erl b/src/efka_transport.erl index e4850ca..e482894 100644 --- a/src/efka_transport.erl +++ b/src/efka_transport.erl @@ -32,9 +32,9 @@ %%% API %%%=================================================================== --spec auth_request(Pid :: pid(), AuthBin :: binary()) -> no_return(). -auth_request(Pid, AuthBin) when is_pid(Pid), is_binary(AuthBin) -> - gen_server:cast(Pid, {auth_request, AuthBin}). +-spec auth_request(Pid :: pid(), AuthRequest :: any()) -> no_return(). +auth_request(Pid, AuthRequest) when is_pid(Pid) -> + 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(). -rpc_reply(undefined, PacketId, Response) when is_integer(PacketId), is_binary(Response) -> +-spec rpc_reply(Pid :: pid() | undefined, PacketId :: integer(), Response :: any()) -> no_return(). +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 = message_codec:encode_request(PacketId, ?MESSAGE_AUTH_REQUEST, AuthRequest), + ok = ssl:send(Socket, Packet), %% 需要等待auth返回的结果 receive - {ssl, Socket, <>} -> - {ok, #auth_reply{} = Reply} = message_codec:decode(ReplyBin), - ParentPid ! {auth_reply, {ok, Reply}}, - {noreply, State}; - {ssl, Socket, Info} -> - logger:warning("[efka_transport] get invalid auth_reply: ~p", [Info]), - ParentPid ! {auth_reply, {error, invalid_auth_reply}}, + {ssl, Socket, ReplyBin} -> + case message_codec:decode_response(ReplyBin) of + {ok, PacketId, Reply} -> + ParentPid ! {auth_reply, {ok, Reply}}; + {ok, ReplyPacketId, _Reply} -> + logger:warning("[efka_transport] get unexpected auth_reply packet_id: ~p", [ReplyPacketId]), + 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, <>), + ok = ssl:send(Socket, Packet), {noreply, State}; %% 服务push的消息的回复 handle_cast({rpc_reply, PacketId, Reply}, State = #state{socket = Socket}) -> - ok = ssl:send(Socket, <>), + 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, <>}, State = #state{socket = Socket, parent_pid = ParentPid}) -> - {ok, CastRequest} = message_codec:decode(CastBin), - ParentPid ! {server_cast, CastRequest}, - {noreply, State}; - -handle_info({ssl, Socket, <>}, State = #state{socket = Socket, parent_pid = ParentPid}) -> - {ok, RPCRequest} = message_codec:decode(RPCRequestBin), - ParentPid ! {server_rpc, PacketId, RPCRequest}, +%% 服务器主动推送的数据 +handle_info({ssl, Socket, PacketBin}, State = #state{socket = Socket, parent_pid = ParentPid}) -> + case message_codec:decode_request(PacketBin) of + {ok, PacketId, Request} -> + ParentPid ! {server_rpc, PacketId, Request}; + error -> + case message_codec:decode_cast(PacketBin) of + {ok, CastRequest} -> + 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}) -> diff --git a/src/message/message_codec.erl b/src/message/message_codec.erl index c4c6cc9..450d10f 100644 --- a/src/message/message_codec.erl +++ b/src/message/message_codec.erl @@ -9,133 +9,122 @@ -module(message_codec). -author("anlicheng"). -include("message.hrl"). - --define(I8, 1). --define(I16, 2). --define(I32, 3). --define(Bytes, 4). +-include("message_pb.hrl"). %% 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), - <>. -encode0(#auth_request{uuid = UUID, username = Username, salt = Salt, token = Token, timestamp = Timestamp}) -> - iolist_to_binary([ - marshal(?Bytes, UUID), - marshal(?Bytes, Username), - marshal(?Bytes, Salt), - marshal(?Bytes, Token), - 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) - ]); + case frame_for_type(MessageType, Message, 0) of + {request, Frame} -> + message_pb:encode_msg(Frame); + {response, Frame} -> + message_pb:encode_msg(Frame); + {cast, Frame} -> + message_pb:encode_msg(Frame) + end. -encode0(#jsonrpc_request{method = Method, params = Params}) -> - ReqBody = erlang:term_to_binary(#{<<"method">> => Method, <<"params">> => Params}), - iolist_to_binary([ - marshal(?Bytes, ReqBody) - ]); -encode0(#data{route_key = RouteKey, metric = Metric}) -> - iolist_to_binary([ - marshal(?Bytes, RouteKey), - 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 encode_request(PacketId :: integer(), MessageType :: integer(), Message :: any()) -> binary(). +encode_request(PacketId, MessageType, Message) when is_integer(PacketId), is_integer(MessageType) -> + case frame_for_type(MessageType, Message, PacketId) of + {request, Frame} -> + message_pb:encode_msg(Frame); + _ -> + erlang:error({unsupported_request_type, MessageType}) + end. --spec decode(Bin :: binary()) -> {ok, Message :: any()} | error. -decode(<>) -> - case unmarshal(Packet) of - {ok, Fields} -> - decode0(PacketType, Fields); - error -> +-spec encode_response(PacketId :: integer(), MessageType :: integer(), Message :: any()) -> binary(). +encode_response(PacketId, MessageType, Message) when is_integer(PacketId), is_integer(MessageType) -> + case frame_for_type(MessageType, Message, PacketId) of + {response, Frame} -> + message_pb:encode_msg(Frame); + _ -> + 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}}; -decode0(?MESSAGE_JSONRPC_REPLY, [ReplyBin]) -> - case erlang:binary_to_term(ReplyBin) of - #{<<"result">> := Result} -> - {ok, #jsonrpc_reply{result = Result}}; - #{<<"error">> := Error} -> - {ok, #jsonrpc_reply{error = Error}}; - _ -> + +-spec decode_response(Bin :: binary()) -> {ok, PacketId :: integer(), Message :: any()} | error. +decode_response(Bin) when is_binary(Bin) -> + try + Frame = message_pb:decode_msg(Bin, 'ResponseFrame'), + decode_response_frame(Frame) + catch + _:_ -> error - end; -decode0(?MESSAGE_PUB, [Topic, Qos, Content]) -> - {ok, #pub{topic = Topic, qos = Qos, content = Content}}; -decode0(?MESSAGE_COMMAND, [CommandType, Command]) -> - {ok, #command{command_type = CommandType, command = Command}}; -decode0(?MESSAGE_AUTH_REPLY, [Code, Payload]) -> - {ok, #auth_reply{code = Code, payload = Payload}}; -decode0(?MESSAGE_JSONRPC_REQUEST, [ReqBody]) -> - #{<<"method">> := Method, <<"params">> := Params} = erlang:binary_to_term(ReqBody), - {ok, #jsonrpc_request{method = Method, params = Params}}; -decode0(?MESSAGE_DATA, [RouteKey, Metric]) -> - {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. + end. + +-spec decode_cast(Bin :: binary()) -> {ok, Message :: any()} | error. +decode_cast(Bin) when is_binary(Bin) -> + try + Frame = message_pb:decode_msg(Bin, 'CastFrame'), + decode_cast_frame(Frame) + catch + _:_ -> + error + end. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %%% helper methods %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% --spec marshal(Type :: integer(), Field :: any()) -> binary(). -marshal(?I8, Field) when is_integer(Field) -> - <>; -marshal(?I16, Field) when is_integer(Field) -> - <>; -marshal(?I32, Field) when is_integer(Field) -> - <>; -marshal(?Bytes, Field) when is_binary(Field) -> - Len = byte_size(Field), - <>. +frame_for_type(?MESSAGE_AUTH_REQUEST, Message = #'AuthRequest'{}, PacketId) -> + {request, #'RequestFrame'{packet_id = PacketId, body = {auth_request, Message}}}; +frame_for_type(?MESSAGE_JSONRPC_REQUEST, Message = #'RpcRequest'{}, PacketId) -> + {request, #'RequestFrame'{packet_id = PacketId, body = {rpc_request, Message}}}; +frame_for_type(?MESSAGE_AUTH_REPLY, Message = #'AuthReply'{}, PacketId) -> + {response, #'ResponseFrame'{packet_id = PacketId, body = {auth_reply, Message}}}; +frame_for_type(?MESSAGE_JSONRPC_REPLY, Message = #'RpcReply'{}, PacketId) -> + {response, #'ResponseFrame'{packet_id = PacketId, body = {rpc_reply, Message}}}; +frame_for_type(?MESSAGE_PUB, Message = #'Pub'{}, _PacketId) -> + {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. -unmarshal(Bin) when is_binary(Bin) -> - unmarshal(Bin, []). -unmarshal(<<>>, Acc) -> - {ok, lists:reverse(Acc)}; -unmarshal(<>, Acc) -> - unmarshal(Rest, [F|Acc]); -unmarshal(<>, Acc) -> - unmarshal(Rest, [F|Acc]); -unmarshal(<>, Acc) -> - unmarshal(Rest, [F|Acc]); -unmarshal(<>, Acc) -> - unmarshal(Rest, [F|Acc]); -unmarshal(_, _) -> +decode_request_frame(#'RequestFrame'{packet_id = PacketId, body = {auth_request, Message}}) + when is_integer(PacketId), PacketId > 0 -> + {ok, PacketId, Message}; +decode_request_frame(#'RequestFrame'{packet_id = PacketId, body = {rpc_request, Message}}) + when is_integer(PacketId), PacketId > 0 -> + {ok, PacketId, Message}; +decode_request_frame(#'RequestFrame'{packet_id = PacketId, body = {container_request, Message}}) + when is_integer(PacketId), PacketId > 0 -> + {ok, PacketId, Message}; +decode_request_frame(_) -> + error. + +decode_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {auth_reply, Message}}) + 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.