From c17750a5d9fa273325ee68cfb36998c9e1197321 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sun, 26 Apr 2026 15:39:50 +0800 Subject: [PATCH] fix codec --- src/transport/cloud_wire.erl | 81 ------------------------------- src/transport/tcp/ssl_channel.erl | 13 ++--- 2 files changed, 7 insertions(+), 87 deletions(-) delete mode 100644 src/transport/cloud_wire.erl diff --git a/src/transport/cloud_wire.erl b/src/transport/cloud_wire.erl deleted file mode 100644 index a788ee3..0000000 --- a/src/transport/cloud_wire.erl +++ /dev/null @@ -1,81 +0,0 @@ --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}). diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl index 7574ebf..23ba4db 100644 --- a/src/transport/tcp/ssl_channel.erl +++ b/src/transport/tcp/ssl_channel.erl @@ -100,19 +100,19 @@ handle_call(_Request, _From, State) -> %% 发送消息, 基于pub/sub机制 handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) -> - Packet = cloud_wire:encode({message, {pub, #{topic => Topic, qos => Qos, content => Content}}}), + Packet = term_to_binary({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}) -> - Packet = cloud_wire:encode({message, {auth_control, Command}}), + Packet = term_to_binary({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}) -> - Packet = cloud_wire:encode({request, PacketId, Body}), + Packet = term_to_binary({request, PacketId, Body}), TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {request_timeout, PacketId}), Transport:send(Socket, Packet), @@ -140,8 +140,9 @@ 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, PacketBin}, State = #state{transport = Transport, socket = Socket, host_pid = HostPid, inflight = Inflight}) -> - case cloud_wire:decode(PacketBin) of +handle_info({ssl, Socket, PacketBin}, State = #state{transport = Transport, socket = Socket, host_pid = HostPid, inflight = Inflight}) + when is_binary(PacketBin) -> + case binary_to_term(PacketBin, [safe]) of {request, PacketId, Body} -> handle_request_frame(PacketId, Body, Transport, Socket, State); {message, Body} -> @@ -280,7 +281,7 @@ handle_response_frame(PacketId, Reply, _Inflight, State) -> -spec send_reply_frame(module(), any(), non_neg_integer(), tuple()) -> any(). send_reply_frame(Transport, Socket, PacketId, Reply) -> - Packet = cloud_wire:encode({response, PacketId, Reply}), + Packet = term_to_binary({response, PacketId, Reply}), Transport:send(Socket, Packet). -spec decode_reply({ok, binary()} | {error, integer(), binary()}) ->