anlicheng/iot_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix proto

Browse Source
This commit is contained in:
anlicheng 2026-04-26 15:33:33 +08:00
parent 464c04a277
commit a2a4b0cb6d
6 changed files with 234 additions and 6159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

275
include/message_pb.hrl
View File

@ -1,275 +0,0 @@
%% -*- coding: utf-8 -*-
%% Automatically generated, do not edit
%% Generated by gpb_compile version 4.21.7
-ifndef(message_pb).
-define(message_pb, true).
-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'()} | {container_request, message_pb:'ContainerRequest'()} | undefined % oneof
}).
-endif.
-ifndef('AUTHREQUEST_PB_H').
-define('AUTHREQUEST_PB_H', true).
-record('AuthRequest',
{uuid = <<>> :: iodata() | undefined, % = 1, optional
token = <<>> :: iodata() | undefined, % = 2, optional
timestamp = 0 :: integer() | undefined % = 3, optional, 32 bits
}).
-endif.
-ifndef('REPLYFRAME.ERROR_PB_H').
-define('REPLYFRAME.ERROR_PB_H', true).
-record('ReplyFrame.Error',
{code = 0 :: integer() | undefined, % = 1, optional, 32 bits
message = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('REPLYFRAME_PB_H').
-define('REPLYFRAME_PB_H', true).
-record('ReplyFrame',
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
reply :: {result, iodata()} | {error, message_pb:'ReplyFrame.Error'()} | undefined % oneof
}).
-endif.
-ifndef('CASTFRAME.PUB_PB_H').
-define('CASTFRAME.PUB_PB_H', true).
-record('CastFrame.Pub',
{topic = <<>> :: iodata() | undefined, % = 1, optional
qos = 0 :: integer() | undefined, % = 2, optional, 32 bits
content = <<>> :: iodata() | undefined % = 3, optional
}).
-endif.
-ifndef('CASTFRAME.COMMAND_PB_H').
-define('CASTFRAME.COMMAND_PB_H', true).
-record('CastFrame.Command',
{command :: {auth, message_pb:'CastFrame.Command.Authorization'()} | undefined % oneof
}).
-endif.
-ifndef('CASTFRAME.COMMAND.AUTHORIZATION_PB_H').
-define('CASTFRAME.COMMAND.AUTHORIZATION_PB_H', true).
-record('CastFrame.Command.Authorization',
{cmd = 'DEACTIVATE' :: 'DEACTIVATE' | 'ACTIVATE' | integer() | undefined % = 1, optional, enum CastFrame.Command.Authorization.CommandType
}).
-endif.
-ifndef('CASTFRAME.DATA_PB_H').
-define('CASTFRAME.DATA_PB_H', true).
-record('CastFrame.Data',
{route_key = <<>> :: iodata() | undefined, % = 1, optional
metric = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('CASTFRAME.TASKEVENT_PB_H').
-define('CASTFRAME.TASKEVENT_PB_H', true).
-record('CastFrame.TaskEvent',
{task_id = 0 :: integer() | undefined, % = 1, optional, 32 bits
type = <<>> :: iodata() | undefined, % = 2, optional
stream = <<>> :: iodata() | undefined % = 3, optional
}).
-endif.
-ifndef('CASTFRAME_PB_H').
-define('CASTFRAME_PB_H', true).
-record('CastFrame',
{body :: {pub, message_pb:'CastFrame.Pub'()} | {command, message_pb:'CastFrame.Command'()} | {data, message_pb:'CastFrame.Data'()} | {task_event, message_pb:'CastFrame.TaskEvent'()} | undefined % oneof
}).
-endif.
-ifndef('CONTAINERREF_PB_H').
-define('CONTAINERREF_PB_H', true).
-record('ContainerRef',
{id = <<>> :: unicode:chardata() | undefined, % = 1, optional
name = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.LIST_PB_H').
-define('CONTAINERREQUEST.LIST_PB_H', true).
-record('ContainerRequest.List',
{all = false :: boolean() | 0 | 1 | undefined % = 1, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.DEPLOY_PB_H').
-define('CONTAINERREQUEST.DEPLOY_PB_H', true).
-record('ContainerRequest.Deploy',
{task_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
params = undefined :: message_pb:'ContainerDeployParams'() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.START_PB_H').
-define('CONTAINERREQUEST.START_PB_H', true).
-record('ContainerRequest.Start',
{target = undefined :: message_pb:'ContainerRef'() | undefined % = 1, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.STOP_PB_H').
-define('CONTAINERREQUEST.STOP_PB_H', true).
-record('ContainerRequest.Stop',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
timeout_seconds = 0 :: non_neg_integer() | undefined % = 2, optional, 32 bits
}).
-endif.
-ifndef('CONTAINERREQUEST.KILL_PB_H').
-define('CONTAINERREQUEST.KILL_PB_H', true).
-record('ContainerRequest.Kill',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
signal = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.REMOVE_PB_H').
-define('CONTAINERREQUEST.REMOVE_PB_H', true).
-record('ContainerRequest.Remove',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
force = false :: boolean() | 0 | 1 | undefined, % = 2, optional
remove_volumes = false :: boolean() | 0 | 1 | undefined % = 3, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.CONFIG_PB_H').
-define('CONTAINERREQUEST.CONFIG_PB_H', true).
-record('ContainerRequest.Config',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
config = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST_PB_H').
-define('CONTAINERREQUEST_PB_H', true).
-record('ContainerRequest',
{action :: {list, message_pb:'ContainerRequest.List'()} | {deploy, message_pb:'ContainerRequest.Deploy'()} | {start, message_pb:'ContainerRequest.Start'()} | {stop, message_pb:'ContainerRequest.Stop'()} | {kill, message_pb:'ContainerRequest.Kill'()} | {remove, message_pb:'ContainerRequest.Remove'()} | {config, message_pb:'ContainerRequest.Config'()} | undefined % oneof
}).
-endif.
-ifndef('CONTAINERDEPLOYPARAMS_PB_H').
-define('CONTAINERDEPLOYPARAMS_PB_H', true).
-record('ContainerDeployParams',
{container_name = <<>> :: unicode:chardata() | undefined, % = 1, optional
container_dir = <<>> :: unicode:chardata() | undefined, % = 2, optional
create = undefined :: message_pb:'DockerCreateOptions'() | undefined % = 3, optional
}).
-endif.
-ifndef('DOCKERCREATEOPTIONS_PB_H').
-define('DOCKERCREATEOPTIONS_PB_H', true).
-record('DockerCreateOptions',
{config = undefined :: message_pb:'DockerContainerConfig'() | undefined, % = 1, optional
host_config = undefined :: message_pb:'DockerHostConfig'() | undefined, % = 2, optional
networking_config = undefined :: message_pb:'DockerNetworkingConfig'() | undefined % = 3, optional
}).
-endif.
-ifndef('DOCKERCONTAINERCONFIG_PB_H').
-define('DOCKERCONTAINERCONFIG_PB_H', true).
-record('DockerContainerConfig',
{image = <<>> :: unicode:chardata() | undefined, % = 1, optional
cmd = [] :: [unicode:chardata()] | undefined, % = 2, repeated
entrypoint = [] :: [unicode:chardata()] | undefined, % = 3, repeated
env = [] :: [unicode:chardata()] | undefined, % = 4, repeated
labels = [] :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 5
volumes = [] :: [unicode:chardata()] | undefined, % = 6, repeated
user = <<>> :: unicode:chardata() | undefined, % = 7, optional
working_dir = <<>> :: unicode:chardata() | undefined, % = 8, optional
hostname = <<>> :: unicode:chardata() | undefined, % = 9, optional
exposed_ports = [] :: [message_pb:'DockerExposedPort'()] | undefined, % = 10, repeated
healthcheck = undefined :: message_pb:'Healthcheck'() | undefined % = 11, optional
}).
-endif.
-ifndef('DOCKERHOSTCONFIG_PB_H').
-define('DOCKERHOSTCONFIG_PB_H', true).
-record('DockerHostConfig',
{binds = [] :: [unicode:chardata()] | undefined, % = 1, repeated
network_mode = <<>> :: unicode:chardata() | undefined, % = 2, optional
restart_policy = undefined :: message_pb:'RestartPolicy'() | undefined, % = 3, optional
privileged = false :: boolean() | 0 | 1 | undefined, % = 4, optional
cap_add = [] :: [unicode:chardata()] | undefined, % = 5, repeated
cap_drop = [] :: [unicode:chardata()] | undefined, % = 6, repeated
devices = [] :: [message_pb:'DeviceMapping'()] | undefined, % = 7, repeated
memory = 0 :: non_neg_integer() | undefined, % = 8, optional, 64 bits
memory_reservation = 0 :: non_neg_integer() | undefined, % = 9, optional, 64 bits
nano_cpus = 0 :: non_neg_integer() | undefined, % = 10, optional, 64 bits
cpu_shares = 0 :: non_neg_integer() | undefined, % = 11, optional, 64 bits
ulimits = [] :: [message_pb:'Ulimit'()] | undefined, % = 12, repeated
tmpfs = [] :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 13
sysctls = [] :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 14
extra_hosts = [] :: [unicode:chardata()] | undefined % = 15, repeated
}).
-endif.
-ifndef('DOCKERNETWORKINGCONFIG_PB_H').
-define('DOCKERNETWORKINGCONFIG_PB_H', true).
-record('DockerNetworkingConfig',
{endpoints = [] :: [message_pb:'DockerNetworkEndpoint'()] | undefined % = 1, repeated
}).
-endif.
-ifndef('DOCKERNETWORKENDPOINT_PB_H').
-define('DOCKERNETWORKENDPOINT_PB_H', true).
-record('DockerNetworkEndpoint',
{name = <<>> :: unicode:chardata() | undefined % = 1, optional
}).
-endif.
-ifndef('DOCKEREXPOSEDPORT_PB_H').
-define('DOCKEREXPOSEDPORT_PB_H', true).
-record('DockerExposedPort',
{container_port = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
protocol = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('HEALTHCHECK_PB_H').
-define('HEALTHCHECK_PB_H', true).
-record('Healthcheck',
{test = [] :: [unicode:chardata()] | undefined, % = 1, repeated
interval_ns = 0 :: non_neg_integer() | undefined, % = 2, optional, 64 bits
timeout_ns = 0 :: non_neg_integer() | undefined, % = 3, optional, 64 bits
retries = 0 :: non_neg_integer() | undefined % = 4, optional, 32 bits
}).
-endif.
-ifndef('RESTARTPOLICY_PB_H').
-define('RESTARTPOLICY_PB_H', true).
-record('RestartPolicy',
{name = <<>> :: unicode:chardata() | undefined, % = 1, optional
maximum_retry_count = 0 :: non_neg_integer() | undefined % = 2, optional, 32 bits
}).
-endif.
-ifndef('DEVICEMAPPING_PB_H').
-define('DEVICEMAPPING_PB_H', true).
-record('DeviceMapping',
{path_on_host = <<>> :: unicode:chardata() | undefined, % = 1, optional
path_in_container = <<>> :: unicode:chardata() | undefined, % = 2, optional
cgroup_permissions = <<>> :: unicode:chardata() | undefined % = 3, optional
}).
-endif.
-ifndef('ULIMIT_PB_H').
-define('ULIMIT_PB_H', true).
-record('Ulimit',
{name = <<>> :: unicode:chardata() | undefined, % = 1, optional
soft = 0 :: non_neg_integer() | undefined, % = 2, optional, 64 bits
hard = 0 :: non_neg_integer() | undefined % = 3, optional, 64 bits
}).
-endif.
-endif.

29
rebar.config
View File

@ -1,34 +1,5 @@
{erl_opts, [debug_info]}.
{plugins, [
{rebar3_gpb_plugin, ".*", {git, "https://github.com/lrascao/rebar3_gpb_plugin.git", {tag, "2.23.8"}}}
]}.
% ======================
% 核心：强制指定 proto 路径 + 输出到你的 apps/sdlan
% ======================
{gpb_opts, [
{i, "../proto"}, % 共享 proto 文件路径
{src_dirs, ["../proto"]}, % 源码目录（必须）
{f, ["message.proto"]},
recursive, % 递归查找 proto 文件
{module_name_suffix, "_pb"}, % 生成模块后缀
{o_erl, "src/protobuf"}, % .erl 输出目录
{o_hrl, "include"}, % .hrl 输出目录
include_as_lib, % gpb.hrl 通过 -include_lib("gpb/include/gpb.hrl")
{strings_as_binaries, true}, % proto string → Erlang binary
type_specs, % 生成 type specs
report, % 编译报告
verbose % 打印详细信息
]}.
{provider_hooks, [
{pre, [
{compile, {protobuf, compile}},
{clean, {protobuf, clean}}
]}
]}.
{deps, [
{poolboy, ".*", {git, "https://github.com/devinus/poolboy.git", {tag, "1.5.1"}}},
{hackney, ".*", {git, "https://github.com/benoitc/hackney.git", {tag, "1.25.0"}}},

216
src/docker/docker_container_builder.erl
View File

@ -2,70 +2,71 @@
%%% @author
%%% @copyright (C) 2026, <COMPANY>
%%% @doc
%%% ContainerRequest protobuf builder helpers.
%%% ContainerRequest wire builder helpers.
%%% @end
%%%-------------------------------------------------------------------
-module(docker_container_builder).
-include("message_pb.hrl").
-export([list_request/0, config_request/2, deploy_request/2, start_request/1, stop_request/1, kill_request/1, remove_request/1]).
-spec list_request() -> message_pb:'ContainerRequest'().
-spec list_request() -> map().
list_request() ->
#'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = true}}}.
#{action => list, all => true}.
-spec config_request(ContainerName :: binary(), ConfigJson :: binary()) -> message_pb:'ContainerRequest'().
-spec config_request(binary(), binary()) -> map().
config_request(ContainerName, ConfigJson) when is_binary(ContainerName), is_binary(ConfigJson) ->
#'ContainerRequest'{action = {config, #'ContainerRequest.Config'{
target = container_ref(ContainerName),
config = ConfigJson
}}}.
#{
action => config,
target => container_ref(ContainerName),
config => ConfigJson
}.
-spec deploy_request(TaskId :: integer(), Config :: map()) ->
{ok, message_pb:'ContainerRequest'()} | {error, binary()}.
-spec deploy_request(integer(), map()) -> {ok, map()} | {error, binary()}.
deploy_request(TaskId, Config) when is_integer(TaskId), is_map(Config), TaskId >= 0 ->
try
ensure_supported_deploy_config(Config),
validate_deploy_config(Config),
Params = build_container_deploy_params(Config),
{ok, #'ContainerRequest'{action = {deploy, #'ContainerRequest.Deploy'{task_id = TaskId, params = Params}}}}
{ok, #{action => deploy, task_id => TaskId, params => Params}}
catch
throw:{error, Reason} ->
{error, Reason}
end.
-spec start_request(ContainerName :: binary()) -> message_pb:'ContainerRequest'().
-spec start_request(binary()) -> map().
start_request(ContainerName) when is_binary(ContainerName) ->
#'ContainerRequest'{action = {start, #'ContainerRequest.Start'{target = container_ref(ContainerName)}}}.
#{action => start, target => container_ref(ContainerName)}.
-spec stop_request(ContainerName :: binary()) -> message_pb:'ContainerRequest'().
-spec stop_request(binary()) -> map().
stop_request(ContainerName) when is_binary(ContainerName) ->
#'ContainerRequest'{action = {stop, #'ContainerRequest.Stop'{
target = container_ref(ContainerName),
timeout_seconds = 0
}}}.
#{
action => stop,
target => container_ref(ContainerName),
timeout_seconds => 0
}.
-spec kill_request(ContainerName :: binary()) -> message_pb:'ContainerRequest'().
-spec kill_request(binary()) -> map().
kill_request(ContainerName) when is_binary(ContainerName) ->
#'ContainerRequest'{action = {kill, #'ContainerRequest.Kill'{
target = container_ref(ContainerName),
signal = <<>>
}}}.
#{
action => kill,
target => container_ref(ContainerName),
signal => <<>>
}.
-spec remove_request(ContainerName :: binary()) -> message_pb:'ContainerRequest'().
-spec remove_request(binary()) -> map().
remove_request(ContainerName) when is_binary(ContainerName) ->
#'ContainerRequest'{action = {remove, #'ContainerRequest.Remove'{
target = container_ref(ContainerName),
force = false,
remove_volumes = false
}}}.
#{
action => remove,
target => container_ref(ContainerName),
force => false,
remove_volumes => false
}.
-spec container_ref(ContainerName :: binary()) -> message_pb:'ContainerRef'().
-spec container_ref(binary()) -> map().
container_ref(ContainerName) when is_binary(ContainerName) ->
#'ContainerRef'{name = ContainerName}.
#{name => ContainerName, id => <<>>}.
-spec ensure_supported_deploy_config(Config :: map()) -> ok.
-spec ensure_supported_deploy_config(map()) -> ok.
ensure_supported_deploy_config(Config) when is_map(Config) ->
UnsupportedKeys = [Key || Key <- [<<"ports">>], maps:is_key(Key, Config)],
case UnsupportedKeys of
@ -76,7 +77,7 @@ ensure_supported_deploy_config(Config) when is_map(Config) ->
throw({error, <<"unsupported container config keys: ", Unsupported/binary>>})
end.
-spec validate_deploy_config(Config :: map()) -> ok.
-spec validate_deploy_config(map()) -> ok.
validate_deploy_config(Config) when is_map(Config) ->
Required = [
{<<"image">>, binary},
@ -179,7 +180,7 @@ type_name({map, {binary, any}}) ->
type_name(boolean) ->
<<"boolean">>.
-spec check_type(Value :: any(), any()) -> boolean().
-spec check_type(any(), any()) -> boolean().
check_type(Value, binary) ->
is_binary(Value);
check_type(Value, integer) ->
@ -205,98 +206,81 @@ check_type(Value, boolean) ->
check_type(_, _) ->
false.
-spec build_container_deploy_params(Config :: map()) -> message_pb:'ContainerDeployParams'().
-spec build_container_deploy_params(map()) -> map().
build_container_deploy_params(Config) when is_map(Config) ->
ContainerName = maps:get(<<"container_name">>, Config),
ContainerDir = maps:get(<<"container_dir">>, Config, <<>>),
Create = build_docker_create_options(Config),
#'ContainerDeployParams'{
container_name = ContainerName,
container_dir = ContainerDir,
create = Create
#{
container_name => maps:get(<<"container_name">>, Config),
container_dir => maps:get(<<"container_dir">>, Config, <<>>),
create => build_docker_create_options(Config)
}.
-spec build_docker_create_options(Config :: map()) -> message_pb:'DockerCreateOptions'().
-spec build_docker_create_options(map()) -> map().
build_docker_create_options(Config) when is_map(Config) ->
#'DockerCreateOptions'{
config = build_docker_container_config(Config),
host_config = build_docker_host_config(Config),
networking_config = build_docker_networking_config(Config)
#{
config => build_docker_container_config(Config),
host_config => build_docker_host_config(Config),
networking_config => build_docker_networking_config(Config)
}.
-spec build_docker_container_config(Config :: map()) -> message_pb:'DockerContainerConfig'().
-spec build_docker_container_config(map()) -> map().
build_docker_container_config(Config) when is_map(Config) ->
Volumes = build_container_volumes(maps:get(<<"volumes">>, Config, [])),
ExposedPorts = build_exposed_ports(maps:get(<<"expose">>, Config, [])),
Healthcheck = build_healthcheck(maps:get(<<"healthcheck">>, Config, undefined)),
#'DockerContainerConfig'{
image = maps:get(<<"image">>, Config),
cmd = maps:get(<<"command">>, Config),
entrypoint = maps:get(<<"entrypoint">>, Config, []),
env = maps:get(<<"envs">>, Config, []),
labels = maps:to_list(maps:get(<<"labels">>, Config, #{})),
volumes = Volumes,
user = maps:get(<<"user">>, Config, <<>>),
working_dir = maps:get(<<"working_dir">>, Config, <<>>),
hostname = maps:get(<<"hostname">>, Config, <<>>),
exposed_ports = ExposedPorts,
healthcheck = Healthcheck
#{
image => maps:get(<<"image">>, Config),
cmd => maps:get(<<"command">>, Config),
entrypoint => maps:get(<<"entrypoint">>, Config, []),
env => maps:get(<<"envs">>, Config, []),
labels => maps:get(<<"labels">>, Config, #{}),
volumes => build_container_volumes(maps:get(<<"volumes">>, Config, [])),
user => maps:get(<<"user">>, Config, <<>>),
working_dir => maps:get(<<"working_dir">>, Config, <<>>),
hostname => maps:get(<<"hostname">>, Config, <<>>),
exposed_ports => build_exposed_ports(maps:get(<<"expose">>, Config, [])),
healthcheck => build_healthcheck(maps:get(<<"healthcheck">>, Config, undefined))
}.
-spec build_docker_host_config(Config :: map()) -> message_pb:'DockerHostConfig'().
-spec build_docker_host_config(map()) -> map().
build_docker_host_config(Config) when is_map(Config) ->
Binds = build_host_binds(maps:get(<<"volumes">>, Config, [])),
RestartPolicy = build_restart_policy(maps:get(<<"restart">>, Config)),
Devices = build_device_mappings(maps:get(<<"devices">>, Config, [])),
Ulimits = build_ulimits(maps:get(<<"ulimits">>, Config, #{})),
Tmpfs = build_tmpfs_options(maps:get(<<"tmpfs">>, Config, [])),
Memory = default_uint64(parse_optional_size_bytes(maps:get(<<"mem_limit">>, Config, undefined), <<"mem_limit">>)),
MemoryReservation = default_uint64(parse_optional_size_bytes(maps:get(<<"mem_reservation">>, Config, undefined), <<"mem_reservation">>)),
NanoCpus = default_uint64(parse_optional_nano_cpus(maps:get(<<"cpus">>, Config, undefined))),
CpuShares = default_uint64(maps:get(<<"cpu_shares">>, Config, undefined)),
#'DockerHostConfig'{
binds = Binds,
network_mode = maps:get(<<"network_mode">>, Config, <<>>),
restart_policy = RestartPolicy,
privileged = maps:get(<<"privileged">>, Config, false),
cap_add = maps:get(<<"cap_add">>, Config, []),
cap_drop = maps:get(<<"cap_drop">>, Config, []),
devices = Devices,
memory = Memory,
memory_reservation = MemoryReservation,
nano_cpus = NanoCpus,
cpu_shares = CpuShares,
ulimits = Ulimits,
tmpfs = Tmpfs,
sysctls = maps:to_list(maps:get(<<"sysctls">>, Config, #{})),
extra_hosts = maps:get(<<"extra_hosts">>, Config, [])
#{
binds => build_host_binds(maps:get(<<"volumes">>, Config, [])),
network_mode => maps:get(<<"network_mode">>, Config, <<>>),
restart_policy => build_restart_policy(maps:get(<<"restart">>, Config)),
privileged => maps:get(<<"privileged">>, Config, false),
cap_add => maps:get(<<"cap_add">>, Config, []),
cap_drop => maps:get(<<"cap_drop">>, Config, []),
devices => build_device_mappings(maps:get(<<"devices">>, Config, [])),
memory => default_uint64(parse_optional_size_bytes(maps:get(<<"mem_limit">>, Config, undefined), <<"mem_limit">>)),
memory_reservation => default_uint64(parse_optional_size_bytes(maps:get(<<"mem_reservation">>, Config, undefined), <<"mem_reservation">>)),
nano_cpus => default_uint64(parse_optional_nano_cpus(maps:get(<<"cpus">>, Config, undefined))),
cpu_shares => default_uint64(maps:get(<<"cpu_shares">>, Config, undefined)),
ulimits => build_ulimits(maps:get(<<"ulimits">>, Config, #{})),
tmpfs => maps:from_list(build_tmpfs_options(maps:get(<<"tmpfs">>, Config, []))),
sysctls => maps:get(<<"sysctls">>, Config, #{}),
extra_hosts => maps:get(<<"extra_hosts">>, Config, [])
}.
-spec build_docker_networking_config(Config :: map()) -> message_pb:'DockerNetworkingConfig'().
-spec build_docker_networking_config(map()) -> map().
build_docker_networking_config(Config) when is_map(Config) ->
Networks = maps:get(<<"networks">>, Config, []),
#'DockerNetworkingConfig'{
endpoints = [#'DockerNetworkEndpoint'{name = Network} || Network <- Networks]
}.
#{endpoints => [#{name => Network} || Network <- Networks]}.
-spec build_restart_policy(binary()) -> message_pb:'RestartPolicy'().
-spec build_restart_policy(binary()) -> map().
build_restart_policy(Restart0) when is_binary(Restart0) ->
case binary:split(Restart0, <<":">>) of
[Name, RetryCountBin] ->
#'RestartPolicy'{name = Name, maximum_retry_count = parse_uint32(RetryCountBin, <<"restart">>)};
#{name => Name, maximum_retry_count => parse_uint32(RetryCountBin, <<"restart">>)};
[Name] ->
#'RestartPolicy'{name = Name, maximum_retry_count = 0}
#{name => Name, maximum_retry_count => 0}
end.
-spec build_healthcheck(undefined | map()) -> undefined | message_pb:'Healthcheck'().
-spec build_healthcheck(undefined | map()) -> undefined | map().
build_healthcheck(undefined) ->
undefined;
build_healthcheck(Healthcheck) when is_map(Healthcheck) ->
#'Healthcheck'{
test = maps:get(<<"test">>, Healthcheck, []),
interval_ns = parse_duration_ns(maps:get(<<"interval">>, Healthcheck, <<"0s">>), <<"healthcheck.interval">>),
timeout_ns = parse_duration_ns(maps:get(<<"timeout">>, Healthcheck, <<"0s">>), <<"healthcheck.timeout">>),
retries = maps:get(<<"retries">>, Healthcheck, 0)
#{
test => maps:get(<<"test">>, Healthcheck, []),
interval_ns => parse_duration_ns(maps:get(<<"interval">>, Healthcheck, <<"0s">>), <<"healthcheck.interval">>),
timeout_ns => parse_duration_ns(maps:get(<<"timeout">>, Healthcheck, <<"0s">>), <<"healthcheck.timeout">>),
retries => maps:get(<<"retries">>, Healthcheck, 0)
}.
-spec default_uint64(undefined | non_neg_integer()) -> non_neg_integer().
@ -340,46 +324,46 @@ volume_bind(HostPath, ContainerPath, true) when is_binary(HostPath), is_binary(C
volume_bind(HostPath, ContainerPath, false) when is_binary(HostPath), is_binary(ContainerPath) ->
<<HostPath/binary, ":", ContainerPath/binary>>.
-spec build_exposed_ports([binary()]) -> [message_pb:'DockerExposedPort'()].
-spec build_exposed_ports([binary()]) -> [map()].
build_exposed_ports(ExposeSpecs) when is_list(ExposeSpecs) ->
[build_exposed_port(ExposeSpec) || ExposeSpec <- ExposeSpecs].
-spec build_exposed_port(binary()) -> message_pb:'DockerExposedPort'().
-spec build_exposed_port(binary()) -> map().
build_exposed_port(ExposeSpec) when is_binary(ExposeSpec) ->
case binary:split(ExposeSpec, <<"/">>) of
[PortBin] ->
#'DockerExposedPort'{container_port = parse_uint32(PortBin, <<"expose">>), protocol = <<"tcp">>};
#{container_port => parse_uint32(PortBin, <<"expose">>), protocol => <<"tcp">>};
[PortBin, Protocol] ->
#'DockerExposedPort'{container_port = parse_uint32(PortBin, <<"expose">>), protocol = Protocol}
#{container_port => parse_uint32(PortBin, <<"expose">>), protocol => Protocol}
end.
-spec build_device_mappings([binary()]) -> [message_pb:'DeviceMapping'()].
-spec build_device_mappings([binary()]) -> [map()].
build_device_mappings(DeviceSpecs) when is_list(DeviceSpecs) ->
[build_device_mapping(DeviceSpec) || DeviceSpec <- DeviceSpecs].
-spec build_device_mapping(binary()) -> message_pb:'DeviceMapping'().
-spec build_device_mapping(binary()) -> map().
build_device_mapping(DeviceSpec) when is_binary(DeviceSpec) ->
case binary:split(DeviceSpec, <<":">>, [global]) of
[HostPath, ContainerPath] when HostPath =/= <<>>, ContainerPath =/= <<>> ->
#'DeviceMapping'{path_on_host = HostPath, path_in_container = ContainerPath, cgroup_permissions = <<"rwm">>};
#{path_on_host => HostPath, path_in_container => ContainerPath, cgroup_permissions => <<"rwm">>};
[HostPath, ContainerPath, Permissions] when HostPath =/= <<>>, ContainerPath =/= <<>>, Permissions =/= <<>> ->
#'DeviceMapping'{path_on_host = HostPath, path_in_container = ContainerPath, cgroup_permissions = Permissions};
#{path_on_host => HostPath, path_in_container => ContainerPath, cgroup_permissions => Permissions};
_ ->
throw({error, <<"invalid device mapping">>})
end.
-spec build_ulimits(map()) -> [message_pb:'Ulimit'()].
-spec build_ulimits(map()) -> [map()].
build_ulimits(Ulimits) when is_map(Ulimits) ->
[build_ulimit(Name, Value) || {Name, Value} <- maps:to_list(Ulimits)].
-spec build_ulimit(binary(), binary()) -> message_pb:'Ulimit'().
-spec build_ulimit(binary(), binary()) -> map().
build_ulimit(Name, Value) when is_binary(Name), is_binary(Value) ->
case binary:split(Value, <<":">>) of
[SoftBin, HardBin] ->
#'Ulimit'{name = Name, soft = parse_uint64(SoftBin, <<"ulimits.soft">>), hard = parse_uint64(HardBin, <<"ulimits.hard">>)};
#{name => Name, soft => parse_uint64(SoftBin, <<"ulimits.soft">>), hard => parse_uint64(HardBin, <<"ulimits.hard">>)};
[LimitBin] ->
Limit = parse_uint64(LimitBin, <<"ulimits.limit">>),
#'Ulimit'{name = Name, soft = Limit, hard = Limit}
#{name => Name, soft => Limit, hard => Limit}
end.
-spec build_tmpfs_options([binary()]) -> [{binary(), binary()}].

5657
src/protobuf/message_pb.erl
View File

File diff suppressed because it is too large Load Diff

81
src/transport/cloud_wire.erl Normal file
View File

@ -0,0 +1,81 @@
-module(cloud_wire).
-export([encode/1, decode/1]).
-spec encode(term()) -> binary().
encode(Term) ->
term_to_binary(validate(Term)).
-spec decode(binary()) -> term().
decode(Bin) when is_binary(Bin) ->
validate(binary_to_term(Bin, [safe])).
-spec validate(term()) -> term().
validate({request, PacketId, {auth_request, #{uuid := UUID, token := Token, timestamp := Timestamp} = Auth}})
when is_integer(PacketId), PacketId > 0, is_binary(UUID), is_binary(Token), is_integer(Timestamp) ->
{request, PacketId, {auth_request, Auth}};
validate({request, PacketId, {container_request, Request}})
when is_integer(PacketId), PacketId > 0 ->
{request, PacketId, {container_request, validate_container_request(Request)}};
validate({response, PacketId, {ok, Result}})
when is_integer(PacketId), PacketId > 0, is_binary(Result) ->
{response, PacketId, {ok, Result}};
validate({response, PacketId, {error, Code, Message}})
when is_integer(PacketId), PacketId > 0, is_integer(Code), is_binary(Message) ->
{response, PacketId, {error, Code, Message}};
validate({message, {data, #{route_key := RouteKey, metric := Metric} = Body}})
when is_binary(RouteKey), is_binary(Metric) ->
{message, {data, Body}};
validate({message, {task_event, #{task_id := TaskId, type := Type, stream := Stream} = Body}})
when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
{message, {task_event, Body}};
validate({message, {pub, #{topic := Topic, qos := Qos, content := Content} = Body}})
when is_binary(Topic), is_integer(Qos), is_binary(Content) ->
{message, {pub, Body}};
validate({message, {auth_control, Command}})
when Command =:= activate; Command =:= deactivate ->
{message, {auth_control, Command}};
validate(Term) ->
error({invalid_wire_term, Term}).
-spec validate_container_request(term()) -> map().
validate_container_request(#{action := list} = Request) ->
true = is_boolean(maps:get(all, Request, true)),
Request;
validate_container_request(#{action := deploy, task_id := TaskId, params := Params} = Request)
when is_integer(TaskId), is_map(Params) ->
true = is_binary(maps:get(container_name, Params, <<>>)),
true = is_binary(maps:get(container_dir, Params, <<>>)),
true = is_map(maps:get(create, Params, #{})),
Request;
validate_container_request(#{action := start, target := Target} = Request) ->
validate_target(Target),
Request;
validate_container_request(#{action := stop, target := Target, timeout_seconds := TimeoutSeconds} = Request)
when is_integer(TimeoutSeconds), TimeoutSeconds >= 0 ->
validate_target(Target),
Request;
validate_container_request(#{action := kill, target := Target, signal := Signal} = Request)
when is_binary(Signal) ->
validate_target(Target),
Request;
validate_container_request(#{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes} = Request)
when is_boolean(Force), is_boolean(RemoveVolumes) ->
validate_target(Target),
Request;
validate_container_request(#{action := config, target := Target, config := Config} = Request)
when is_binary(Config) ->
validate_target(Target),
Request;
validate_container_request(Request) ->
error({invalid_container_request, Request}).
-spec validate_target(term()) -> ok.
validate_target(#{name := Name, id := Id}) when is_binary(Name), is_binary(Id) ->
ok;
validate_target(#{name := Name}) when is_binary(Name) ->
ok;
validate_target(#{id := Id}) when is_binary(Id) ->
ok;
validate_target(Target) ->
error({invalid_container_target, Target}).

133
src/transport/tcp/ssl_channel.erl
View File

@ -8,8 +8,6 @@
%%%-------------------------------------------------------------------
-module(ssl_channel).
-author("licheng5").
-include("protocol.hrl").
-include("message_pb.hrl").
-behaviour(ranch_protocol).
-define(MAX_PACKET_ID, 16#FFFFFFFF).
@ -48,20 +46,17 @@ pub(Pid, Topic, Qos, Content) when is_pid(Pid), is_binary(Topic), is_integer(Qos
gen_server:cast(Pid, {pub, Topic, Qos, Content}).
%% 向通道中写入消息
-spec command(Pid :: pid(), Command :: message_pb:'CastFrame.Command'()) -> no_return().
command(Pid, Command) when is_pid(Pid) ->
-spec command(Pid :: pid(), Command :: activate | deactivate) -> no_return().
command(Pid, Command) when is_pid(Pid), (Command =:= activate orelse Command =:= deactivate) ->
gen_server:cast(Pid, {command, Command}).
-spec activate(Pid :: pid(), Auth :: boolean()) -> no_return().
activate(Pid, Auth) when is_pid(Pid), is_boolean(Auth) ->
Cmd = case Auth of true -> 'ACTIVATE'; false -> 'DEACTIVATE' end,
Command = #'CastFrame.Command'{
command = {auth, #'CastFrame.Command.Authorization'{cmd = Cmd}}
},
Command = case Auth of true -> activate; false -> deactivate end,
gen_server:cast(Pid, {command, Command}).
-spec container_call(Pid :: pid(), ReceiverPid :: pid(), Request :: message_pb:'ContainerRequest'()) -> Ref :: reference().
container_call(Pid, ReceiverPid, Request) when is_pid(Pid), is_pid(ReceiverPid), is_record(Request, 'ContainerRequest') ->
-spec container_call(Pid :: pid(), ReceiverPid :: pid(), Request :: map()) -> Ref :: reference().
container_call(Pid, ReceiverPid, Request) when is_pid(Pid), is_pid(ReceiverPid), is_map(Request) ->
Ref = make_ref(),
gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {container_request, Request}}),
Ref.
@ -105,28 +100,21 @@ handle_call(_Request, _From, State) ->
%% 发送消息, 基于pub/sub机制
handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) ->
Encoded = message_pb:encode_msg(#'CastFrame'{
body = {pub, #'CastFrame.Pub'{topic = Topic, qos = Qos, content = Content}}
}),
Transport:send(Socket, <<?FRAME_CAST, Encoded/binary>>),
Packet = cloud_wire:encode({message, {pub, #{topic => Topic, qos => Qos, content => Content}}}),
Transport:send(Socket, Packet),
{noreply, State};
%% 发送Command消息
handle_cast({command, Command}, State = #state{transport = Transport, socket = Socket}) ->
Encoded = message_pb:encode_msg(#'CastFrame'{
body = {command, Command}
}),
Transport:send(Socket, <<?FRAME_CAST, Encoded/binary>>),
Packet = cloud_wire:encode({message, {auth_control, Command}}),
Transport:send(Socket, Packet),
{noreply, State};
%% 推送需要响应的请求
handle_cast({request_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
Encoded = message_pb:encode_msg(#'RequestFrame'{
packet_id = PacketId,
body = Body
}),
Packet = cloud_wire:encode({request, PacketId, Body}),
TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {request_timeout, PacketId}),
Transport:send(Socket, <<?FRAME_REQUEST, Encoded/binary>>),
Transport:send(Socket, Packet),
RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef},
{noreply, State#state{
@ -152,24 +140,15 @@ handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, h
logger:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]),
{stop, Reason, State};
handle_info({ssl, Socket, <<?FRAME_REQUEST, FrameBin/binary>>},
State = #state{transport = Transport, socket = Socket}) ->
RequestFrame = message_pb:decode_msg(FrameBin, 'RequestFrame'),
handle_request_frame(RequestFrame, Transport, Socket, State);
handle_info({ssl, Socket, <<?FRAME_CAST, FrameBin/binary>>},
State = #state{socket = Socket, host_pid = HostPid}) ->
CastFrame = message_pb:decode_msg(FrameBin, 'CastFrame'),
handle_cast_frame(CastFrame, HostPid, State);
handle_info({ssl, Socket, <<?FRAME_REPLY, FrameBin/binary>>},
State = #state{socket = Socket, inflight = Inflight}) ->
ReplyFrame = message_pb:decode_msg(FrameBin, 'ReplyFrame'),
handle_reply_frame(ReplyFrame, Inflight, State);
handle_info({ssl, Socket, Frame}, State = #state{socket = Socket}) ->
logger:warning("[ssl_channel] get unknown ssl message: ~p, state: ~p", [Frame, State]),
{noreply, State};
handle_info({ssl, Socket, PacketBin}, State = #state{transport = Transport, socket = Socket, host_pid = HostPid, inflight = Inflight}) ->
case cloud_wire:decode(PacketBin) of
{request, PacketId, Body} ->
handle_request_frame(PacketId, Body, Transport, Socket, State);
{message, Body} ->
handle_message_frame(Body, HostPid, State);
{response, PacketId, Response} ->
handle_response_frame(PacketId, Response, Inflight, State)
end;
handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) ->
logger:notice("[ssl_channel] ssl socket closed"),
@ -217,9 +196,9 @@ inc_packet_id(?MAX_PACKET_ID) ->
inc_packet_id(PacketId) when PacketId > 0, PacketId < ?MAX_PACKET_ID ->
PacketId + 1.
-spec handle_request_frame(message_pb:'RequestFrame'(), module(), any(), #state{}) -> {noreply, #state{}} | {stop, #state{}}.
handle_request_frame(#'RequestFrame'{packet_id = PacketId,
body = {auth_request, #'AuthRequest'{uuid = UUID, token = Token, timestamp = Timestamp}}},
-spec handle_request_frame(non_neg_integer(), tuple(), module(), any(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}.
handle_request_frame(PacketId,
{auth_request, #{uuid := UUID, token := Token, timestamp := Timestamp}},
Transport, Socket, State) ->
logger:debug("[ws_channel] auth uuid: ~p", [UUID]),
@ -235,54 +214,53 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId,
case iot_host:attach_channel(HostPid, self()) of
ok ->
erlang:monitor(process, HostPid),
send_reply_frame(Transport, Socket, PacketId, {result, <<"ok">>}),
send_reply_frame(Transport, Socket, PacketId, {ok, <<"ok">>}),
{noreply, State#state{uuid = UUID, host_pid = HostPid}};
{denied, Reason} when is_binary(Reason) ->
erlang:monitor(process, HostPid),
send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyFrame.Error'{code = 1, message = Reason}}),
send_reply_frame(Transport, Socket, PacketId, {error, 1, Reason}),
logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
{noreply, State#state{uuid = UUID, host_pid = HostPid}};
{error, Reason} when is_binary(Reason) ->
send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyFrame.Error'{code = 2, message = Reason}}),
send_reply_frame(Transport, Socket, PacketId, {error, 2, Reason}),
logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
{stop, Reason, State}
end
end;
{error, Reason} ->
send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyFrame.Error'{code = 2, message = Reason}}),
send_reply_frame(Transport, Socket, PacketId, {error, 2, Reason}),
logger:warning("[ws_channel] uuid: ~p, token: ~p, auth failed, reason: ~p", [UUID, Token, Reason]),
{stop, Reason, State}
end;
handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {container_request, ContainerRequest}}, _Transport, _Socket, State) ->
handle_request_frame(PacketId, {container_request, ContainerRequest}, _Transport, _Socket, State) ->
logger:warning("[ws_channel] unsupported request message type: container_request, packet_id: ~p, request: ~p", [PacketId, ContainerRequest]),
{stop, normal, State};
handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = undefined}, _Transport, _Socket, State) ->
logger:warning("[ws_channel] empty request frame, packet_id: ~p", [PacketId]),
handle_request_frame(PacketId, Body, _Transport, _Socket, State) ->
logger:warning("[ws_channel] unsupported request body, packet_id: ~p, body: ~p", [PacketId, Body]),
{stop, normal, State}.
-spec handle_cast_frame(message_pb:'CastFrame'(), undefined | pid(), #state{}) ->
-spec handle_message_frame(tuple(), undefined | pid(), #state{}) ->
{noreply, #state{}}.
handle_cast_frame(#'CastFrame'{body = {data, Data}}, HostPid, State) when is_pid(HostPid) ->
iot_host:handle(HostPid, decode_data_frame(Data)),
handle_message_frame({data, #{route_key := RouteKey, metric := Metric}}, HostPid, State) when is_pid(HostPid) ->
iot_host:handle(HostPid, {data, RouteKey, Metric}),
{noreply, State};
handle_cast_frame(#'CastFrame'{body = {task_event, CastMessage}}, HostPid, State) when is_pid(HostPid) ->
handle_event_stream_frame(CastMessage),
handle_message_frame({task_event, Event}, HostPid, State) when is_pid(HostPid) ->
handle_event_stream_frame(Event),
{noreply, State};
handle_cast_frame(#'CastFrame'{body = Body}, _HostPid, State) ->
logger:warning("[ssl_channel] unsupported cast message type: command, body: ~p", [Body]),
handle_message_frame(Body, _HostPid, State) ->
logger:warning("[ssl_channel] unsupported message body: ~p", [Body]),
{noreply, State}.
-spec handle_event_stream_frame(message_pb:'CastFrame.TaskEvent'()) -> any().
handle_event_stream_frame(#'CastFrame.TaskEvent'{task_id = TaskId, type = Type0, stream = Reason0}) when Type0 =:= <<"close">> ->
iot_event_stream_observer:stream_close(TaskId, iolist_to_binary(Reason0));
handle_event_stream_frame(#'CastFrame.TaskEvent'{task_id = TaskId, type = Type, stream = Stream}) ->
-spec handle_event_stream_frame(map()) -> any().
handle_event_stream_frame(#{task_id := TaskId, type := <<"close">>, stream := Reason}) ->
iot_event_stream_observer:stream_close(TaskId, Reason);
handle_event_stream_frame(#{task_id := TaskId, type := Type, stream := Stream}) ->
logger:debug("[ssl_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]),
iot_event_stream_observer:stream_data(TaskId, Type, Stream).
-spec handle_reply_frame(message_pb:'ReplyFrame'(), map(), #state{}) ->
-spec handle_response_frame(non_neg_integer(), tuple(), map(), #state{}) ->
{noreply, #state{}}.
handle_reply_frame(#'ReplyFrame'{packet_id = PacketId, reply = Reply}, Inflight, State)
when PacketId > 0 ->
handle_response_frame(PacketId, Reply, Inflight, State) when PacketId > 0 ->
case maps:take(PacketId, Inflight) of
error ->
{noreply, State};
@ -296,28 +274,21 @@ handle_reply_frame(#'ReplyFrame'{packet_id = PacketId, reply = Reply}, Inflight,
end,
{noreply, State#state{inflight = NInflight}}
end;
handle_reply_frame(#'ReplyFrame'{packet_id = PacketId, reply = undefined}, _Inflight, State) ->
logger:warning("[ws_channel] empty reply frame, packet_id: ~p", [PacketId]),
{noreply, State};
handle_reply_frame(#'ReplyFrame'{packet_id = PacketId, reply = Reply}, _Inflight, State) ->
logger:warning("[ws_channel] unexpected reply frame, packet_id: ~p, reply: ~p", [PacketId, Reply]),
handle_response_frame(PacketId, Reply, _Inflight, State) ->
logger:warning("[ws_channel] unexpected response frame, packet_id: ~p, reply: ~p", [PacketId, Reply]),
{noreply, State}.
-spec send_reply_frame(module(), any(), non_neg_integer(), tuple()) -> any().
send_reply_frame(Transport, Socket, PacketId, Reply) ->
Encoded = message_pb:encode_msg(#'ReplyFrame'{packet_id = PacketId, reply = Reply}),
Transport:send(Socket, <<?FRAME_REPLY, Encoded/binary>>).
Packet = cloud_wire:encode({response, PacketId, Reply}),
Transport:send(Socket, Packet).
-spec decode_data_frame(message_pb:'CastFrame.Data'()) -> {data, binary(), binary()}.
decode_data_frame(#'CastFrame.Data'{route_key = RouteKey0, metric = Metric0}) ->
{data, iolist_to_binary(RouteKey0), iolist_to_binary(Metric0)}.
-spec decode_reply({result, iodata()} | {error, message_pb:'ReplyFrame.Error'()} | undefined) ->
-spec decode_reply({ok, binary()} | {error, integer(), binary()}) ->
{ok, binary()} | {error, integer(), binary()} | undefined.
decode_reply({result, ResultBin}) ->
{ok, iolist_to_binary(ResultBin)};
decode_reply({error, #'ReplyFrame.Error'{code = Code, message = Message}}) ->
{error, Code, unicode:characters_to_binary(Message)};
decode_reply({ok, ResultBin}) ->
{ok, ResultBin};
decode_reply({error, Code, Message}) ->
{error, Code, Message};
decode_reply(undefined) ->
undefined.