132 lines
4.8 KiB
Erlang
132 lines
4.8 KiB
Erlang
%%%-------------------------------------------------------------------
|
|
%%% @author anlicheng
|
|
%%% @copyright (C) 2025, <COMPANY>
|
|
%%% @doc
|
|
%%%
|
|
%%% @end
|
|
%%% Created : 17. 9月 2025 16:05
|
|
%%%-------------------------------------------------------------------
|
|
-module(message_codec).
|
|
-author("anlicheng").
|
|
-include("message.hrl").
|
|
|
|
-define(I32, 1).
|
|
-define(Bytes, 2).
|
|
|
|
%% API
|
|
-export([encode/2, decode/1]).
|
|
|
|
-spec encode(MessageType :: integer(), Message :: any()) -> binary().
|
|
encode(MessageType, Message) when is_integer(MessageType) ->
|
|
Bin = encode0(Message),
|
|
<<MessageType, Bin/binary>>.
|
|
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(#rpc_reply{code = Code, payload = Message}) ->
|
|
iolist_to_binary([
|
|
marshal(?I32, Code),
|
|
marshal(?Bytes, Message)
|
|
]);
|
|
encode0(#pub{topic = Topic, content = Content}) ->
|
|
iolist_to_binary([
|
|
marshal(?Bytes, Topic),
|
|
marshal(?Bytes, Content)
|
|
]);
|
|
encode0(#command{command_type = CommandType, command = Command}) ->
|
|
iolist_to_binary([
|
|
marshal(?I32, CommandType),
|
|
marshal(?Bytes, Command)
|
|
]);
|
|
encode0(#rpc_deploy{task_id = TaskId, config = Config}) ->
|
|
iolist_to_binary([
|
|
marshal(?I32, TaskId),
|
|
marshal(?Bytes, Config)
|
|
]);
|
|
encode0(#rpc_container{method = Method, container_name = ContainerName, params = Params}) ->
|
|
iolist_to_binary([
|
|
marshal(?Bytes, Method),
|
|
marshal(?Bytes, ContainerName),
|
|
marshal(?Bytes, Params)
|
|
]);
|
|
encode0(#data{service_id = ServiceId, device_uuid = DeviceUUID, route_key = RouteKey, metric = Metric}) ->
|
|
iolist_to_binary([
|
|
marshal(?Bytes, ServiceId),
|
|
marshal(?Bytes, DeviceUUID),
|
|
marshal(?Bytes, RouteKey),
|
|
marshal(?Bytes, Metric)
|
|
]);
|
|
encode0(#event{service_id = ServiceId, event_type = EventType, params = Params}) ->
|
|
iolist_to_binary([
|
|
marshal(?Bytes, ServiceId),
|
|
marshal(?I32, EventType),
|
|
marshal(?Bytes, Params)
|
|
]);
|
|
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.
|
|
decode(<<PacketType:8, Packet/binary>>) ->
|
|
case unmarshal(Packet) of
|
|
{ok, Fields} ->
|
|
decode0(PacketType, Fields);
|
|
error ->
|
|
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_RPC_REPLY, [Code, Message]) ->
|
|
{ok, #rpc_reply{code = Code, payload = Message}};
|
|
decode0(?MESSAGE_PUB, [Topic, Content]) ->
|
|
{ok, #pub{topic = Topic, content = Content}};
|
|
decode0(?MESSAGE_COMMAND, [CommandType, Command]) ->
|
|
{ok, #command{command_type = CommandType, command = Command}};
|
|
decode0(?MESSAGE_RPC_DEPLOY, [TaskId, Config]) ->
|
|
{ok, #rpc_deploy{task_id = TaskId, config = Config}};
|
|
decode0(?MESSAGE_RPC_CONTAINER, [Method, ContainerName, Params]) ->
|
|
{ok, #rpc_container{method = Method, container_name = ContainerName, params = Params}};
|
|
decode0(?MESSAGE_DATA, [ServiceId, DeviceUUID, RouteKey, Metric]) ->
|
|
{ok, #data{service_id = ServiceId, device_uuid = DeviceUUID, route_key = RouteKey, metric = Metric}};
|
|
decode0(?MESSAGE_EVENT, [ServiceId, EventType, Params]) ->
|
|
{ok, #event{service_id = ServiceId, event_type = EventType, params = Params}};
|
|
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().
|
|
marshal(?I32, Field) when is_integer(Field) ->
|
|
<<?I32, Field:32>>;
|
|
marshal(?Bytes, Field) when is_binary(Field) ->
|
|
Len = byte_size(Field),
|
|
<<?Bytes, Len:16, Field/binary>>;
|
|
marshal(?Bytes, Field0) ->
|
|
Field = unicode:characters_to_binary(Field0),
|
|
Len = byte_size(Field),
|
|
<<?Bytes, Len:16, Field/binary>>.
|
|
|
|
-spec unmarshal(Bin :: binary()) -> {ok, Components :: [any()]} | error.
|
|
unmarshal(Bin) when is_binary(Bin) ->
|
|
unmarshal(Bin, []).
|
|
unmarshal(<<>>, Acc) ->
|
|
{ok, lists:reverse(Acc)};
|
|
unmarshal(<<?I32, F:32, Rest/binary>>, Acc) ->
|
|
unmarshal(Rest, [F|Acc]);
|
|
unmarshal(<<?Bytes, Len:16, F:Len/binary, Rest/binary>>, Acc) ->
|
|
unmarshal(Rest, [F|Acc]);
|
|
unmarshal(_, _) ->
|
|
error.
|