fix

src/devtools/endpoint_kafka_test.erl

@@ -11,9 +11,9 @@
 -include("endpoint.hrl").
 
 %% API
--export([start_test/0, test_consumer/0]).
+-export([start_manual/0, test_consumer/0]).
 
-start_test() ->
+start_manual() ->
     Name = endpoint:get_name(100),
     {ok, Pid} = endpoint_kafka:start_link(Name, #endpoint{
         id = 100,

src/devtools/eval_test.erl

@@ -13,13 +13,5 @@
 -export([test/0]).
 
 test() ->
-
-    {ok, Content} = file:read_file("/tmp/test.erl"),
-
-    {ok, Tokens, _} = erl_scan:string(binary_to_list(Content)),
-    {ok, ExprList} = erl_parse:parse_exprs(Tokens),
-
-    {value, F, _NewBindings} = erl_eval:exprs(ExprList, []),
-    F(#{name => <<"test">>}).
-
+    {error, disabled}.
 

src/endpoint/endpoint.erl

@@ -31,17 +31,17 @@ start_link(Endpoint = #endpoint{id = Id, config = #kafka_endpoint{}}) ->
     LocalName = get_name(Id),
     endpoint_kafka:start_link(LocalName, Endpoint).
 
--spec get_name(Id :: integer()) -> atom().
+-spec get_name(Id :: integer()) -> term().
 get_name(Id) when is_integer(Id) ->
-    list_to_atom("endpoint:" ++ integer_to_list(Id)).
+    {endpoint, Id}.
 
 -spec get_pid(Id :: integer()) -> undefined | pid().
 get_pid(Id) when is_integer(Id) ->
-    whereis(get_name(Id)).
+    gproc:whereis_name({n, l, get_name(Id)}).
 
--spec get_alias_name(Name :: binary()) -> atom().
+-spec get_alias_name(Name :: binary()) -> term().
 get_alias_name(Name) when is_binary(Name) ->
-    list_to_atom("endpoint:" ++ binary_to_list(Name)).
+    {endpoint_alias, Name}.
 
 -spec get_alias_pid(Name :: binary()) -> undefined | pid().
 get_alias_pid(Name) when is_binary(Name) ->
@@ -92,16 +92,27 @@ endpoint_record(#{<<"id">> := Id, <<"matcher">> := Matcher, <<"title">> := Title
 
 -spec parse_config(Protocol :: binary(), Config :: map()) -> {ok, #mqtt_endpoint{} | #kafka_endpoint{} | #http_endpoint{}} | {error, Errors :: [Error :: binary()]}.
 parse_config(<<"mqtt">>, #{<<"host">> := Host, <<"port">> := Port0, <<"client_id">> := ClientId, <<"username">> := Username, <<"password">> := Password, <<"topic">> := Topic, <<"qos">> := Qos}) ->
-    Port = if is_binary(Port0) -> binary_to_integer(Port0); is_integer(Port0) -> Port0 end,
+    {Port, PortErrors} = case parse_integer(Port0) of
+        {ok, ParsedPort} ->
+            {ParsedPort, []};
+        error ->
+            {undefined, [<<"port invalid">>]}
+    end,
 
-    Errors = lists:filtermap(fun(Term) ->
+    CheckTerms = case Port of
+        undefined ->
+            [{host, Host}, {username, Username}, {password, Password}, {topic, Topic}, {qos, Qos}];
+        _ ->
+            [{host, Host}, {port, Port}, {username, Username}, {password, Password}, {topic, Topic}, {qos, Qos}]
+    end,
+    Errors = PortErrors ++ lists:filtermap(fun(Term) ->
         case check_mqtt_argument(Term) of
             ok ->
                 false;
             {error, Error} ->
                 {true, Error}
         end
-                             end, [{host, Host}, {port, Port}, {username, Username}, {password, Password}, {topic, Topic}, {qos, Qos}]),
+                             end, CheckTerms),
     case Errors =:= [] of
         true ->
             {ok, #mqtt_endpoint{
@@ -181,8 +192,8 @@ parse_config(_, _) ->
 -spec parse_kafka_bootstrap_servers(BootstrapServers :: [binary()]) -> Servers :: [{Host :: string(), Port :: integer()}].
 parse_kafka_bootstrap_servers(BootstrapServers) when is_list(BootstrapServers) ->
     lists:map(fun(S) ->
-        [Host0, Port0] = binary:split(S, <<":">>),
-        {binary_to_list(Host0), binary_to_integer(Port0)}
+        {ok, Host, Port} = parse_kafka_bootstrap_server(S),
+        {Host, Port}
               end, BootstrapServers).
 
 -spec parse_kafka_mechanism(Mechanism0 :: binary()) -> atom().
@@ -231,17 +242,10 @@ check_kafka_argument({bootstrap_servers, BootstrapServers}) ->
     case is_list(BootstrapServers) andalso length(BootstrapServers) > 0 of
         true ->
             InvalidServers = lists:filtermap(fun(S) ->
-                case binary:split(S, <<":">>) of
-                    [Host0, Port0] ->
-                        Host = binary_to_list(Host0),
-                        Port = binary_to_integer(Port0),
-                        case not string:is_empty(Host) andalso (is_integer(Port) andalso Port > 0) of
-                            true ->
-                                false;
-                            false ->
-                                {true, S}
-                        end;
-                    _ ->
+                case parse_kafka_bootstrap_server(S) of
+                    {ok, _Host, _Port} ->
+                        false;
+                    error ->
                         {true, S}
                 end
                                              end, BootstrapServers),
@@ -322,3 +326,34 @@ check_mqtt_argument({qos, Qos}) ->
         false ->
             {error, <<"qos invalid">>}
     end.
+
+-spec parse_integer(term()) -> {ok, integer()} | error.
+parse_integer(Value) when is_integer(Value) ->
+    {ok, Value};
+parse_integer(Value) when is_binary(Value) ->
+    try binary_to_integer(Value) of
+        Int ->
+            {ok, Int}
+    catch
+        _:_ ->
+            error
+    end;
+parse_integer(_) ->
+    error.
+
+-spec parse_kafka_bootstrap_server(term()) -> {ok, string(), integer()} | error.
+parse_kafka_bootstrap_server(Server) when is_binary(Server) ->
+    case binary:split(Server, <<":">>) of
+        [Host0, Port0] ->
+            Host = binary_to_list(Host0),
+            case parse_integer(Port0) of
+                {ok, Port} when Host =/= [], Port > 0 ->
+                    {ok, Host, Port};
+                _ ->
+                    error
+            end;
+        _ ->
+            error
+    end;
+parse_kafka_bootstrap_server(_) ->
+    error.

src/endpoint/endpoint_buffer.erl

@@ -15,7 +15,7 @@
 -define(RETRY_INTERVAL, 5000).
 %% 最大重试次数，不包含首次发送
 -define(MAX_RETRY_TIMES, 3).
-%% 与 endpoint_outbox 的单条记录限制保持一致；fast path 也不能绕过该限制。
+%% 与 endpoint_outbox 的单条记录限制保持一致。
 -define(MAX_PAYLOAD_BYTES, 32 * 1024).
 
 -export([new/2, append/2, append_only/2, trigger_next/1, trigger_n/1, handle_timeout/4, cleanup/1, recover_inflight/1, ack/2, stat/1, resize/2]).
@@ -29,8 +29,6 @@
     outbox :: endpoint_outbox:outbox(),
     %% 当前待 ack 的数据及其重试定时器 #{Id => {TimerRef, Payload, RetryTimes, Source}}
     timer_map = #{} :: #{integer() => timer_entry()},
-    %% 内存 fast path 使用负数 id，避免与 outbox 的正整数 seq 冲突。
-    next_memory_id = -1 :: integer(),
     %% 窗口大小，允许最大的未确认消息数
     window_size = 10,
     %% 未确认的消息数
@@ -52,21 +50,17 @@ new(Endpoint = #endpoint{id = Id}, WindowSize) when is_integer(WindowSize), Wind
     #buffer{outbox = Outbox, endpoint = Endpoint, window_size = WindowSize}.
 
 -spec append(Payload :: binary(), Buffer :: #buffer{}) -> NBuffer :: #buffer{}.
-append(Payload, Buffer = #buffer{outbox = Outbox, window_size = WindowSize, flight_num = FlightNum}) when is_binary(Payload) ->
+append(Payload, Buffer = #buffer{}) when is_binary(Payload) ->
     case validate_payload_size(Payload, Buffer) of
         ok ->
-            case FlightNum < WindowSize andalso outbox_empty(Outbox) of
-                true ->
-                    dispatch_memory(Payload, Buffer);
-                false ->
-                    case append_to_outbox(Payload, Buffer) of
-                        {ok, NBuffer} ->
-                            trigger_next(NBuffer);
-                        {dropped, NBuffer} ->
-                            trigger_next(NBuffer);
-                        {error, NBuffer} ->
-                            NBuffer
-                    end
+            %% Persist before dispatch so a VM crash does not lose in-flight data.
+            case append_to_outbox(Payload, Buffer) of
+                {ok, NBuffer} ->
+                    trigger_next(NBuffer);
+                {dropped, NBuffer} ->
+                    trigger_next(NBuffer);
+                {error, NBuffer} ->
+                    NBuffer
             end;
         error ->
             Buffer
@@ -228,11 +222,6 @@ validate_payload_size(Payload, Buffer) ->
             error
     end.
 
--spec outbox_empty(endpoint_outbox:outbox()) -> boolean().
-outbox_empty(Outbox) ->
-    Stat = endpoint_outbox:stat(Outbox),
-    maps:get(write_seq, Stat, 0) =:= maps:get(acked_seq, Stat, 0).
-
 -spec append_to_outbox(binary(), buffer()) -> {ok | dropped | error, buffer()}.
 append_to_outbox(Payload, Buffer = #buffer{outbox = Outbox}) ->
     case endpoint_outbox:append(Payload, Outbox) of
@@ -247,15 +236,6 @@ append_to_outbox(Payload, Buffer = #buffer{outbox = Outbox}) ->
             {error, Buffer}
     end.
 
--spec dispatch_memory(binary(), buffer()) -> buffer().
-dispatch_memory(Payload, Buffer = #buffer{next_memory_id = Id, flight_num = FlightNum}) ->
-    ReceiverPid = self(),
-    ReceiverPid ! {next_data, Id, Payload},
-    schedule_retry(Id, Payload, 0, memory, Buffer#buffer{
-        next_memory_id = Id - 1,
-        flight_num = FlightNum + 1
-    }).
-
 -spec schedule_retry(integer(), binary(), non_neg_integer(), flight_source(), buffer()) -> buffer().
 schedule_retry(Id, Payload, RetryTimes, Source, Buffer = #buffer{timer_map = TimerMap})
         when is_integer(Id), is_binary(Payload), is_integer(RetryTimes), RetryTimes >= 0 ->

src/endpoint/endpoint_http.erl

@@ -30,10 +30,10 @@
 %%%===================================================================
 
 %% @doc Spawns the server and registers the local name (unique)
--spec(start_link(LocalName :: atom(), Endpoint :: #endpoint{}) ->
+-spec(start_link(LocalName :: term(), Endpoint :: #endpoint{}) ->
     {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
-start_link(LocalName, Endpoint = #endpoint{config = #http_endpoint{}}) when is_atom(LocalName) ->
-    gen_server:start_link({local, LocalName}, ?MODULE, [Endpoint], []).
+start_link(LocalName, Endpoint = #endpoint{config = #http_endpoint{}}) ->
+    gen_server:start_link({via, gproc, {n, l, LocalName}}, ?MODULE, [Endpoint], []).
 
 %%%===================================================================
 %%% gen_server callbacks

src/endpoint/endpoint_kafka.erl

@@ -34,10 +34,10 @@
 %%% API
 %%%===================================================================
 
--spec start_link(LocalName :: atom(), Endpoint :: #endpoint{}) ->
+-spec start_link(LocalName :: term(), Endpoint :: #endpoint{}) ->
     {ok, pid()} | ignore | {error, term()}.
-start_link(LocalName, Endpoint = #endpoint{}) when is_atom(LocalName) ->
-    gen_statem:start_link({local, LocalName}, ?MODULE, [Endpoint], []).
+start_link(LocalName, Endpoint = #endpoint{}) ->
+    gen_statem:start_link({via, gproc, {n, l, LocalName}}, ?MODULE, [Endpoint], []).
 
 %%%===================================================================
 %%% gen_statem callbacks

src/endpoint/endpoint_mqtt.erl

@@ -35,10 +35,10 @@
 %%% API
 %%%===================================================================
 
--spec start_link(LocalName :: atom(), Endpoint :: #endpoint{}) ->
+-spec start_link(LocalName :: term(), Endpoint :: #endpoint{}) ->
     {ok, pid()} | ignore | {error, term()}.
-start_link(LocalName, Endpoint = #endpoint{}) when is_atom(LocalName) ->
-    gen_statem:start_link({local, LocalName}, ?MODULE, [Endpoint], []).
+start_link(LocalName, Endpoint = #endpoint{}) ->
+    gen_statem:start_link({via, gproc, {n, l, LocalName}}, ?MODULE, [Endpoint], []).
 
 %%%===================================================================
 %%% gen_statem callbacks

src/endpoint/endpoint_sup.erl

@@ -59,8 +59,24 @@ ensured_endpoint_started(Endpoint = #endpoint{}) ->
 -spec delete_endpoint(Id :: integer()) -> ok | {error, Reason :: any()}.
 delete_endpoint(Id) when is_integer(Id) ->
     Name = endpoint:get_name(Id),
-    supervisor:terminate_child(?MODULE, Name),
-    supervisor:delete_child(?MODULE, Name).
+    case supervisor:terminate_child(?MODULE, Name) of
+        ok ->
+            delete_endpoint_child(Name);
+        {error, not_found} ->
+            delete_endpoint_child(Name);
+        {error, Reason} ->
+            {error, Reason}
+    end.
+
+delete_endpoint_child(Name) ->
+    case supervisor:delete_child(?MODULE, Name) of
+        ok ->
+            ok;
+        {error, not_found} ->
+            ok;
+        {error, Reason} ->
+            {error, Reason}
+    end.
 
 child_spec(Endpoint = #endpoint{id = Id}) ->
     Name = endpoint:get_name(Id),

src/host/iot_host.erl

@@ -50,20 +50,19 @@
 -spec get_pid(UUID :: binary()) -> undefined | pid().
 get_pid(UUID) when is_binary(UUID) ->
     Name = get_name(UUID),
-    whereis(Name).
+    gproc:whereis_name({n, l, Name}).
 
--spec get_name(UUID :: binary()) -> atom().
+-spec get_name(UUID :: binary()) -> term().
 get_name(UUID) when is_binary(UUID) ->
-    binary_to_atom(<<"iot_host:", UUID/binary>>).
+    {iot_host, UUID}.
 
--spec get_alias_name(HostId :: integer()) -> atom().
+-spec get_alias_name(HostId :: integer()) -> term().
 get_alias_name(HostId0) when is_integer(HostId0) ->
-    HostId = integer_to_binary(HostId0),
-    binary_to_atom(<<"iot_host_id:", HostId/binary>>).
+    {iot_host_id, HostId0}.
 
 -spec kill(UUID :: binary()) -> no_return().
 kill(UUID) when is_binary(UUID) ->
-    case whereis(get_name(UUID)) of
+    case get_pid(UUID) of
         undefined ->
             ok;
         Pid ->
@@ -152,8 +151,8 @@ heartbeat(Pid) when is_pid(Pid) ->
 %% @doc Creates a gen_statem process which calls Module:init/1 to
 %% initialize. To ensure a synchronized start-up procedure, this
 %% function does not return until Module:init/1 has returned.
-start_link(Name, UUID) when is_atom(Name), is_binary(UUID) ->
-    gen_statem:start_link({local, Name}, ?MODULE, [UUID], []).
+start_link(Name, UUID) when is_binary(UUID) ->
+    gen_statem:start_link({via, gproc, {n, l, Name}}, ?MODULE, [UUID], []).
 
 %%%===================================================================
 %%% gen_statem callbacks
@@ -306,15 +305,7 @@ handle_event(cast, heartbeat, _, State = #state{heartbeat_counter = HeartbeatCou
 %% 没有收到心跳包，主机下线, 设备状态不变
 handle_event(info, {timeout, _, heartbeat_ticker}, _, State = #state{uuid = UUID, heartbeat_counter = 0, channel_pid = ChannelPid}) ->
     logger:warning("[iot_host] uuid: ~p, heartbeat lost, devices will unknown", [UUID]),
-    {ok, #host_info{status = Status}} = iot_api_client:get_host_by_uuid(UUID),
-    case Status of
-        ?HOST_NOT_JOINED ->
-            logger:debug("[iot_host] host: ~p, host_maybe_offline, host not joined, can not change to offline", [UUID]);
-        ?HOST_OFFLINE ->
-            logger:debug("[iot_host] host: ~p, host_maybe_offline, host now is offline, do nothing", [UUID]);
-        ?HOST_ONLINE ->
-            iot_api_client:change_host_status(UUID, ?HOST_OFFLINE)
-    end,
+    maybe_mark_host_offline(UUID),
 
     %% 关闭channel，主机需要重新连接，才能保存状态的一致
     is_pid(ChannelPid) andalso ssl_channel:stop(ChannelPid, closed),
@@ -387,3 +378,20 @@ flush_reply(Ref) ->
     after 0 ->
         ok
     end.
+
+-spec maybe_mark_host_offline(binary()) -> ok.
+maybe_mark_host_offline(UUID) ->
+    case iot_api_client:get_host_by_uuid(UUID) of
+        {ok, #host_info{status = ?HOST_NOT_JOINED}} ->
+            logger:debug("[iot_host] host: ~p, host_maybe_offline, host not joined, can not change to offline", [UUID]),
+            ok;
+        {ok, #host_info{status = ?HOST_OFFLINE}} ->
+            logger:debug("[iot_host] host: ~p, host_maybe_offline, host now is offline, do nothing", [UUID]),
+            ok;
+        {ok, #host_info{status = ?HOST_ONLINE}} ->
+            _ = iot_api_client:change_host_status(UUID, ?HOST_OFFLINE),
+            ok;
+        Other ->
+            logger:warning("[iot_host] host: ~p, load status failed while marking offline: ~p", [UUID, Other]),
+            ok
+    end.

src/host/iot_host_sup.erl

@@ -24,14 +24,18 @@ ensured_host_started(UUID) when is_binary(UUID) ->
     case iot_host:get_pid(UUID) of
         undefined ->
             %% 尝试删除下host对应的信息
-            delete_host(UUID),
-            case supervisor:start_child(?MODULE, child_spec(UUID)) of
-                {ok, Pid} when is_pid(Pid) ->
-                    {ok, Pid};
-                {error, {'already_started', Pid}} when is_pid(Pid) ->
-                    {ok, Pid};
-                {error, Error} ->
-                    {error, Error}
+            case delete_host(UUID) of
+                ok ->
+                    case supervisor:start_child(?MODULE, child_spec(UUID)) of
+                        {ok, Pid} when is_pid(Pid) ->
+                            {ok, Pid};
+                        {error, {'already_started', Pid}} when is_pid(Pid) ->
+                            {ok, Pid};
+                        {error, Error} ->
+                            {error, Error}
+                    end;
+                {error, Reason} ->
+                    {error, Reason}
             end;
         Pid when is_pid(Pid) ->
             {ok, Pid}
@@ -39,14 +43,35 @@ ensured_host_started(UUID) when is_binary(UUID) ->
 
 delete_host(UUID) when is_binary(UUID) ->
     Id = iot_host:get_name(UUID),
-    ok = supervisor:terminate_child(?MODULE, Id),
+    case supervisor:terminate_child(?MODULE, Id) of
+        ok ->
+            delete_host_child(Id, UUID);
+        {error, not_found} ->
+            delete_host_child(Id, UUID);
+        {error, Reason} ->
+            logger:warning("[iot_host_sup] terminate host: ~p failed: ~p", [UUID, Reason]),
+            {error, Reason}
+    end.
+
+delete_host_child(Id, UUID) ->
     case supervisor:delete_child(?MODULE, Id) of
+        ok ->
+            ok;
+        {error, not_found} ->
+            ok;
         {error, running} ->
             %% ensure killed then delete again
             iot_host:kill(UUID),
-            supervisor:delete_child(?MODULE, Id);
-        _ ->
-            ok
+            case supervisor:delete_child(?MODULE, Id) of
+                ok ->
+                    ok;
+                {error, not_found} ->
+                    ok;
+                {error, Reason} ->
+                    {error, Reason}
+            end;
+        {error, Reason} ->
+            {error, Reason}
     end.
 
 -spec child_spec(UUID :: binary()) -> map().

src/iot_app.erl

@@ -13,18 +13,32 @@ start(_StartType, _StartArgs) ->
     io:setopts([{encoding, unicode}]),
     %% 加速内存的回收
     erlang:system_flag(fullsweep_after, 16),
-    %% 启动mnesia数据库
-    start_mnesia(),
+    try
+        %% 启动mnesia数据库
+        start_mnesia(),
 
-    %% 启动http服务
-    start_http_server(),
+        %% 启动http服务。当前 supervisor 初始化依赖本机 simulator API，因此保留此顺序。
+        start_http_server(),
 
-    %% 启动ssl服务
-    start_ssl_server(),
+        %% 启动ssl服务
+        start_ssl_server(),
 
-    iot_sup:start_link().
+        case iot_sup:start_link() of
+            {ok, _Pid} = Ok ->
+                Ok;
+            Error ->
+                stop_started_services(),
+                Error
+        end
+    catch
+        Class:Reason:Stack ->
+            stop_started_services(),
+            logger:warning("[iot_app] start failed, class: ~p, reason: ~p, stack: ~p", [Class, Reason, Stack]),
+            {error, Reason}
+    end.
 
 stop(_State) ->
+    stop_started_services(),
     ok.
 
 %% internal functions
@@ -96,6 +110,12 @@ start_ssl_server() ->
     {ok, _} = ranch:start_listener(ssl_server, ranch_ssl, TransOpts, ssl_channel, []),
     logger:debug("[iot_app] the ssl server start at: ~p", [Port]).
 
+stop_started_services() ->
+    _ = cowboy:stop_listener(http_listener),
+    _ = ranch:stop_listener(ssl_server),
+    _ = mnesia:stop(),
+    ok.
+
 -spec ensure_mnesia_schema() -> any().
 ensure_mnesia_schema() ->
     case mnesia:system_info(use_dir) of

src/iot_util.erl

@@ -117,18 +117,9 @@ hex(N) ->
 
 %% 转换映射器
 -spec parse_mapper(Mapper :: binary() | string()) -> error | {ok, F :: fun((binary(), any()) -> any())}.
-parse_mapper(Mapper) when is_binary(Mapper) ->
-    parse_mapper(binary_to_list(Mapper));
-parse_mapper(Mapper) when is_list(Mapper) ->
-    {ok, Tokens, _} = erl_scan:string(Mapper),
-    {ok, ExprList} = erl_parse:parse_exprs(Tokens),
-    {value, F, _} = erl_eval:exprs(ExprList, []),
-    case is_function(F, 2) orelse is_function(F, 3) of
-        true ->
-            {ok, F};
-        false ->
-            error
-    end.
+parse_mapper(_Mapper) ->
+    %% Dynamic Erlang eval is intentionally disabled in runtime code.
+    error.
 
 -spec float_to_binary(Num :: number(), integer()) -> binary().
 float_to_binary(V, _) when is_integer(V) ->
@@ -142,4 +133,4 @@ assert(true, _) ->
 assert(false, F) when is_function(F) ->
     F();
 assert(false, Msg) ->
-    throw(Msg).
+    throw(Msg).

src/transport/http/endpoint_handler.erl

@@ -171,7 +171,8 @@ handle_request("POST", "/endpoint/test", _, #{<<"protocol">> := <<"kafka">>, <<"
                                undefined ->
                                    BaseConfig
                            end,
-            ClientId = list_to_atom("brod_client_test:" ++ iot_util:rand_bytes(16)),
+            ClientId = brod_client_test,
+            _ = catch brod:stop_client(ClientId),
 
             case catch brod:start_link_client(BootstrapServers, ClientId, ClientConfig) of
                 {ok, _ClientPid} ->
@@ -181,6 +182,7 @@ handle_request("POST", "/endpoint/test", _, #{<<"protocol">> := <<"kafka">>, <<"
                             {ok, 200, iot_util:json_data(<<"ok">>)};
                         {error, Reason} ->
                             logger:debug("[endpint_handler] start_producer: ~p, get error: ~p", [ClientId, Reason]),
+                            _ = catch brod:stop_client(ClientId),
                             {ok, 200, iot_util:json_error(-1, <<"config kafka server failed">>)}
                     end;
                 Error ->

src/transport/http/http_protocol.erl

@@ -12,14 +12,25 @@
 %% API
 -export([init/2]).
 
+-define(MAX_BODY_BYTES, 10 * 1024 * 1024).
+
 init(Req0, Opts = [Mod|_]) ->
     ok = iot_log:set_metadata(),
     Method = binary_to_list(cowboy_req:method(Req0)),
     Path = binary_to_list(cowboy_req:path(Req0)),
     GetParams0 = cowboy_req:parse_qs(Req0),
     GetParams = maps:from_list(GetParams0),
-    {ok, PostParams, Req1} = parse_body(Req0),
+    case parse_body(Req0) of
+        {ok, PostParams, Req1} ->
+            handle_request(Mod, Method, Path, GetParams, PostParams, Req1, Opts);
+        {error, StatusCode, Resp, Req1} ->
+            Req2 = cowboy_req:reply(StatusCode, #{
+                <<"Content-Type">> => <<"application/json;charset=utf-8">>
+            }, Resp, Req1),
+            {ok, Req2, Opts}
+    end.
 
+handle_request(Mod, Method, Path, GetParams, PostParams, Req1, Opts) ->
     try Mod:handle_request(Method, Path, GetParams, PostParams) of
         {ok, StatusCode, Resp} ->
             logger:debug("[http_protocol] request path: ~p, get_params: ~p, post_params: ~p, response: ~ts",
@@ -61,17 +72,20 @@ parse_body(Req0) ->
     ContentType = cowboy_req:header(<<"content-type">>, Req0),
     case ContentType of
         <<"application/json", _/binary>> ->
-            {ok, Body, Req1} = read_body(Req0),
-            case Body =/= <<"">> of
-                 true ->
-                    {ok, catch json:decode(Body), Req1};
-                false ->
-                    {ok, #{}, Req1}
+            case read_body(Req0) of
+                {ok, Body, Req1} ->
+                    decode_json_body(Body, Req1);
+                {error, payload_too_large, Req1} ->
+                    {error, 413, iot_util:json_error(413, <<"payload too large">>), Req1}
             end;
         <<"application/x-www-form-urlencoded">> ->
-            {ok, PostParams0, Req1} = cowboy_req:read_urlencoded_body(Req0),
-            PostParams = maps:from_list(PostParams0),
-            {ok, PostParams, Req1};
+            case cowboy_req:read_urlencoded_body(Req0) of
+                {ok, PostParams0, Req1} ->
+                    PostParams = maps:from_list(PostParams0),
+                    {ok, PostParams, Req1};
+                {more, _PostParams0, Req1} ->
+                    {error, 413, iot_util:json_error(413, <<"payload too large">>), Req1}
+            end;
         _ ->
             {ok, #{}, Req0}
     end.
@@ -82,7 +96,33 @@ read_body(Req) ->
 read_body(Req, AccData) ->
     case cowboy_req:read_body(Req) of
         {ok, Data, Req1} ->
-            {ok, <<AccData/binary, Data/binary>>, Req1};
+            NAccData = <<AccData/binary, Data/binary>>,
+            case byte_size(NAccData) =< ?MAX_BODY_BYTES of
+                true ->
+                    {ok, NAccData, Req1};
+                false ->
+                    {error, payload_too_large, Req1}
+            end;
         {more, Data, Req1} ->
-            read_body(Req1, <<AccData/binary, Data/binary>>)
+            NAccData = <<AccData/binary, Data/binary>>,
+            case byte_size(NAccData) =< ?MAX_BODY_BYTES of
+                true ->
+                    read_body(Req1, NAccData);
+                false ->
+                    {error, payload_too_large, Req1}
+            end
+    end.
+
+decode_json_body(<<"">>, Req) ->
+    {ok, #{}, Req};
+decode_json_body(Body, Req) ->
+    case catch json:decode(Body) of
+        {'EXIT', _} ->
+            {error, 400, iot_util:json_error(400, <<"invalid json">>), Req};
+        {error, _} ->
+            {error, 400, iot_util:json_error(400, <<"invalid json">>), Req};
+        Decoded when is_map(Decoded); is_list(Decoded) ->
+            {ok, Decoded, Req};
+        _ ->
+            {error, 400, iot_util:json_error(400, <<"invalid json">>), Req}
     end.

src/transport/tcp/ssl_channel.erl

@@ -142,13 +142,19 @@ handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, h
 
 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
+    case catch binary_to_term(PacketBin, [safe]) of
         {request, PacketId, Body} ->
             handle_request_frame(PacketId, Body, Transport, Socket, State);
         {message, Body} ->
             handle_message_frame(Body, HostPid, State);
         {response, PacketId, Response} ->
-            handle_response_frame(PacketId, Response, Inflight, State)
+            handle_response_frame(PacketId, Response, Inflight, State);
+        {'EXIT', Reason} ->
+            logger:warning("[ssl_channel] invalid packet: ~p", [Reason]),
+            {stop, bad_packet, State};
+        Other ->
+            logger:warning("[ssl_channel] unsupported packet: ~p", [Other]),
+            {stop, bad_packet, State}
     end;
 
 handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) ->
@@ -300,7 +306,8 @@ decode_reply(_Reply) ->
     ok | {error, Reason :: binary()}.
 auth(Token, UUID, Timestamp) when is_binary(Token), is_binary(UUID), is_integer(Timestamp) ->
     %% 1分钟内有效
-    case iot_util:current_time() - Timestamp =< 60 of
+    Now = iot_util:current_time(),
+    case Timestamp =< Now andalso Now - Timestamp =< 60 of
         true ->
             case efka_client_store:auth(UUID, Token) of
                 true ->