apps/efka/src/iot/efka_iot_client.erl

@@ -189,13 +189,14 @@ handle_event({call, From}, dropped_message_count, _StateName, State = #state{dro
     {keep_state, State, [{reply, From, DroppedCount}]};
 
 %% 异步建立到服务器的连接
-handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State = #state{next_pkt_id = PktId}) ->
+handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) ->
     case connect_socket() of
         {ok, Socket} ->
-            AuthPacket = auth_packet(PktId),
+            {Ref, State1} = next_packet_id(State),
+            AuthPacket = auth_packet(Ref),
             ok = ssl:send(Socket, AuthPacket),
-            logger:debug("[efka_iot_client] send auth request, packet_id: ~p", [PktId]),
+            logger:debug("[efka_iot_client] send auth request, ref: ~p", [Ref]),
-            {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, State1#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
         {error, _Reason} ->
             schedule_reconnect(),
             {keep_state, State#state{socket = undefined}}
@@ -451,6 +452,10 @@ decode_pb_frame(Payload, MsgName) ->
             {error, {bad_protobuf, MsgName, Class, Reason}}
     end.
 
+-spec next_packet_id(#state{}) -> {pos_integer(), #state{}}.
+next_packet_id(State = #state{next_pkt_id = PacketId}) ->
+    {PacketId, State#state{next_pkt_id = PacketId + 1}}.
+
 -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}.
 connect_socket() ->
     {ok, Props} = application:get_env(efka, iot_server),
@@ -551,7 +556,7 @@ handle_container_request(Ref, Request, Socket) ->
 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)};
+    #{<<"action">> => <<"deploy">>, <<"task_id">> => TaskId, <<"params">> => decode_term_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}}}) ->
@@ -571,14 +576,14 @@ decode_container_target(#'Command.Container.ContainerTarget'{name = Name, id = I
 decode_container_target(undefined) ->
     #{<<"name">> => <<>>, <<"id">> => <<>>}.
 
--spec decode_json_payload(iodata()) -> term().
+-spec decode_term_payload(iodata()) -> term().
-decode_json_payload(Payload) ->
+decode_term_payload(Payload) ->
     Bin = iolist_to_binary(Payload),
-    try json:decode(Bin) of
+    try binary_to_term(Bin, [safe]) of
         Term ->
             Term
     catch
-        _:_ ->
+        error:_ ->
             Bin
     end.