docs/container_command_docker_json.md

@@ -15,41 +15,31 @@
 `iot` 通过 TLS 长连接向 `efka` 下发容器命令：
 
 ```erlang
-{<<"command">>, Ref, {<<"container">>, CommandMap}}
+{command, Ref, {container, CommandMap}}
 ```
 
 `efka` 执行后回复：
 
 ```erlang
-{<<"command_response">>, Ref, {<<"container">>, Reply}}
+{command_response, Ref, {container, Reply}}
 ```
 
 `Reply` 取值：
 
 ```erlang
-<<"ok">>
-{<<"ok">>, Result}
-{<<"error">>, Reason}
+ok
+{ok, Result}
+{error, Reason}
 ```
 
-`Ref` 是 `crypto:strong_rand_bytes(16)` 生成的 16 字节 binary。网络帧只使用 `binary_to_term(PacketBin, [safe])` 可解码的 safe term；协议 label、业务 label、map key 和 action 都使用 binary。
-
-只有 `efka_client` 处于 `activated` 状态时，容器命令才会正常执行；处于非 activated 状态时会返回错误。
+只有 `efka_client` 处于 `activated` 状态时，容器命令才会正常执行；处于 `restricted` 或其他状态时会返回错误。
 
 ## 2. 容器命令 map
 
-`efka_client` 收到网络协议里的 binary-key `CommandMap` 后，直接按 binary key 和 binary action 做函数参数匹配，不再做整包 atom-key 转换。
-
-- command 顶层、`target`、deploy `params`、`create` 中间结构都使用 binary key。
-- `<<"action">>` 的取值使用 binary：`<<"list">>`、`<<"deploy">>`、`<<"start">>`、`<<"stop">>`、`<<"kill">>`、`<<"remove">>`、`<<"config">>`。
-- `<<"__undefined__">>` 只在 Docker 参数构造读取可选字段时按需当成 `undefined` 处理。
-
-下面各小节描述的是 `efka` 直接接收和处理的 map 格式。
-
 ### list
 
 ```erlang
-#{<<"action">> => <<"list">>}
+#{action => list}
 ```
 
 执行：
@@ -68,9 +58,9 @@ GET /containers/json?all=true
 
 ```erlang
 #{
-    <<"action">> => <<"deploy">>,
-    <<"task_id">> => TaskId,
-    <<"params">> => Params
+    action => deploy,
+    task_id => TaskId,
+    params => Params
 }
 ```
 
@@ -86,8 +76,8 @@ docker_deploy_manager:deploy(TaskId, Params)
 
 ```erlang
 #{
-    <<"action">> => <<"start">>,
-    <<"target">> => Target
+    action => start,
+    target => Target
 }
 ```
 
@@ -107,9 +97,9 @@ POST /containers/{name_or_id}/start
 
 ```erlang
 #{
-    <<"action">> => <<"stop">>,
-    <<"target">> => Target,
-    <<"timeout_seconds">> => TimeoutSeconds
+    action => stop,
+    target => Target,
+    timeout_seconds => TimeoutSeconds
 }
 ```
 
@@ -129,9 +119,9 @@ POST /containers/{name_or_id}/stop?t={TimeoutSeconds}
 
 ```erlang
 #{
-    <<"action">> => <<"kill">>,
-    <<"target">> => Target,
-    <<"signal">> => Signal
+    action => kill,
+    target => Target,
+    signal => Signal
 }
 ```
 
@@ -145,10 +135,10 @@ docker_commands:kill_container(ContainerNameOrId, Signal)
 
 ```erlang
 #{
-    <<"action">> => <<"remove">>,
-    <<"target">> => Target,
-    <<"force">> => Force,
-    <<"remove_volumes">> => RemoveVolumes
+    action => remove,
+    target => Target,
+    force => Force,
+    remove_volumes => RemoveVolumes
 }
 ```
 
@@ -162,9 +152,9 @@ docker_commands:remove_container(ContainerNameOrId, Force, RemoveVolumes)
 
 ```erlang
 #{
-    <<"action">> => <<"config">>,
-    <<"target">> => Target,
-    <<"config">> => Config
+    action => config,
+    target => Target,
+    config => Config
 }
 ```
 
@@ -182,8 +172,8 @@ docker_helper:update_container_config(ContainerNameOrId, iolist_to_binary(Config
 
 ```erlang
 #{
-    <<"name">> => ContainerName,
-    <<"id">> => ContainerId
+    name => ContainerName,
+    id => ContainerId
 }
 ```
 
@@ -199,8 +189,8 @@ docker_helper:update_container_config(ContainerNameOrId, iolist_to_binary(Config
 
 ```erlang
 #{
-    <<"container_name">> => ContainerName,
-    <<"create">> => Create
+    container_name => ContainerName,
+    create => Create
 }
 ```
 
@@ -225,9 +215,9 @@ efka.root_dir/container_name/
 
 ```erlang
 #{
-    <<"config">> => ContainerConfig,
-    <<"host_config">> => HostConfig,
-    <<"networking_config">> => NetworkingConfig
+    config => ContainerConfig,
+    host_config => HostConfig,
+    networking_config => NetworkingConfig
 }
 ```
 
@@ -350,8 +340,8 @@ Body = iolist_to_binary(json:encode(Options))
 
 ```erlang
 [
-    #{<<"container_port">> => 80, <<"protocol">> => <<"tcp">>},
-    #{<<"container_port">> => 53, <<"protocol">> => <<"udp">>}
+    #{container_port => 80, protocol => <<"tcp">>},
+    #{container_port => 53, protocol => <<"udp">>}
 ]
 ```
 
@@ -379,10 +369,10 @@ Body = iolist_to_binary(json:encode(Options))
 
 ```erlang
 #{
-    <<"test">> => [<<"CMD-SHELL">>, <<"curl -f http://localhost || exit 1">>],
-    <<"interval_ns">> => 30000000000,
-    <<"timeout_ns">> => 10000000000,
-    <<"retries">> => 3
+    test => [<<"CMD-SHELL">>, <<"curl -f http://localhost || exit 1">>],
+    interval_ns => 30000000000,
+    timeout_ns => 10000000000,
+    retries => 3
 }
 ```
 
@@ -450,7 +440,7 @@ efka 自动补丁后输出：
 输入：
 
 ```erlang
-#{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0}
+#{name => <<"always">>, maximum_retry_count => 0}
 ```
 
 输出：
@@ -468,7 +458,7 @@ efka 自动补丁后输出：
 输入：
 
 ```erlang
-#{<<"name">> => <<"on-failure">>, <<"maximum_retry_count">> => 3}
+#{name => <<"on-failure">>, maximum_retry_count => 3}
 ```
 
 输出：
@@ -491,9 +481,9 @@ efka 自动补丁后输出：
 ```erlang
 [
     #{
-        <<"path_on_host">> => <<"/dev/ttyUSB0">>,
-        <<"path_in_container">> => <<"/dev/ttyUSB0">>,
-        <<"cgroup_permissions">> => <<"rwm">>
+        path_on_host => <<"/dev/ttyUSB0">>,
+        path_in_container => <<"/dev/ttyUSB0">>,
+        cgroup_permissions => <<"rwm">>
     }
 ]
 ```
@@ -522,10 +512,10 @@ efka 自动补丁后输出：
 
 ```erlang
 #{
-    <<"memory">> => 536870912,
-    <<"memory_reservation">> => 268435456,
-    <<"nano_cpus">> => 1500000000,
-    <<"cpu_shares">> => 512
+    memory => 536870912,
+    memory_reservation => 268435456,
+    nano_cpus => 1500000000,
+    cpu_shares => 512
 }
 ```
 
@@ -550,7 +540,7 @@ efka 自动补丁后输出：
 
 ```erlang
 [
-    #{<<"name">> => <<"nofile">>, <<"soft">> => 1024, <<"hard">> => 2048}
+    #{name => <<"nofile">>, soft => 1024, hard => 2048}
 ]
 ```
 
@@ -638,9 +628,9 @@ efka 自动补丁后输出：
 
 ```erlang
 #{
-    <<"endpoints">> => [
-        #{<<"name">> => <<"bridge">>},
-        #{<<"name">> => <<"mynet">>}
+    endpoints => [
+        #{name => <<"bridge">>},
+        #{name => <<"mynet">>}
     ]
 }
 ```
@@ -671,26 +661,26 @@ efka 自动补丁后输出：
 收到的 deploy command：
 
 ```erlang
-{<<"command">>, Ref, {<<"container">>, #{
-    <<"action">> => <<"deploy">>,
-    <<"task_id">> => 1001,
-    <<"params">> => #{
-        <<"container_name">> => <<"my_nginx">>,
-        <<"create">> => #{
-            <<"config">> => #{
-                <<"image">> => <<"docker.io/library/nginx:latest">>,
-                <<"cmd">> => [<<"nginx">>, <<"-g">>, <<"daemon off;">>],
-                <<"env">> => [<<"ENV=prod">>],
-                <<"volumes">> => [<<"/data">>],
-                <<"exposed_ports">> => [#{<<"container_port">> => 80, <<"protocol">> => <<"tcp">>}]
+{command, Ref, {container, #{
+    action => deploy,
+    task_id => 1001,
+    params => #{
+        container_name => <<"my_nginx">>,
+        create => #{
+            config => #{
+                image => <<"docker.io/library/nginx:latest">>,
+                cmd => [<<"nginx">>, <<"-g">>, <<"daemon off;">>],
+                env => [<<"ENV=prod">>],
+                volumes => [<<"/data">>],
+                exposed_ports => [#{container_port => 80, protocol => <<"tcp">>}]
             },
-            <<"host_config">> => #{
-                <<"binds">> => [<<"/host/data:/data">>],
-                <<"restart_policy">> => #{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0},
-                <<"memory">> => 536870912
+            host_config => #{
+                binds => [<<"/host/data:/data">>],
+                restart_policy => #{name => <<"always">>, maximum_retry_count => 0},
+                memory => 536870912
             },
-            <<"networking_config">> => #{
-                <<"endpoints">> => [#{<<"name">> => <<"bridge">>}]
+            networking_config => #{
+                endpoints => [#{name => <<"bridge">>}]
             }
         }
     }

docs/efka_iot_protocol.md

@@ -2,99 +2,95 @@
 
 本文档描述 `efka` 与 `iot` 之间的 TLS 长连接协议。当前协议由 Erlang term 直接序列化，发送端使用 `term_to_binary/1`，接收端使用 `binary_to_term(PacketBin, [safe])`。
 
-协议帧只使用 safe external term：顶层 label、业务 label、map key 使用 binary；`Ref` 使用 `crypto:strong_rand_bytes(16)` 生成，是 16 字节 binary。网络协议里不发送 Erlang `reference()`，也不依赖动态创建 atom。
-
 ## 传输层
 
 - `efka` 作为 TLS client 连接 `iot`。
 - `iot` 作为 TLS server 接收多个 `efka` 连接，一个连接对应一个 `ssl_channel` 进程。
 - socket 使用 `{packet, 4}`，每个 Erlang term binary 作为一个完整包发送。
-- `Ref` 使用 `crypto:strong_rand_bytes(16)` 生成，只在当前连接的 inflight 表内匹配。
+- `Ref` 使用 `make_ref()` 生成，只在当前连接的 inflight 表内匹配。
 
 ## 顶层帧
 
 协议顶层 tuple 用来表达交互语义：
 
 ```erlang
-{<<"request">>, Ref, Body}
-{<<"response">>, Ref, Reply}
-{<<"command">>, Ref, {Domain, Payload}}
-{<<"command_response">>, Ref, {Domain, Reply}}
-{<<"message">>, Body}
+{request, Ref, Body}
+{response, Ref, Reply}
+{command, Ref, {Domain, Payload}}
+{command_response, Ref, {Domain, Reply}}
+{message, Body}
 ```
 
 语义说明：
 
 | 帧 | 方向 | 语义 |
 | --- | --- | --- |
-| `{<<"request">>, Ref, Body}` | efka -> iot | efka 发起请求，需要 iot 回复 |
-| `{<<"response">>, Ref, Reply}` | iot -> efka | iot 对 efka request 的回复 |
-| `{<<"command">>, Ref, {Domain, Payload}}` | iot -> efka | iot 下发命令，需要 efka 回复 |
-| `{<<"command_response">>, Ref, {Domain, Reply}}` | efka -> iot | efka 对 iot command 的回复 |
-| `{<<"message">>, Body}` | 双向 | 异步消息，不要求回复 |
+| `{request, Ref, Body}` | efka -> iot | efka 发起请求，需要 iot 回复 |
+| `{response, Ref, Reply}` | iot -> efka | iot 对 efka request 的回复 |
+| `{command, Ref, {Domain, Payload}}` | iot -> efka | iot 下发命令，需要 efka 回复 |
+| `{command_response, Ref, {Domain, Reply}}` | efka -> iot | efka 对 iot command 的回复 |
+| `{message, Body}` | 双向 | 异步消息，不要求回复 |
 
 `command` 和 `command_response` 的 `Domain` 表示业务域，目前支持：
 
-- `<<"container">>`
+- `container`
 
 ## 鉴权请求
 
 初始连接由 `efka` 发起鉴权 request。每条 TLS 连接只允许一次鉴权；`iot` 侧鉴权成功后会在 `ssl_channel` 标记该连接已鉴权，如果同一连接再次发送 `auth_request`，`iot` 会直接关闭连接。
 
 ```erlang
-{<<"request">>, Ref, {<<"auth_request">>, #{
-    <<"uuid">> => UUID,
-    <<"token">> => Token,
-    <<"timestamp">> => Timestamp
+{request, Ref, {auth_request, #{
+    uuid => UUID,
+    token => Token,
+    timestamp => Timestamp
 }}}
 ```
 
 `iot` 回复：
 
 ```erlang
-{<<"response">>, Ref, {<<"auth_response">>, <<"ok">>}}
-{<<"response">>, Ref, {<<"auth_response">>, {<<"error">>, {<<"failed">>, Reason}}}}
+{response, Ref, {auth_response, ok}}
+{response, Ref, {auth_response, {error, {failed, Reason}}}}
 ```
 
 处理语义：
 
-- `<<"ok">>`：`efka` 进入 `activated` 状态。
-- `{<<"error">>, {<<"failed">>, Reason}}`：鉴权失败，`iot` 返回失败响应后关闭连接；`efka` 进入重连流程。
+- `ok`：`efka` 进入 `activated` 状态。
+- `{error, {failed, Reason}}`：鉴权失败，`iot` 返回失败响应后关闭连接；`efka` 进入重连流程。
 
 ## 授权控制
 
 `/host/activate` 只修改 `iot` 本地和持久化的 host 授权状态，不再向 `efka` 下发 auth command。`efka` 可以继续保持连接并发送数据，是否处理这些数据由 `iot_host` 当前状态决定。
 
-因此当前协议没有 `{<<"command">>, Ref, {<<"auth">>, ...}}` 和 `{<<"command_response">>, Ref, {<<"auth">>, ...}}`。授权关闭时，`iot_host` 保持 channel 在线，但不处理上报数据；授权重新打开后，已在线的 channel 可以继续使用。
+因此当前协议没有 `{command, Ref, {auth, ...}}` 和 `{command_response, Ref, {auth, ...}}`。授权关闭时，`iot_host` 保持 channel 在线，但不处理上报数据；授权重新打开后，已在线的 channel 可以继续使用。
 
 ## 容器管理命令
 
 `iot` 对 `efka` 的容器管理使用 command 语义：
 
 ```erlang
-{<<"command">>, Ref, {<<"container">>, CommandMap}}
+{command, Ref, {container, CommandMap}}
 ```
 
 `efka` 回复：
 
 ```erlang
-{<<"command_response">>, Ref, {<<"container">>, Reply}}
+{command_response, Ref, {container, Reply}}
 ```
 
 `Reply` 取值：
 
 ```erlang
-<<"ok">>
-{<<"ok">>, Result}
-{<<"error">>, Reason}
+ok
+{ok, Result}
+{error, Reason}
 ```
 
-`CommandMap` 使用 binary key 和 binary action；`efka` 接收后直接按 binary key/action 匹配，Docker 参数链路继续使用 binary-key map，不再转换成 atom-key map。
-
 ### list
 
 ```erlang
-#{<<"action">> => <<"list">>}
+#{action => list}
 ```
 
 返回当前 `efka` 主机上的容器列表。
@@ -103,9 +99,9 @@
 
 ```erlang
 #{
-    <<"action">> => <<"deploy">>,
-    <<"task_id">> => TaskId,
-    <<"params">> => Params
+    action => deploy,
+    task_id => TaskId,
+    params => Params
 }
 ```
 
@@ -115,8 +111,8 @@
 
 ```erlang
 #{
-    <<"action">> => <<"start">>,
-    <<"target">> => Target
+    action => start,
+    target => Target
 }
 ```
 
@@ -124,9 +120,9 @@
 
 ```erlang
 #{
-    <<"action">> => <<"stop">>,
-    <<"target">> => Target,
-    <<"timeout_seconds">> => TimeoutSeconds
+    action => stop,
+    target => Target,
+    timeout_seconds => TimeoutSeconds
 }
 ```
 
@@ -134,9 +130,9 @@
 
 ```erlang
 #{
-    <<"action">> => <<"kill">>,
-    <<"target">> => Target,
-    <<"signal">> => Signal
+    action => kill,
+    target => Target,
+    signal => Signal
 }
 ```
 
@@ -144,10 +140,10 @@
 
 ```erlang
 #{
-    <<"action">> => <<"remove">>,
-    <<"target">> => Target,
-    <<"force">> => Force,
-    <<"remove_volumes">> => RemoveVolumes
+    action => remove,
+    target => Target,
+    force => Force,
+    remove_volumes => RemoveVolumes
 }
 ```
 
@@ -155,9 +151,9 @@
 
 ```erlang
 #{
-    <<"action">> => <<"config">>,
-    <<"target">> => Target,
-    <<"config">> => Config
+    action => config,
+    target => Target,
+    config => Config
 }
 ```
 
@@ -169,8 +165,8 @@
 
 ```erlang
 #{
-    <<"name">> => ContainerName,
-    <<"id">> => ContainerId
+    name => ContainerName,
+    id => ContainerId
 }
 ```
 
@@ -183,9 +179,9 @@
 ### efka -> iot: data
 
 ```erlang
-{<<"message">>, {<<"data">>, #{
-    <<"route_key">> => RouteKey,
-    <<"metric">> => Metric
+{message, {data, #{
+    route_key => RouteKey,
+    metric => Metric
 }}}
 ```
 
@@ -194,20 +190,20 @@
 ### efka -> iot: task_event
 
 ```erlang
-{<<"message">>, {<<"task_event">>, #{
-    <<"task_id">> => TaskId,
-    <<"type">> => Type,
-    <<"stream">> => Stream
+{message, {task_event, #{
+    task_id => TaskId,
+    type => Type,
+    stream => Stream
 }}}
 ```
 
 任务事件流关闭时：
 
 ```erlang
-{<<"message">>, {<<"task_event">>, #{
-    <<"task_id">> => TaskId,
-    <<"type">> => <<"close">>,
-    <<"stream">> => Reason
+{message, {task_event, #{
+    task_id => TaskId,
+    type => <<"close">>,
+    stream => Reason
 }}}
 ```
 
@@ -222,10 +218,10 @@ GET /event_stream?uuid=<host_uuid>&task_id=<task_id>
 ### iot -> efka: pub
 
 ```erlang
-{<<"message">>, {<<"pub">>, #{
-    <<"topic">> => Topic,
-    <<"qos">> => Qos,
-    <<"content">> => Content
+{message, {pub, #{
+    topic => Topic,
+    qos => Qos,
+    content => Content
 }}}
 ```
 
@@ -252,6 +248,5 @@ GET /event_stream?uuid=<host_uuid>&task_id=<task_id>
 - 旧容器回复：`{response, Ref, {container_response, ...}}`
 - 旧授权控制：`{message, {auth_control, Command}}`
 - 已移除的 auth command：`{command, Ref, {auth, activate | deactivate}}`
-- 旧 Ref：Erlang `reference()`，例如 `make_ref()` 生成的值。
 
 如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。

src/docker/docker_commands.erl

@@ -35,7 +35,7 @@ check_image_exist(Image) when is_binary(Image) ->
 
 -spec create_container(ContainerDir :: string(), Params :: map()) ->
     {ok, ContainerId :: binary()} | {error, Reason :: any()}.
-create_container(ContainerDir, #{<<"container_name">> := ContainerName0, <<"create">> := CreateOpts})
+create_container(ContainerDir, #{container_name := ContainerName0, create := CreateOpts})
     when is_list(ContainerDir), is_binary(ContainerName0), is_map(CreateOpts) ->
     Url = lists:flatten(io_lib:format("/containers/create?name=~s", [binary_to_list(ContainerName0)])),
     Options = docker_container_builder:build_options(ContainerName0, ContainerDir, CreateOpts),

src/docker/docker_container_builder.erl

@@ -23,30 +23,30 @@ patch_create_options(ContainerName, ConfigFile, undefined) ->
     patch_create_options(ContainerName, ConfigFile, #{});
 patch_create_options(ContainerName, ConfigFile, Create0)
   when is_binary(ContainerName), is_binary(ConfigFile), is_map(Create0) ->
-    Config = patch_container_config(ContainerName, ensure_container_config(field(Create0, <<"config">>, undefined))),
-    HostConfig = patch_host_config(ConfigFile, ensure_host_config(field(Create0, <<"host_config">>, undefined))),
+    Config = patch_container_config(ContainerName, ensure_container_config(maps:get(config, Create0, undefined))),
+    HostConfig = patch_host_config(ConfigFile, ensure_host_config(maps:get(host_config, Create0, undefined))),
     Create0#{
-        <<"config">> => Config,
-        <<"host_config">> => HostConfig
+        config => Config,
+        host_config => HostConfig
     }.
 
 -spec patch_container_config(binary(), map()) -> map().
 patch_container_config(ContainerName, Config0) when is_binary(ContainerName), is_map(Config0) ->
-    Env0 = [to_binary(EnvItem) || EnvItem <- field(Config0, <<"env">>, [])],
-    Volumes0 = [to_binary(Volume) || Volume <- field(Config0, <<"volumes">>, [])],
+    Env0 = [to_binary(EnvItem) || EnvItem <- maps:get(env, Config0, [])],
+    Volumes0 = [to_binary(Volume) || Volume <- maps:get(volumes, Config0, [])],
     ConfigVolume = <<"/usr/local/etc/service.conf">>,
     Envs = add_unique_front([<<"CONTAINER_NAME=", ContainerName/binary>>], Env0),
     Volumes = add_unique_front([ConfigVolume], Volumes0),
     Config0#{
-        <<"env">> => Envs,
-        <<"volumes">> => Volumes
+        env => Envs,
+        volumes => Volumes
     }.
 
 -spec patch_host_config(binary(), map()) -> map().
 patch_host_config(ConfigFile, HostConfig0) when is_binary(ConfigFile), is_map(HostConfig0) ->
-    Binds0 = [to_binary(Bind) || Bind <- field(HostConfig0, <<"binds">>, [])],
+    Binds0 = [to_binary(Bind) || Bind <- maps:get(binds, HostConfig0, [])],
     ConfigBind = <<ConfigFile/binary, ":/usr/local/etc/service.conf">>,
-    HostConfig0#{<<"binds">> => add_unique_front([ConfigBind], Binds0)}.
+    HostConfig0#{binds => add_unique_front([ConfigBind], Binds0)}.
 
 -spec ensure_container_config(map() | undefined) -> map().
 ensure_container_config(undefined) ->
@@ -62,39 +62,39 @@ ensure_host_config(HostConfig) when is_map(HostConfig) ->
 
 -spec build_create_options(map()) -> map().
 build_create_options(Create0) when is_map(Create0) ->
-    Config = field(Create0, <<"config">>, #{}),
-    HostConfig = field(Create0, <<"host_config">>, #{}),
-    Endpoints = networking_config_endpoints(field(Create0, <<"networking_config">>, undefined)),
+    Config = maps:get(config, Create0, #{}),
+    HostConfig = maps:get(host_config, Create0, #{}),
+    Endpoints = networking_config_endpoints(maps:get(networking_config, Create0, undefined)),
     #{
-        <<"Image">> => to_binary(field(Config, <<"image">>, <<>>)),
-        <<"Cmd">> => [to_binary(CommandItem) || CommandItem <- field(Config, <<"cmd">>, [])],
-        <<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- field(Config, <<"entrypoint">>, [])],
-        <<"Env">> => [to_binary(EnvItem) || EnvItem <- field(Config, <<"env">>, [])],
-        <<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(field(Config, <<"labels">>, #{}))]),
-        <<"Volumes">> => build_volumes(field(Config, <<"volumes">>, [])),
-        <<"User">> => to_binary(field(Config, <<"user">>, <<>>)),
-        <<"WorkingDir">> => to_binary(field(Config, <<"working_dir">>, <<>>)),
-        <<"Hostname">> => to_binary(field(Config, <<"hostname">>, <<>>)),
-        <<"ExposedPorts">> => build_expose(field(Config, <<"exposed_ports">>, [])),
+        <<"Image">> => to_binary(maps:get(image, Config, <<>>)),
+        <<"Cmd">> => [to_binary(CommandItem) || CommandItem <- maps:get(cmd, Config, [])],
+        <<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- maps:get(entrypoint, Config, [])],
+        <<"Env">> => [to_binary(EnvItem) || EnvItem <- maps:get(env, Config, [])],
+        <<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(maps:get(labels, Config, #{}))]),
+        <<"Volumes">> => build_volumes(maps:get(volumes, Config, [])),
+        <<"User">> => to_binary(maps:get(user, Config, <<>>)),
+        <<"WorkingDir">> => to_binary(maps:get(working_dir, Config, <<>>)),
+        <<"Hostname">> => to_binary(maps:get(hostname, Config, <<>>)),
+        <<"ExposedPorts">> => build_expose(maps:get(exposed_ports, Config, [])),
         <<"NetworkingConfig">> => build_networking_config(Endpoints),
-        <<"Healthcheck">> => build_healthcheck(field(Config, <<"healthcheck">>, undefined)),
+        <<"Healthcheck">> => build_healthcheck(maps:get(healthcheck, Config, undefined)),
         <<"HostConfig">> => fold_merge([
-            build_binds(field(HostConfig, <<"binds">>, [])),
-            build_network_mode(field(HostConfig, <<"network_mode">>, <<>>)),
-            build_restart(field(HostConfig, <<"restart_policy">>, undefined)),
-            build_privileged(field(HostConfig, <<"privileged">>, false)),
-            build_cap_add_drop(field(HostConfig, <<"cap_add">>, []), field(HostConfig, <<"cap_drop">>, [])),
-            build_devices(field(HostConfig, <<"devices">>, [])),
+            build_binds(maps:get(binds, HostConfig, [])),
+            build_network_mode(maps:get(network_mode, HostConfig, <<>>)),
+            build_restart(maps:get(restart_policy, HostConfig, undefined)),
+            build_privileged(maps:get(privileged, HostConfig, false)),
+            build_cap_add_drop(maps:get(cap_add, HostConfig, []), maps:get(cap_drop, HostConfig, [])),
+            build_devices(maps:get(devices, HostConfig, [])),
             build_resources(
-                field(HostConfig, <<"memory">>, 0),
-                field(HostConfig, <<"memory_reservation">>, 0),
-                field(HostConfig, <<"nano_cpus">>, 0),
-                field(HostConfig, <<"cpu_shares">>, 0)
+                maps:get(memory, HostConfig, 0),
+                maps:get(memory_reservation, HostConfig, 0),
+                maps:get(nano_cpus, HostConfig, 0),
+                maps:get(cpu_shares, HostConfig, 0)
             ),
-            build_ulimits(field(HostConfig, <<"ulimits">>, [])),
-            build_tmpfs(field(HostConfig, <<"tmpfs">>, #{})),
-            build_sysctls(field(HostConfig, <<"sysctls">>, #{})),
-            build_extra_hosts(field(HostConfig, <<"extra_hosts">>, []))
+            build_ulimits(maps:get(ulimits, HostConfig, [])),
+            build_tmpfs(maps:get(tmpfs, HostConfig, #{})),
+            build_sysctls(maps:get(sysctls, HostConfig, #{})),
+            build_extra_hosts(maps:get(extra_hosts, HostConfig, []))
         ])
     }.
 
@@ -102,7 +102,7 @@ build_create_options(Create0) when is_map(Create0) ->
 networking_config_endpoints(undefined) ->
     [];
 networking_config_endpoints(NetworkingConfig) when is_map(NetworkingConfig) ->
-    field(NetworkingConfig, <<"endpoints">>, []).
+    maps:get(endpoints, NetworkingConfig, []).
 
 -spec fold_merge([map()]) -> map().
 fold_merge(List) ->
@@ -141,7 +141,7 @@ build_networking_config(Endpoints) when is_list(Endpoints) ->
         [] ->
             #{};
         _ ->
-            NetCfg = maps:from_list([{to_binary(Name), #{}} || #{<<"name">> := Name} <- Endpoints]),
+            NetCfg = maps:from_list([{to_binary(Name), #{}} || #{name := Name} <- Endpoints]),
             #{<<"EndpointsConfig">> => NetCfg}
     end.
 
@@ -156,10 +156,10 @@ build_healthcheck(undefined) ->
     #{};
 build_healthcheck(Healthcheck) when is_map(Healthcheck) ->
     #{
-        <<"Test">> => [to_binary(Item) || Item <- field(Healthcheck, <<"test">>, [])],
-        <<"Interval">> => field(Healthcheck, <<"interval_ns">>, 0),
-        <<"Timeout">> => field(Healthcheck, <<"timeout_ns">>, 0),
-        <<"Retries">> => field(Healthcheck, <<"retries">>, 0)
+        <<"Test">> => [to_binary(Item) || Item <- maps:get(test, Healthcheck, [])],
+        <<"Interval">> => maps:get(interval_ns, Healthcheck, 0),
+        <<"Timeout">> => maps:get(timeout_ns, Healthcheck, 0),
+        <<"Retries">> => maps:get(retries, Healthcheck, 0)
     }.
 
 -spec build_restart(map() | undefined) -> map().
@@ -167,9 +167,9 @@ build_restart(undefined) ->
     #{};
 build_restart(RestartPolicy0) when is_map(RestartPolicy0) ->
     RestartPolicy = #{
-        <<"Name">> => to_binary(field(RestartPolicy0, <<"name">>, <<>>))
+        <<"Name">> => to_binary(maps:get(name, RestartPolicy0, <<>>))
     },
-    case field(RestartPolicy0, <<"maximum_retry_count">>, 0) of
+    case maps:get(maximum_retry_count, RestartPolicy0, 0) of
         0 ->
             #{<<"RestartPolicy">> => RestartPolicy};
         RetryCount ->
@@ -210,9 +210,9 @@ build_devices(Devices) when is_list(Devices) ->
                 <<"PathInContainer">> => to_binary(ContainerPath),
                 <<"CgroupPermissions">> => device_permissions(Permissions)
             } || #{
-                <<"path_on_host">> := HostPath,
-                <<"path_in_container">> := ContainerPath,
-                <<"cgroup_permissions">> := Permissions
+                path_on_host := HostPath,
+                path_in_container := ContainerPath,
+                cgroup_permissions := Permissions
             } <- Devices],
             #{<<"Devices">> => DevObjs}
     end.
@@ -255,7 +255,7 @@ build_ulimits(Ulimits) when is_list(Ulimits) ->
                 <<"Name">> => to_binary(Name),
                 <<"Soft">> => Soft,
                 <<"Hard">> => Hard
-            } || #{<<"name">> := Name, <<"soft">> := Soft, <<"hard">> := Hard} <- Ulimits]}
+            } || #{name := Name, soft := Soft, hard := Hard} <- Ulimits]}
     end.
 
 -spec build_sysctls(map()) -> map().
@@ -286,7 +286,7 @@ build_extra_hosts(Hosts) when is_list(Hosts) ->
     end.
 
 -spec normalize_expose_port(map()) -> binary().
-normalize_expose_port(#{<<"container_port">> := Port, <<"protocol">> := Protocol}) ->
+normalize_expose_port(#{container_port := Port, protocol := Protocol}) ->
     PortBin = integer_to_binary(Port),
     ProtocolBin = to_binary(Protocol),
     case ProtocolBin of
@@ -316,16 +316,6 @@ add_unique_front([Item | Rest], List) ->
     end,
     add_unique_front(Rest, NList).
 
--spec field(map(), binary(), term()) -> term().
-field(Map, Key, Default) when is_map(Map), is_binary(Key) ->
-    safe_value(maps:get(Key, Map, Default)).
-
--spec safe_value(term()) -> term().
-safe_value(<<"__undefined__">>) ->
-    undefined;
-safe_value(Value) ->
-    Value.
-
 -spec to_binary(binary() | list() | atom() | any()) -> binary().
 to_binary(Value) when is_binary(Value) ->
     Value;

src/docker/docker_deploy_manager.erl

@@ -49,7 +49,7 @@ init([]) ->
 -spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}.
 handle_call({deploy, TaskId, Params}, _From, State = #state{root_dir = RootDir, task_map = TaskMap})
   when is_map(Params) ->
-    ContainerName = maps:get(<<"container_name">>, Params),
+    ContainerName = maps:get(container_name, Params),
     {ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName),
     {ok, {TaskPid, _Ref}} = docker_deployer:start_monitor(TaskId, ContainerDir, Params),
     logger:debug("[docker_deploy_manager] start deploy task_id: ~p, params: ~p", [TaskId, Params]),

src/docker/docker_deployer.erl

@@ -199,13 +199,13 @@ short_container_id(ContainerId) when is_binary(ContainerId) ->
     ContainerId.
 
 -spec deploy_container_name(map()) -> binary().
-deploy_container_name(#{<<"container_name">> := ContainerName}) when is_binary(ContainerName) ->
+deploy_container_name(#{container_name := ContainerName}) when is_binary(ContainerName) ->
     ContainerName;
 deploy_container_name(_) ->
     throw({deploy_error, <<"invalid deploy params: container_name missing">>}).
 
 -spec deploy_image(map()) -> binary().
-deploy_image(#{<<"create">> := #{<<"config">> := #{<<"image">> := Image}}}) when is_binary(Image) ->
+deploy_image(#{create := #{config := #{image := Image}}}) when is_binary(Image) ->
     Image;
 deploy_image(_) ->
     throw({deploy_error, <<"invalid deploy params: image missing">>}).

src/transport/efka_client.erl

@@ -32,7 +32,7 @@
 -record(state, {
     socket :: undefined | ssl:sslsocket(),
     %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
-    auth_ref = undefined :: undefined | binary(),
+    auth_ref = undefined :: undefined | reference(),
     dropped_message_count = 0 :: non_neg_integer()
 }).
 
@@ -90,7 +90,7 @@ callback_mode() ->
 %% 异步发送数据, 连接存在时候直接发送；否则缓存到DETS
 -spec handle_event(term(), term(), atom(), #state{}) -> term().
 handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) ->
-    Packet = term_to_binary({<<"message">>, {<<"data">>, #{<<"route_key">> => RouteKey, <<"metric">> => Metric}}}),
+    Packet = term_to_binary({message, {data, #{route_key => RouteKey, metric => Metric}}}),
     case StateName of
         ?STATE_ACTIVATED ->
             ok = ssl:send(Socket, Packet),
@@ -103,12 +103,12 @@ handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{so
 %% Task的stream流，只做实时的
 handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     logger:debug("[efka_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
-    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => Type, <<"stream">> => Stream}}}),
+    Packet = term_to_binary({message, {task_event, #{task_id => TaskId, type => Type, stream => Stream}}}),
     ok = ssl:send(Socket, Packet),
     {keep_state, State};
 
 handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => <<"close">>, <<"stream">> => Reason}}}),
+    Packet = term_to_binary({message, {task_event, #{task_id => TaskId, type => <<"close">>, stream => Reason}}}),
     ok = ssl:send(Socket, Packet),
     {keep_state, State};
 
@@ -127,10 +127,9 @@ handle_event({call, From}, dropped_message_count, _StateName, State = #state{dro
 handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) ->
     case connect_socket() of
         {ok, Socket} ->
-            Ref = request_ref(),
+            Ref = make_ref(),
             AuthPacket = auth_packet(Ref),
             ok = ssl:send(Socket, AuthPacket),
-            logger:debug("[efka_client] send auth request, ref: ~p", [Ref]),
             {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
         {error, _Reason} ->
             schedule_reconnect(),
@@ -158,16 +157,8 @@ handle_event(info, flush_cache, _, State) ->
 
 %% 处理收到的ssl消息
 handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) ->
-    try binary_to_term(PacketBin, [safe]) of
-        Packet ->
-            {keep_state, State, [{next_event, internal, Packet}]}
-    catch
-        error:Error ->
-            logger:warning("[efka_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]),
-            disconnect(Socket),
-            schedule_reconnect(),
-            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}
-    end;
+    Packet = binary_to_term(PacketBin, [safe]),
+    {keep_state, State, [{next_event, internal, Packet}]};
 handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) ->
     logger:debug("[efka_client] ssl error: ~p", [Reason]),
     disconnect(Socket),
@@ -180,32 +171,57 @@ handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
 %%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
 
 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。
-handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    handle_container_command(Ref, Request, Socket),
+handle_event(internal, {command, Ref, {container, #{action := list}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    Reply = docker_commands:get_containers(),
+    send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, State = #state{socket = Socket}) ->
+handle_event(internal, {command, Ref, {container, #{action := deploy, task_id := TaskId, params := Params}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    Reply = docker_deploy_manager:deploy(TaskId, Params),
+    send_container_response(Socket, Ref, Reply),
+    {keep_state, State};
+handle_event(internal, {command, Ref, {container, #{action := start, target := Target}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    Reply = docker_commands:start_container(container_target(Target)),
+    send_container_response(Socket, Ref, Reply),
+    {keep_state, State};
+handle_event(internal, {command, Ref, {container, #{action := stop, target := Target, timeout_seconds := TimeoutSeconds}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
+    send_container_response(Socket, Ref, Reply),
+    {keep_state, State};
+handle_event(internal, {command, Ref, {container, #{action := kill, target := Target, signal := Signal}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
+    send_container_response(Socket, Ref, Reply),
+    {keep_state, State};
+handle_event(internal, {command, Ref, {container, #{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
+    send_container_response(Socket, Ref, Reply),
+    {keep_state, State};
+handle_event(internal, {command, Ref, {container, #{action := config, target := Target, config := Config}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
+    send_container_response(Socket, Ref, Reply),
+    {keep_state, State};
+handle_event(internal, {command, Ref, {container, Request}}, _StateName, State = #state{socket = Socket}) ->
     logger:notice("[efka_client] get an invalid command: ~p, agent invalid", [Request]),
     send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
     {keep_state, State};
 
 %% 处理response
-handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
+handle_event(internal, {response, AuthRef, {auth_response, ok}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
     logger:debug("[efka_client] auth success"),
     {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined}, [{next_event, info, flush_cache}]};
-handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, {<<"error">>, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
+handle_event(internal, {response, AuthRef, {auth_response, {error, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
     logger:debug("[efka_client] auth failed, reason: ~p", [Reason]),
     disconnect(Socket),
     schedule_reconnect(),
     {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
-handle_event(internal, {<<"response">>, _Ref, Reply}, StateName, State) ->
+handle_event(internal, {response, _Ref, Reply}, StateName, State) ->
     logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
-handle_event(internal, {<<"command_response">>, _Ref, Reply}, StateName, State) ->
+handle_event(internal, {command_response, _Ref, Reply}, StateName, State) ->
     logger:warning("[efka_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
 
 %% 处理Pub/Sub机制
-handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) ->
+handle_event(internal, {message, {pub, #{topic := Topic, qos := Qos, content := Content}}}, ?STATE_ACTIVATED, State) ->
     logger:debug("[efka_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
     efka_subscription:publish(Topic, Qos, Content),
     {keep_state, State};
@@ -217,45 +233,10 @@ handle_event(info, Info, _, State = #state{}) ->
     logger:notice("[efka_client] get unknown info: ~p", [Info]),
     {keep_state, State}.
 
--spec handle_container_command(binary(), term(), ssl:sslsocket()) -> ok.
-handle_container_command(Ref, #{<<"action">> := <<"list">>}, Socket) ->
-    Reply = docker_commands:get_containers(),
-    send_container_response(Socket, Ref, Reply),
-    ok;
-handle_container_command(Ref, #{<<"action">> := <<"deploy">>, <<"task_id">> := TaskId, <<"params">> := Params}, Socket) ->
-    Reply = docker_deploy_manager:deploy(TaskId, Params),
-    send_container_response(Socket, Ref, Reply),
-    ok;
-handle_container_command(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) ->
-    Reply = docker_commands:start_container(container_target(Target)),
-    send_container_response(Socket, Ref, Reply),
-    ok;
-handle_container_command(Ref, #{<<"action">> := <<"stop">>, <<"target">> := Target, <<"timeout_seconds">> := TimeoutSeconds}, Socket) ->
-    Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
-    send_container_response(Socket, Ref, Reply),
-    ok;
-handle_container_command(Ref, #{<<"action">> := <<"kill">>, <<"target">> := Target, <<"signal">> := Signal}, Socket) ->
-    Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
-    send_container_response(Socket, Ref, Reply),
-    ok;
-handle_container_command(Ref, #{<<"action">> := <<"remove">>, <<"target">> := Target, <<"force">> := Force, <<"remove_volumes">> := RemoveVolumes}, Socket) ->
-    Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
-    send_container_response(Socket, Ref, Reply),
-    ok;
-handle_container_command(Ref, #{<<"action">> := <<"config">>, <<"target">> := Target, <<"config">> := Config}, Socket) ->
-    Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
-    send_container_response(Socket, Ref, Reply),
-    ok;
-handle_container_command(Ref, Request, Socket) ->
-    logger:notice("[efka_client] get an invalid command: ~p, agent invalid", [Request]),
-    send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
-    ok.
-
 -spec terminate(term(), atom(), #state{}) -> ok.
-terminate(Reason, _StateName, _State = #state{socket = Socket}) ->
+terminate(_Reason, _StateName, _State = #state{socket = Socket}) ->
     disconnect(Socket),
     efka_client_cache:close(),
-    logger:notice("[efka_client] terminate with reason: ~p", [Reason]),
     ok.
 
 -spec code_change(term(), atom(), #state{}, term()) -> {ok, atom(), #state{}}.
@@ -266,17 +247,17 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
--spec auth_packet(binary()) -> binary().
-auth_packet(Ref) when is_binary(Ref) ->
+-spec auth_packet(reference()) -> binary().
+auth_packet(Ref) when is_reference(Ref) ->
     {ok, AuthInfo} = application:get_env(efka, auth),
     UUID = proplists:get_value(uuid, AuthInfo),
     Token = proplists:get_value(token, AuthInfo),
 
     Timestamp = efka_util:timestamp(),
-    term_to_binary({<<"request">>, Ref, {<<"auth_request">>, #{
-        <<"uuid">> => list_to_binary(UUID),
-        <<"token">> => list_to_binary(Token),
-        <<"timestamp">> => Timestamp
+    term_to_binary({request, Ref, {auth_request, #{
+        uuid => list_to_binary(UUID),
+        token => list_to_binary(Token),
+        timestamp => Timestamp
     }}}).
 
 -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}.
@@ -303,45 +284,15 @@ disconnect(Socket) ->
 schedule_reconnect() ->
     erlang:start_timer(5000, self(), create_transport).
 
--spec request_ref() -> binary().
-request_ref() ->
-    crypto:strong_rand_bytes(16).
-
--spec send_container_response(ssl:sslsocket(), binary(), term()) -> ok.
+-spec send_container_response(ssl:sslsocket(), reference(), term()) -> ok.
 send_container_response(Socket, Ref, Reply) ->
-    Packet = term_to_binary({<<"command_response">>, Ref, {<<"container">>, safe_reply(Reply)}}),
+    Packet = term_to_binary({command_response, Ref, {container, Reply}}),
     ok = ssl:send(Socket, Packet).
 
--spec safe_reply(term()) -> term().
-safe_reply(ok) ->
-    <<"ok">>;
-safe_reply({ok, Result}) ->
-    {<<"ok">>, safe_term(Result)};
-safe_reply({error, Reason}) ->
-    {<<"error">>, safe_term(Reason)}.
-
--spec safe_term(term()) -> term().
-safe_term(true) ->
-    true;
-safe_term(false) ->
-    false;
-safe_term(undefined) ->
-    <<"__undefined__">>;
-safe_term(Value) when is_atom(Value) ->
-    atom_to_binary(Value, utf8);
-safe_term(Value) when is_map(Value) ->
-    maps:from_list([{safe_term(K), safe_term(V)} || {K, V} <- maps:to_list(Value)]);
-safe_term(Value) when is_list(Value) ->
-    [safe_term(Item) || Item <- Value];
-safe_term(Value) when is_tuple(Value) ->
-    list_to_tuple([safe_term(Item) || Item <- tuple_to_list(Value)]);
-safe_term(Value) ->
-    Value.
-
 -spec container_target(map()) -> binary().
 container_target(Target) when is_map(Target) ->
-    NameBin = to_binary(maps:get(<<"name">>, Target, <<>>)),
-    IdBin = to_binary(maps:get(<<"id">>, Target, <<>>)),
+    NameBin = to_binary(maps:get(name, Target, <<>>)),
+    IdBin = to_binary(maps:get(id, Target, <<>>)),
     case NameBin of
         <<>> ->
             true = IdBin =/= <<>>,