From 0e7bee5188fa974b33e4b6cafc3e265c44093c1f Mon Sep 17 00:00:00 2001
From: anlicheng <244108715@qq.com>
Date: Tue, 7 Jul 2026 23:56:13 +0800
Subject: [PATCH] fix protocol

---
 apps/efka/src/iot/efka_iot_client.erl | 154 ++++++++++----------------
 1 file changed, 56 insertions(+), 98 deletions(-)

diff --git a/apps/efka/src/iot/efka_iot_client.erl b/apps/efka/src/iot/efka_iot_client.erl
index f79808c..0355dd3 100644
--- a/apps/efka/src/iot/efka_iot_client.erl
+++ b/apps/efka/src/iot/efka_iot_client.erl
@@ -40,8 +40,8 @@
     socket :: undefined | ssl:sslsocket(),
     outbox :: efka_iot_outbox:outbox(),
     streams = #{},
-    %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
-    auth_ref = undefined :: undefined | pos_integer(),
+    %% 保存当前 auth 请求的 packet id，用来建立 auth 请求和响应的对应关系
+    auth_pkt_id = undefined :: undefined | pos_integer(),
     next_pkt_id = 1 :: pos_integer(),
     ping_timer_ref = undefined :: undefined | reference(),
     dropped_message_count = 0 :: non_neg_integer()
@@ -195,7 +195,7 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State =
             AuthPacket = auth_packet(PktId),
             ok = ssl:send(Socket, AuthPacket),
             logger:debug("[efka_iot_client] send auth request, packet_id: ~p", [PktId]),
-            {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = PktId, next_pkt_id = PktId + 1}, [{state_timeout, 5000, auth_timeout}]};
+            {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_pkt_id = PktId, next_pkt_id = PktId + 1}, [{state_timeout, 5000, auth_timeout}]};
         {error, _Reason} ->
             schedule_reconnect(),
             {keep_state, State#state{socket = undefined}}
@@ -205,7 +205,7 @@ handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = S
     logger:debug("[efka_iot_client] auth request timeout"),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_pkt_id = undefined}};
 
 handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #state{socket = Socket, ping_timer_ref = TimerRef}) ->
     Packet = encode_message_frame(ping),
@@ -217,7 +217,7 @@ handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #sta
             disconnect(Socket),
             schedule_reconnect(),
             NState = close_all_streams({send_ping_failed, Reason}, State),
-            {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
+            {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}}
     end;
 handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) ->
     {keep_state, State};
@@ -254,7 +254,7 @@ handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket})
             cancel_ssl_ping(State),
             schedule_reconnect(),
             NState = close_all_streams(bad_packet, State),
-            {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
+            {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}}
     end;
 handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) ->
     logger:debug("[efka_iot_client] ssl error: ~p", [Reason]),
@@ -262,14 +262,14 @@ handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) ->
     cancel_ssl_ping(State),
     schedule_reconnect(),
     NState = close_all_streams(ssl_error, State),
-    {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}};
 handle_event(info, {ssl_closed, Socket}, _, State = #state{}) ->
     logger:debug("[efka_iot_client] ssl closed"),
     disconnect(Socket),
     cancel_ssl_ping(State),
     schedule_reconnect(),
     NState = close_all_streams(ssl_closed, State),
-    {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}};
 
 handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName, State = #state{streams = Streams}) ->
     case take_stream_by_monitor(MonitorRef, WorkerPid, Streams) of
@@ -282,7 +282,7 @@ handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName,
                 undefined ->
                     {keep_state, State#state{streams = NStreams}};
                 Socket ->
-                    Packet = encode_stream_frame(StreamId, {reset, safe_term({worker_down, Reason})}),
+                    Packet = encode_stream_frame(StreamId, {reset, {worker_down, Reason}}),
                     ok = ssl:send(Socket, Packet),
                     {keep_state, State#state{streams = NStreams}}
             end
@@ -291,25 +291,24 @@ handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName,
 %%% 处理内部消息。TLS 收到的数据先经过 protobuf 解码，再由这里按本地事件分发。
 
 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。
-handle_event(internal, #'Command'{packet_id = Ref, body = {container, ContainerCommand}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    handle_container_command(Ref, ContainerCommand, Socket),
+handle_event(internal, #'Command'{packet_id = PktId, body = {container, ContainerCommand}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    handle_container_command(PktId, ContainerCommand, Socket),
     {keep_state, State};
-handle_event(internal, #'Command'{packet_id = Ref, body = {container, ContainerCommand}}, _StateName, State = #state{socket = Socket}) ->
-    Request = decode_container_command(ContainerCommand),
-    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
-    send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
+handle_event(internal, #'Command'{packet_id = PktId, body = {container, ContainerCommand}}, _StateName, State = #state{socket = Socket}) ->
+    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [ContainerCommand]),
+    send_container_response(Socket, PktId, {error, <<"agent invalid">>}),
     {keep_state, State};
 
 %% 处理response
-handle_event(internal, #'Response'{packet_id = AuthRef, body = {auth_response, #'Response.AuthResponse'{}}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
+handle_event(internal, #'Response'{packet_id = AuthPktId, body = {auth_response, #'Response.AuthResponse'{}}}, ?STATE_AUTH, State = #state{auth_pkt_id = AuthPktId}) ->
     logger:debug("[efka_iot_client] auth success"),
-    State1 = schedule_ssl_ping(State#state{auth_ref = undefined}),
+    State1 = schedule_ssl_ping(State#state{auth_pkt_id = undefined}),
     {next_state, ?STATE_ACTIVATED, State1, [{next_event, info, flush_cache}]};
-handle_event(internal, #'Response'{packet_id = AuthRef, body = {error, #'Response.Error'{reason = Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
+handle_event(internal, #'Response'{packet_id = AuthPktId, body = {error, #'Response.Error'{reason = Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_pkt_id = AuthPktId}) ->
     logger:debug("[efka_iot_client] auth failed, reason: ~p", [Reason]),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_pkt_id = undefined}};
 handle_event(internal, #'Response'{} = Reply, StateName, State) ->
     logger:warning("[efka_iot_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
@@ -372,14 +371,14 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
 %%%===================================================================
 
 -spec auth_packet(pos_integer()) -> binary().
-auth_packet(Ref) when is_integer(Ref), Ref > 0 ->
+auth_packet(PktId) when is_integer(PktId), PktId > 0 ->
     {ok, AuthInfo} = application:get_env(efka, auth),
     UUID = proplists:get_value(uuid, AuthInfo),
     Token = proplists:get_value(token, AuthInfo),
 
     Timestamp = efka_util:timestamp(),
     encode_transport_frame(?CLASS_REQUEST, #'Request'{
-        packet_id = Ref,
+        packet_id = PktId,
         body = {auth_request, #'Request.AuthRequest'{
             uuid = list_to_binary(UUID),
             token = list_to_binary(Token),
@@ -492,7 +491,7 @@ cancel_timer(TimerRef) ->
     ok.
 
 -spec send_container_response(ssl:sslsocket(), pos_integer(), term()) -> ok.
-send_container_response(Socket, Ref, Reply) ->
+send_container_response(Socket, PktId, Reply) ->
     Body = case Reply of
                ok ->
                    {result, <<"ok">>};
@@ -504,73 +503,45 @@ send_container_response(Socket, Ref, Reply) ->
                    {result, reply_to_binary(Other)}
            end,
     Packet = encode_transport_frame(?CLASS_COMMAND_RESPONSE, #'CommandResponse'{
-        packet_id = Ref,
+        packet_id = PktId,
         body = Body
     }),
     ok = ssl:send(Socket, Packet).
 
 -spec handle_container_command(pos_integer(), message_pb:'Command.Container'(), ssl:sslsocket()) -> ok.
-handle_container_command(Ref, ContainerCommand, Socket) ->
-    handle_container_request(Ref, decode_container_command(ContainerCommand), Socket).
-
--spec handle_container_request(pos_integer(), map(), ssl:sslsocket()) -> ok.
-handle_container_request(Ref, #{<<"action">> := <<"list">>}, Socket) ->
+handle_container_command(PktId, #'Command.Container'{action = {list, #'Command.Container.ContainerList'{}}}, Socket) ->
     Reply = docker_commands:get_containers(),
-    send_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, PktId, Reply),
     ok;
-handle_container_request(Ref, #{<<"action">> := <<"deploy">>, <<"task_id">> := TaskId, <<"params">> := Params}, Socket) ->
-    Reply = docker_deploy_manager:deploy(TaskId, Params),
-    send_container_response(Socket, Ref, Reply),
+handle_container_command(PktId, #'Command.Container'{action = {deploy, #'Command.Container.ContainerDeploy'{task_id = TaskId, params = Params}}}, Socket) ->
+    Reply = docker_deploy_manager:deploy(TaskId, decode_json_payload(Params)),
+    send_container_response(Socket, PktId, Reply),
     ok;
-handle_container_request(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) ->
+handle_container_command(PktId, #'Command.Container'{action = {start, #'Command.Container.ContainerStart'{target = Target}}}, Socket) ->
     Reply = docker_commands:start_container(container_target(Target)),
-    send_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, PktId, Reply),
     ok;
-handle_container_request(Ref, #{<<"action">> := <<"stop">>, <<"target">> := Target, <<"timeout_seconds">> := TimeoutSeconds}, Socket) ->
+handle_container_command(PktId, #'Command.Container'{action = {stop, #'Command.Container.ContainerStop'{target = Target, timeout_seconds = TimeoutSeconds}}}, Socket) ->
     Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
-    send_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, PktId, Reply),
     ok;
-handle_container_request(Ref, #{<<"action">> := <<"kill">>, <<"target">> := Target, <<"signal">> := Signal}, Socket) ->
+handle_container_command(PktId, #'Command.Container'{action = {kill, #'Command.Container.ContainerKill'{target = Target, signal = Signal}}}, Socket) ->
     Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
-    send_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, PktId, Reply),
     ok;
-handle_container_request(Ref, #{<<"action">> := <<"remove">>, <<"target">> := Target, <<"force">> := Force, <<"remove_volumes">> := RemoveVolumes}, Socket) ->
+handle_container_command(PktId, #'Command.Container'{action = {remove, #'Command.Container.ContainerRemove'{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),
+    send_container_response(Socket, PktId, Reply),
     ok;
-handle_container_request(Ref, #{<<"action">> := <<"config">>, <<"target">> := Target, <<"config">> := Config}, Socket) ->
+handle_container_command(PktId, #'Command.Container'{action = {config, #'Command.Container.ContainerConfig'{target = Target, config = Config}}}, Socket) ->
     Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
-    send_container_response(Socket, Ref, Reply),
+    send_container_response(Socket, PktId, Reply),
     ok;
-handle_container_request(Ref, Request, Socket) ->
-    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
-    send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
+handle_container_command(PktId, ContainerCommand, Socket) ->
+    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [ContainerCommand]),
+    send_container_response(Socket, PktId, {error, <<"agent invalid">>}),
     ok.
 
--spec decode_container_command(message_pb:'Command.Container'()) -> map().
-decode_container_command(#'Command.Container'{action = {list, #'Command.Container.ContainerList'{}}}) ->
-    #{<<"action">> => <<"list">>};
-decode_container_command(#'Command.Container'{action = {deploy, #'Command.Container.ContainerDeploy'{task_id = TaskId, params = Params}}}) ->
-    #{<<"action">> => <<"deploy">>, <<"task_id">> => TaskId, <<"params">> => decode_json_payload(Params)};
-decode_container_command(#'Command.Container'{action = {start, #'Command.Container.ContainerStart'{target = Target}}}) ->
-    #{<<"action">> => <<"start">>, <<"target">> => decode_container_target(Target)};
-decode_container_command(#'Command.Container'{action = {stop, #'Command.Container.ContainerStop'{target = Target, timeout_seconds = TimeoutSeconds}}}) ->
-    #{<<"action">> => <<"stop">>, <<"target">> => decode_container_target(Target), <<"timeout_seconds">> => TimeoutSeconds};
-decode_container_command(#'Command.Container'{action = {kill, #'Command.Container.ContainerKill'{target = Target, signal = Signal}}}) ->
-    #{<<"action">> => <<"kill">>, <<"target">> => decode_container_target(Target), <<"signal">> => Signal};
-decode_container_command(#'Command.Container'{action = {remove, #'Command.Container.ContainerRemove'{target = Target, force = Force, remove_volumes = RemoveVolumes}}}) ->
-    #{<<"action">> => <<"remove">>, <<"target">> => decode_container_target(Target), <<"force">> => Force, <<"remove_volumes">> => RemoveVolumes};
-decode_container_command(#'Command.Container'{action = {config, #'Command.Container.ContainerConfig'{target = Target, config = Config}}}) ->
-    #{<<"action">> => <<"config">>, <<"target">> => decode_container_target(Target), <<"config">> => Config};
-decode_container_command(Other) ->
-    #{<<"action">> => <<"invalid">>, <<"payload">> => safe_term(Other)}.
-
--spec decode_container_target(message_pb:'Command.Container.ContainerTarget'() | undefined) -> map().
-decode_container_target(#'Command.Container.ContainerTarget'{name = Name, id = Id}) ->
-    #{<<"name">> => iolist_to_binary(Name), <<"id">> => iolist_to_binary(Id)};
-decode_container_target(undefined) ->
-    #{<<"name">> => <<>>, <<"id">> => <<>>}.
-
 -spec decode_json_payload(iodata()) -> term().
 decode_json_payload(Payload) ->
     Bin = iolist_to_binary(Payload),
@@ -661,58 +632,45 @@ demonitor_stream(#stream_state{monitor_ref = MonitorRef}) ->
 reply_to_binary(Value) when is_binary(Value) ->
     Value;
 reply_to_binary(Value) ->
-    SafeValue = safe_term(Value),
     try iolist_to_binary(Value) of
         Bin ->
             Bin
     catch
         _:_ ->
-            try iolist_to_binary(json:encode(SafeValue)) of
+            try iolist_to_binary(json:encode(Value)) of
                 JsonBin ->
                     JsonBin
             catch
                 _:_ ->
-                    term_to_binary(SafeValue)
+                    unicode:characters_to_binary(io_lib:format("~p", [Value]))
             end
     end.
 
 -spec reason_to_binary(term()) -> binary().
 reason_to_binary(Reason) when is_binary(Reason) ->
     Reason;
-reason_to_binary(Reason) when is_list(Reason) ->
-    iolist_to_binary(Reason);
 reason_to_binary(Reason) ->
-    term_to_binary(safe_term(Reason)).
+    try iolist_to_binary(Reason) of
+        Bin ->
+            Bin
+    catch
+        _:_ ->
+            unicode:characters_to_binary(io_lib:format("~p", [Reason]))
+    end.
 
--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, <<>>)),
+-spec container_target(message_pb:'Command.Container.ContainerTarget'() | undefined) -> binary().
+container_target(#'Command.Container.ContainerTarget'{name = Name, id = Id}) ->
+    NameBin = iolist_to_binary(Name),
+    IdBin = iolist_to_binary(Id),
     case NameBin of
         <<>> ->
             true = IdBin =/= <<>>,
             IdBin;
         _ ->
             NameBin
-    end.
+    end;
+container_target(undefined) ->
+    error(bad_container_target).
 
 -spec to_binary(binary() | list()) -> binary().
 to_binary(Value) when is_binary(Value) ->