From f0facb3691fd7e0f49e0862c90299c4245182d3d Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Mon, 27 Apr 2026 10:26:41 +0800 Subject: [PATCH] fix --- src/devtools/endpoint_kafka_test.erl | 4 +- src/devtools/eval_test.erl | 10 +--- src/endpoint/endpoint.erl | 77 ++++++++++++++++++------- src/endpoint/endpoint_buffer.erl | 40 ++++--------- src/endpoint/endpoint_http.erl | 6 +- src/endpoint/endpoint_kafka.erl | 6 +- src/endpoint/endpoint_mqtt.erl | 6 +- src/endpoint/endpoint_sup.erl | 20 ++++++- src/host/iot_host.erl | 44 ++++++++------ src/host/iot_host_sup.erl | 49 ++++++++++++---- src/iot_app.erl | 34 ++++++++--- src/iot_util.erl | 17 ++---- src/transport/http/endpoint_handler.erl | 4 +- src/transport/http/http_protocol.erl | 64 ++++++++++++++++---- src/transport/tcp/ssl_channel.erl | 13 ++++- 15 files changed, 255 insertions(+), 139 deletions(-) diff --git a/src/devtools/endpoint_kafka_test.erl b/src/devtools/endpoint_kafka_test.erl index ee5af19..2912d2d 100644 --- a/src/devtools/endpoint_kafka_test.erl +++ b/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, diff --git a/src/devtools/eval_test.erl b/src/devtools/eval_test.erl index 0268655..d2ea8ce 100644 --- a/src/devtools/eval_test.erl +++ b/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}. diff --git a/src/endpoint/endpoint.erl b/src/endpoint/endpoint.erl index 12070ee..dc9ff48 100644 --- a/src/endpoint/endpoint.erl +++ b/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. diff --git a/src/endpoint/endpoint_buffer.erl b/src/endpoint/endpoint_buffer.erl index 94f9cb8..68c39fa 100644 --- a/src/endpoint/endpoint_buffer.erl +++ b/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 -> diff --git a/src/endpoint/endpoint_http.erl b/src/endpoint/endpoint_http.erl index 2377e9e..875cb2b 100644 --- a/src/endpoint/endpoint_http.erl +++ b/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 diff --git a/src/endpoint/endpoint_kafka.erl b/src/endpoint/endpoint_kafka.erl index 10c9a38..4c59c58 100644 --- a/src/endpoint/endpoint_kafka.erl +++ b/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 diff --git a/src/endpoint/endpoint_mqtt.erl b/src/endpoint/endpoint_mqtt.erl index 7e64594..cafd84c 100644 --- a/src/endpoint/endpoint_mqtt.erl +++ b/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 diff --git a/src/endpoint/endpoint_sup.erl b/src/endpoint/endpoint_sup.erl index d998103..cd5c554 100644 --- a/src/endpoint/endpoint_sup.erl +++ b/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), diff --git a/src/host/iot_host.erl b/src/host/iot_host.erl index 8db63dc..29ae5f1 100644 --- a/src/host/iot_host.erl +++ b/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. diff --git a/src/host/iot_host_sup.erl b/src/host/iot_host_sup.erl index aa22324..1c81363 100644 --- a/src/host/iot_host_sup.erl +++ b/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(). diff --git a/src/iot_app.erl b/src/iot_app.erl index 6f6aae4..ce51d39 100644 --- a/src/iot_app.erl +++ b/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 diff --git a/src/iot_util.erl b/src/iot_util.erl index c53eb1f..a074693 100644 --- a/src/iot_util.erl +++ b/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). \ No newline at end of file + throw(Msg). diff --git a/src/transport/http/endpoint_handler.erl b/src/transport/http/endpoint_handler.erl index 51d8eaf..dc4ce9c 100644 --- a/src/transport/http/endpoint_handler.erl +++ b/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 -> diff --git a/src/transport/http/http_protocol.erl b/src/transport/http/http_protocol.erl index c33ef23..185e753 100644 --- a/src/transport/http/http_protocol.erl +++ b/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, <>, Req1}; + NAccData = <>, + 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, <>) + NAccData = <>, + 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. diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl index b616908..ba2e9b3 100644 --- a/src/transport/tcp/ssl_channel.erl +++ b/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 ->