From ad34614bda3690cc5c90b993ea8786a301b2f09a Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sat, 9 May 2026 22:56:41 +0800 Subject: [PATCH] fix binary_to_term --- docs/container_deploy_request.md | 148 +++++++++++++++--------------- docs/efka_iot_protocol.md | 129 +++++++++++++------------- src/host/iot_host.erl | 28 +++--- src/transport/tcp/ssl_channel.erl | 88 +++++++++++------- 4 files changed, 215 insertions(+), 178 deletions(-) diff --git a/docs/container_deploy_request.md b/docs/container_deploy_request.md index 6eed456..6f85f39 100644 --- a/docs/container_deploy_request.md +++ b/docs/container_deploy_request.md @@ -195,14 +195,14 @@ miss requied parameter: <<"image">> ```erlang #{ - action => deploy, - task_id => TaskId, - params => #{ - container_name => ContainerName, - create => #{ - config => ContainerConfig, - host_config => HostConfig, - networking_config => NetworkingConfig + <<"action">> => <<"deploy">>, + <<"task_id">> => TaskId, + <<"params">> => #{ + <<"container_name">> => ContainerName, + <<"create">> => #{ + <<"config">> => ContainerConfig, + <<"host_config">> => HostConfig, + <<"networking_config">> => NetworkingConfig } } } @@ -211,19 +211,21 @@ miss requied parameter: <<"image">> 该 map 会通过 efka/iot 长连接协议下发： ```erlang -{command, Ref, {container, #{ - action => deploy, - task_id => TaskId, - params => Params +{<<"command">>, Ref, {<<"container">>, #{ + <<"action">> => <<"deploy">>, + <<"task_id">> => TaskId, + <<"params">> => Params }}} ``` efka 返回： ```erlang -{command_response, Ref, {container, Reply}} +{<<"command_response">>, Ref, {<<"container">>, Reply}} ``` +其中 `Ref` 是 `crypto:strong_rand_bytes(16)` 生成的 16 字节 binary。网络帧只使用 safe term，协议 label、command map key 和 action 使用 binary；`efka` 收到后直接按 binary key/action 处理。 + HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行错误通常返回 HTTP 400；找不到 host 返回业务错误 code 404。 ## 7. Docker create config 映射 @@ -271,7 +273,7 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行 `create.networking_config` 由 `build_docker_networking_config/1` 生成： ```erlang -#{endpoints => [#{name => Network} || Network <- Networks]} +#{<<"endpoints">> => [#{<<"name">> => Network} || Network <- Networks]} ``` 来源字段： @@ -283,9 +285,9 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行 转换结果： ```erlang -#{endpoints => [ - #{name => <<"bridge">>}, - #{name => <<"mynet">>} +#{<<"endpoints">> => [ + #{<<"name">> => <<"bridge">>}, + #{<<"name">> => <<"mynet">>} ]} ``` @@ -302,7 +304,7 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行 转换： ```erlang -#{name => <<"always">>, maximum_retry_count => 0} +#{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0} ``` 输入： @@ -314,7 +316,7 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行 转换： ```erlang -#{name => <<"on-failure">>, maximum_retry_count => 3} +#{<<"name">> => <<"on-failure">>, <<"maximum_retry_count">> => 3} ``` ### expose @@ -329,9 +331,9 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行 ```erlang [ - #{container_port => 80, protocol => <<"tcp">>}, - #{container_port => 443, protocol => <<"tcp">>}, - #{container_port => 53, protocol => <<"udp">>} + #{<<"container_port">> => 80, <<"protocol">> => <<"tcp">>}, + #{<<"container_port">> => 443, <<"protocol">> => <<"tcp">>}, + #{<<"container_port">> => 53, <<"protocol">> => <<"udp">>} ] ``` @@ -381,14 +383,14 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行 ```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">> }, #{ - path_on_host => <<"/dev/snd">>, - path_in_container => <<"/dev/snd">>, - cgroup_permissions => <<"rwm">> + <<"path_on_host">> => <<"/dev/snd">>, + <<"path_in_container">> => <<"/dev/snd">>, + <<"cgroup_permissions">> => <<"rwm">> } ] ``` @@ -414,8 +416,8 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行 ```erlang [ - #{name => <<"nofile">>, soft => 1024, hard => 2048}, - #{name => <<"nproc">>, soft => 4096, hard => 4096} + #{<<"name">> => <<"nofile">>, <<"soft">> => 1024, <<"hard">> => 2048}, + #{<<"name">> => <<"nproc">>, <<"soft">> => 4096, <<"hard">> => 4096} ] ``` @@ -465,10 +467,10 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行 ```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 } ``` @@ -512,8 +514,8 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行 转换： ```erlang -memory => 536870912, -memory_reservation => 1073741824 +<<"memory">> => 536870912, +<<"memory_reservation">> => 1073741824 ``` 支持单位： @@ -539,7 +541,7 @@ memory_reservation => 1073741824 转换： ```erlang -nano_cpus => 1500000000 +<<"nano_cpus">> => 1500000000 ``` 规则： @@ -563,46 +565,46 @@ nano_cpus => 1500000000 } ``` -对应内部命令示意： +对应下发给 `efka` 的协议命令示意： ```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;">>], - entrypoint => [], - env => [], - labels => #{}, - volumes => [], - user => <<>>, - working_dir => <<>>, - hostname => <<>>, - exposed_ports => [], - healthcheck => undefined +{<<"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;">>], + <<"entrypoint">> => [], + <<"env">> => [], + <<"labels">> => #{}, + <<"volumes">> => [], + <<"user">> => <<>>, + <<"working_dir">> => <<>>, + <<"hostname">> => <<>>, + <<"exposed_ports">> => [], + <<"healthcheck">> => <<"__undefined__">> }, - host_config => #{ - binds => [], - network_mode => <<>>, - restart_policy => #{name => <<"always">>, maximum_retry_count => 0}, - privileged => false, - cap_add => [], - cap_drop => [], - devices => [], - memory => 0, - memory_reservation => 0, - nano_cpus => 0, - cpu_shares => 0, - ulimits => [], - tmpfs => #{}, - sysctls => #{}, - extra_hosts => [] + <<"host_config">> => #{ + <<"binds">> => [], + <<"network_mode">> => <<>>, + <<"restart_policy">> => #{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0}, + <<"privileged">> => false, + <<"cap_add">> => [], + <<"cap_drop">> => [], + <<"devices">> => [], + <<"memory">> => 0, + <<"memory_reservation">> => 0, + <<"nano_cpus">> => 0, + <<"cpu_shares">> => 0, + <<"ulimits">> => [], + <<"tmpfs">> => #{}, + <<"sysctls">> => #{}, + <<"extra_hosts">> => [] }, - networking_config => #{endpoints => []} + <<"networking_config">> => #{<<"endpoints">> => []} } } }}} diff --git a/docs/efka_iot_protocol.md b/docs/efka_iot_protocol.md index c05b99a..257430f 100644 --- a/docs/efka_iot_protocol.md +++ b/docs/efka_iot_protocol.md @@ -2,95 +2,99 @@ 本文档描述 `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` 使用 `make_ref()` 生成，只在当前连接的 inflight 表内匹配。 +- `Ref` 使用 `crypto:strong_rand_bytes(16)` 生成，只在当前连接的 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` 主机上的容器列表。 @@ -99,9 +103,9 @@ ok ```erlang #{ - action => deploy, - task_id => TaskId, - params => Params + <<"action">> => <<"deploy">>, + <<"task_id">> => TaskId, + <<"params">> => Params } ``` @@ -111,8 +115,8 @@ ok ```erlang #{ - action => start, - target => Target + <<"action">> => <<"start">>, + <<"target">> => Target } ``` @@ -120,9 +124,9 @@ ok ```erlang #{ - action => stop, - target => Target, - timeout_seconds => TimeoutSeconds + <<"action">> => <<"stop">>, + <<"target">> => Target, + <<"timeout_seconds">> => TimeoutSeconds } ``` @@ -130,9 +134,9 @@ ok ```erlang #{ - action => kill, - target => Target, - signal => Signal + <<"action">> => <<"kill">>, + <<"target">> => Target, + <<"signal">> => Signal } ``` @@ -140,10 +144,10 @@ ok ```erlang #{ - action => remove, - target => Target, - force => Force, - remove_volumes => RemoveVolumes + <<"action">> => <<"remove">>, + <<"target">> => Target, + <<"force">> => Force, + <<"remove_volumes">> => RemoveVolumes } ``` @@ -151,9 +155,9 @@ ok ```erlang #{ - action => config, - target => Target, - config => Config + <<"action">> => <<"config">>, + <<"target">> => Target, + <<"config">> => Config } ``` @@ -165,8 +169,8 @@ ok ```erlang #{ - name => ContainerName, - id => ContainerId + <<"name">> => ContainerName, + <<"id">> => ContainerId } ``` @@ -179,9 +183,9 @@ ok ### efka -> iot: data ```erlang -{message, {data, #{ - route_key => RouteKey, - metric => Metric +{<<"message">>, {<<"data">>, #{ + <<"route_key">> => RouteKey, + <<"metric">> => Metric }}} ``` @@ -190,20 +194,20 @@ ok ### 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 }}} ``` @@ -218,10 +222,10 @@ GET /event_stream?uuid=&task_id= ### iot -> efka: pub ```erlang -{message, {pub, #{ - topic => Topic, - qos => Qos, - content => Content +{<<"message">>, {<<"pub">>, #{ + <<"topic">> => Topic, + <<"qos">> => Qos, + <<"content">> => Content }}} ``` @@ -242,5 +246,6 @@ GET /event_stream?uuid=&task_id= - 旧容器回复：`{response, Ref, {container_response, ...}}` - 旧授权控制：`{message, {auth_control, Command}}` - 已移除的 auth command：`{command, Ref, {auth, activate | deactivate}}` +- 旧 Ref：Erlang `reference()`，例如 `make_ref()` 生成的值。 如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。 diff --git a/src/host/iot_host.erl b/src/host/iot_host.erl index 8f56320..2f23064 100644 --- a/src/host/iot_host.erl +++ b/src/host/iot_host.erl @@ -101,15 +101,17 @@ get_metric(Pid) when is_pid(Pid) -> attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) -> gen_statem:call(Pid, {attach_channel, ChannelPid}). --spec get_containers(Pid :: pid()) -> {ok, Ref :: reference()} | {error, Reason :: any()}. +-type request_ref() :: binary(). + +-spec get_containers(Pid :: pid()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. get_containers(Pid) when is_pid(Pid) -> container_call(Pid, docker_container_builder:list_request()). --spec config_container(Pid :: pid(), ContainerName :: binary(), ConfigJson :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}. +-spec config_container(Pid :: pid(), ContainerName :: binary(), ConfigJson :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. config_container(Pid, ContainerName, ConfigJson) when is_pid(Pid), is_binary(ContainerName), is_binary(ConfigJson) -> container_call(Pid, docker_container_builder:config_request(ContainerName, ConfigJson)). --spec deploy_container(Pid :: pid(), TaskId :: integer(), Config :: map()) -> {ok, Ref :: reference()} | {error, Reason :: any()}. +-spec deploy_container(Pid :: pid(), TaskId :: integer(), Config :: map()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. deploy_container(Pid, TaskId, Config) when is_pid(Pid), is_integer(TaskId), is_map(Config) -> case docker_container_builder:deploy_request(TaskId, Config) of {ok, Request} -> @@ -118,25 +120,25 @@ deploy_container(Pid, TaskId, Config) when is_pid(Pid), is_integer(TaskId), is_m {error, Reason} end. --spec start_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}. +-spec start_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. start_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> container_call(Pid, docker_container_builder:start_request(ContainerName)). --spec stop_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}. +-spec stop_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. stop_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> container_call(Pid, docker_container_builder:stop_request(ContainerName)). --spec kill_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}. +-spec kill_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. kill_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> container_call(Pid, docker_container_builder:kill_request(ContainerName)). --spec remove_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}. +-spec remove_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> container_call(Pid, docker_container_builder:remove_request(ContainerName)). --spec await_reply(Pid :: pid(), Ref :: reference(), Timeout :: integer()) -> +-spec await_reply(Pid :: pid(), Ref :: request_ref(), Timeout :: integer()) -> ok | {ok, Result :: term()} | {error, Reason :: term()}. -await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_reference(Ref), is_integer(Timeout) -> +await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_binary(Ref), is_integer(Timeout) -> receive {command_reply, Ref, ok} -> ok; @@ -223,7 +225,7 @@ handle_event({call, From}, get_status, _, State = #state{channel_pid = ChannelPi handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) -> case HasSession andalso is_pid(ChannelPid) of true -> - Ref = make_ref(), + Ref = request_ref(), ok = ssl_channel:container_call(ChannelPid, ReceiverPid, Ref, Request), {keep_state, State, [{reply, From, {ok, Ref}}]}; false -> @@ -363,7 +365,7 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) -> %%%=================================================================== -spec container_call(Pid :: pid(), Request :: term()) -> - {ok, Ref :: reference()} | {error, Reason :: any()}. + {ok, Ref :: request_ref()} | {error, Reason :: any()}. container_call(Pid, Request) when is_pid(Pid) -> gen_statem:call(Pid, {container_call, self(), Request}). @@ -392,6 +394,10 @@ flush_reply(Ref) -> ok end. +-spec request_ref() -> request_ref(). +request_ref() -> + crypto:strong_rand_bytes(16). + -spec maybe_mark_host_offline(binary()) -> ok. maybe_mark_host_offline(UUID) -> case iot_api_client:get_host_by_uuid(UUID) of diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl index 9a2375c..7ef6b8f 100644 --- a/src/transport/tcp/ssl_channel.erl +++ b/src/transport/tcp/ssl_channel.erl @@ -37,18 +37,20 @@ timer_ref :: reference() }). +-type request_ref() :: binary(). + %% 向通道中写入消息 -spec pub(Pid :: pid(), Topic :: binary(), Qos :: integer(), Content :: binary()) -> ok. pub(Pid, Topic, Qos, Content) when is_pid(Pid), is_binary(Topic), is_integer(Qos), is_binary(Content) -> gen_server:cast(Pid, {pub, Topic, Qos, Content}). --spec container_call(Pid :: pid(), ReceiverPid :: pid(), Ref :: reference(), Request :: map()) -> ok. -container_call(Pid, ReceiverPid, Ref, Request) when is_pid(Pid), is_pid(ReceiverPid), is_reference(Ref), is_map(Request) -> +-spec container_call(Pid :: pid(), ReceiverPid :: pid(), Ref :: request_ref(), Request :: map()) -> ok. +container_call(Pid, ReceiverPid, Ref, Request) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Ref), is_map(Request) -> gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {container, Request}}), ok. --spec cancel_command_call(Pid :: pid(), Ref :: reference()) -> ok. -cancel_command_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) -> +-spec cancel_command_call(Pid :: pid(), Ref :: request_ref()) -> ok. +cancel_command_call(Pid, Ref) when is_pid(Pid), is_binary(Ref) -> gen_server:call(Pid, {cancel_command_call, Ref}). %% 关闭方法 @@ -85,7 +87,7 @@ handle_call(_Request, _From, State) -> %% 发送消息, 基于pub/sub机制 handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) -> - Packet = term_to_binary({message, {pub, #{topic => Topic, qos => Qos, content => Content}}}), + Packet = term_to_binary({<<"message">>, {<<"pub">>, #{<<"topic">> => Topic, <<"qos">> => Qos, <<"content">> => Content}}}), case Transport:send(Socket, Packet) of ok -> {noreply, State}; @@ -96,7 +98,7 @@ handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, so %% iot 请求 efka 时使用 command/command_response 语义。 handle_cast({command_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) -> - Packet = term_to_binary({command, Ref, Body}), + Packet = term_to_binary({<<"command">>, Ref, encode_command_body(Body)}), case Transport:send(Socket, Packet) of ok -> TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}), @@ -129,17 +131,17 @@ handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, h handle_info({ssl, Socket, PacketBin}, State = #state{socket = Socket}) when is_binary(PacketBin) -> try binary_to_term(PacketBin, [safe]) of - {request, Ref, Body} -> + {<<"request">>, Ref, Body} -> handle_request_frame(Ref, Body, State); - {message, Body} -> + {<<"message">>, Body} -> handle_message_frame(Body, State); - {command_response, Ref, Response} -> + {<<"command_response">>, Ref, Response} -> handle_command_response_frame(Ref, Response, State); Other -> logger:warning("[ssl_channel] unsupported packet: ~p", [Other]), {stop, bad_packet, State} catch error:Error -> - logger:warning("[ssl_channel] binary_to_term get error: ~p", [Error]), + logger:warning("[ssl_channel] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]), {stop, bad_packet, State} end; @@ -171,23 +173,23 @@ code_change(_OldVsn, State, _Extra) -> %%%% helper methods %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% --spec handle_request_frame(reference(), tuple(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. -handle_request_frame(Ref, {auth_request, #{uuid := UUID}}, State = #state{is_authed = true}) -> +-spec handle_request_frame(request_ref(), tuple(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. +handle_request_frame(Ref, {<<"auth_request">>, #{<<"uuid">> := UUID}}, State = #state{is_authed = true}) -> logger:warning("[ws_channel] repeated auth request, ref: ~p, uuid: ~p, close channel", [Ref, UUID]), {stop, repeated_auth, State}; -handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timestamp := Timestamp}}, State = #state{transport = Transport, socket = Socket}) -> +handle_request_frame(Ref, {<<"auth_request">>, #{<<"uuid">> := UUID, <<"token">> := Token, <<"timestamp">> := Timestamp}}, State = #state{transport = Transport, socket = Socket}) -> maybe ok ?= auth(Token, UUID, Timestamp), {ok, HostPid} ?= iot_host:lookup_pid(UUID), ok ?= iot_host:attach_channel(HostPid, self()), erlang:monitor(process, HostPid), - ok = send_reply_frame(Transport, Socket, Ref, {auth_response, ok}), + ok = send_reply_frame(Transport, Socket, Ref, {<<"auth_response">>, <<"ok">>}), logger:debug("[ws_channel] auth uuid: ~p", [UUID]), {noreply, State#state{uuid = UUID, is_authed = true, host_pid = HostPid}} else {error, Reason} -> logger:warning("[ws_channel] uuid: ~p, auth failed with reason: ~p", [UUID, Reason]), - case send_reply_frame(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}) of + case send_reply_frame(Transport, Socket, Ref, {<<"auth_response">>, {<<"error">>, {<<"failed">>, Reason}}}) of ok -> {stop, Reason, State}; {error, SendReason} -> @@ -195,7 +197,7 @@ handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timesta {stop, {send_failed, SendReason}, State} end end; -handle_request_frame(Ref, {container, ContainerCommand}, State) -> +handle_request_frame(Ref, {<<"container">>, ContainerCommand}, State) -> logger:warning("[ws_channel] unsupported request message type: container, ref: ~p, command: ~p", [Ref, ContainerCommand]), {stop, normal, State}; handle_request_frame(Ref, Body, State) -> @@ -204,10 +206,10 @@ handle_request_frame(Ref, Body, State) -> -spec handle_message_frame(tuple(), #state{}) -> {noreply, #state{}}. -handle_message_frame({data, #{route_key := RouteKey, metric := Metric}}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> +handle_message_frame({<<"data">>, #{<<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> iot_host:handle(HostPid, {data, RouteKey, Metric}), {noreply, State}; -handle_message_frame({task_event, Event}, State = #state{uuid = UUID, host_pid = HostPid}) when is_binary(UUID), is_pid(HostPid) -> +handle_message_frame({<<"task_event">>, Event}, State = #state{uuid = UUID, host_pid = HostPid}) when is_binary(UUID), is_pid(HostPid) -> handle_event_stream_frame(UUID, Event), {noreply, State}; handle_message_frame(Body, State) -> @@ -215,18 +217,18 @@ handle_message_frame(Body, State) -> {noreply, State}. -spec handle_event_stream_frame(binary(), map()) -> ok. -handle_event_stream_frame(UUID, #{task_id := TaskId, type := <<"close">>, stream := Reason}) -> +handle_event_stream_frame(UUID, #{<<"task_id">> := TaskId, <<"type">> := <<"close">>, <<"stream">> := Reason}) -> iot_container_task_sup:close(UUID, TaskId, Reason); -handle_event_stream_frame(UUID, #{task_id := TaskId, type := Type, stream := Stream}) -> +handle_event_stream_frame(UUID, #{<<"task_id">> := TaskId, <<"type">> := Type, <<"stream">> := Stream}) -> logger:debug("[ssl_channel] get uuid: ~p, task_id: ~p, type: ~ts, stream: ~ts", [UUID, TaskId, Type, Stream]), iot_container_task_sup:stream(UUID, TaskId, Type, Stream); handle_event_stream_frame(UUID, Event) -> logger:warning("[ssl_channel] invalid task_event, uuid: ~p, event: ~p", [UUID, Event]), ok. --spec handle_command_response_frame(reference(), tuple(), #state{}) -> +-spec handle_command_response_frame(request_ref(), tuple(), #state{}) -> {noreply, #state{}}. -handle_command_response_frame(Ref, Reply, State = #state{inflight = Inflight}) when is_reference(Ref) -> +handle_command_response_frame(Ref, Reply, State = #state{inflight = Inflight}) when is_binary(Ref) -> case maps:take(Ref, Inflight) of error -> {noreply, State}; @@ -239,23 +241,23 @@ handle_command_response_frame(Ref, Reply, State) -> logger:warning("[ws_channel] unexpected command_response frame, ref: ~p, reply: ~p", [Ref, Reply]), {noreply, State}. --spec send_reply_frame(module(), any(), reference(), tuple()) -> ok | {error, term()}. +-spec send_reply_frame(module(), any(), request_ref(), tuple()) -> ok | {error, term()}. send_reply_frame(Transport, Socket, Ref, Reply) -> - Packet = term_to_binary({response, Ref, Reply}), + Packet = term_to_binary({<<"response">>, Ref, Reply}), Transport:send(Socket, Packet). --spec decode_command_response({container, ok | {ok, term()} | {error, term()}} | tuple()) -> +-spec decode_command_response({binary(), term()} | tuple()) -> ok | {ok, term()} | {error, term()}. -decode_command_response({container, ok}) -> +decode_command_response({<<"container">>, <<"ok">>}) -> ok; -decode_command_response({container, {ok, Result}}) -> +decode_command_response({<<"container">>, {<<"ok">>, Result}}) -> {ok, Result}; -decode_command_response({container, {error, Reason}}) -> +decode_command_response({<<"container">>, {<<"error">>, Reason}}) -> {error, Reason}; decode_command_response(_Reply) -> {error, invalid_response}. --spec deliver_command_response(undefined | pid(), reference(), tuple()) -> ok. +-spec deliver_command_response(undefined | pid(), request_ref(), tuple()) -> ok. deliver_command_response(undefined, _Ref, _Reply) -> ok; deliver_command_response(ReceiverPid, Ref, Reply) when is_pid(ReceiverPid) -> @@ -266,21 +268,21 @@ deliver_command_response(ReceiverPid, Ref, Reply) when is_pid(ReceiverPid) -> logger:warning("[ws_channel] get command_response: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref]) end. --spec deliver_command_error(undefined | pid(), reference(), term()) -> ok. +-spec deliver_command_error(undefined | pid(), request_ref(), term()) -> ok. deliver_command_error(ReceiverPid, Ref, Reason) when is_pid(ReceiverPid) -> ReceiverPid ! {command_reply, Ref, {error, Reason}}, ok; deliver_command_error(_ReceiverPid, _Ref, _Reason) -> ok. --spec reply_command_timeout(undefined | pid(), reference()) -> ok. +-spec reply_command_timeout(undefined | pid(), request_ref()) -> ok. reply_command_timeout(ReceiverPid, Ref) when is_pid(ReceiverPid) -> ReceiverPid ! {command_reply, Ref, {error, timeout}}, ok; reply_command_timeout(_ReceiverPid, _Ref) -> ok. --spec reply_command_closed(reference(), #inflight_command{}, term()) -> ok. +-spec reply_command_closed(request_ref(), #inflight_command{}, term()) -> ok. reply_command_closed(Ref, #inflight_command{receiver_pid = ReceiverPid}, Reason) when is_pid(ReceiverPid) -> ReceiverPid ! {command_reply, Ref, {error, {channel_closed, Reason}}}, ok; @@ -304,3 +306,25 @@ auth(Token, UUID, Timestamp) when is_binary(Token), is_binary(UUID), is_integer( false -> {error, <<"invalid timestamp">>} end. + +-spec encode_command_body({container, map()}) -> {binary(), map()}. +encode_command_body({container, Request}) -> + {<<"container">>, safe_term(Request)}. + +-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.