From 8345a5fbc0c1bff2c477e5b91635bb5c26e1fb06 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Mon, 11 May 2026 20:35:44 +0800 Subject: [PATCH] fix endpoint --- apps/endpoint/src/endpoint.app.src | 1 - apps/endpoint/src/endpoint_tester.erl | 93 ++++++++++++++++++++++++++ apps/iot/src/http/endpoint_handler.erl | 85 ++++------------------- apps/iot/src/iot.app.src | 1 + 4 files changed, 108 insertions(+), 72 deletions(-) create mode 100644 apps/endpoint/src/endpoint_tester.erl diff --git a/apps/endpoint/src/endpoint.app.src b/apps/endpoint/src/endpoint.app.src index ac0ecb8..abbec16 100644 --- a/apps/endpoint/src/endpoint.app.src +++ b/apps/endpoint/src/endpoint.app.src @@ -9,7 +9,6 @@ ]}, {mod, {endpoint_app, []}}, {applications, [ - iot, emqtt, brod, hackney, diff --git a/apps/endpoint/src/endpoint_tester.erl b/apps/endpoint/src/endpoint_tester.erl new file mode 100644 index 0000000..262e080 --- /dev/null +++ b/apps/endpoint/src/endpoint_tester.erl @@ -0,0 +1,93 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2026, +%%% @doc +%%% +%%% @end +%%% Created : 11. 5月 2026 20:16 +%%%------------------------------------------------------------------- +-module(endpoint_tester). +-author("anlicheng"). +-include("endpoint.hrl"). + +%% API +-export([test/1]). + +%% http接口的测试 +test(#http_endpoint{url = Url, pool_size = PoolSize}) when is_integer(PoolSize), PoolSize > 0 -> + Body = <<"">>, + ContentType = "application/json", + case httpc:request(post, {Url, [], ContentType, Body}, [], []) of + {ok, _} -> + ok; + {error, Reason} -> + {error, Reason} + end; + +%% 测试mqtt +test(#mqtt_endpoint{host = Host, port = Port, username = Username, password = Password}) -> + %% 用户测试的client + ClientId = "mqtt_client_test:" ++ iot_util:rand_bytes(16), + Opts = [ + {owner, self()}, + {clientid, ClientId}, + {host, binary_to_list(Host)}, + {port, Port}, + {tcp_opts, []}, + {username, binary_to_list(Username)}, + {password, binary_to_list(Password)}, + {keepalive, 86400}, + {auto_ack, true}, + {connect_timeout, 5000}, + {proto_ver, v5}, + {retry_interval, 5000} + ], + + case emqtt:start_link(Opts) of + {ok, ConnPid} -> + case catch emqtt:connect(ConnPid, 5000) of + {ok, _} -> + emqtt:stop(ConnPid), + ok; + {error, _Reason} -> + {error, <<"connect mqtt server failed">>}; + _Error -> + emqtt:stop(ConnPid), + {error, <<"connect mqtt server failed">>} + end; + Other -> + logger:warning("[endpint_handler] test connect mqtt with options: ~p, get error: ~p", [Opts, Other]), + {error, <<"connect mqtt server failed">>} + end; + +%% 测试kafka +test(#kafka_endpoint{sasl_config = SaslConfig, bootstrap_servers = BootstrapServers, topic = Topic}) -> + BaseConfig = [ + {reconnect_cool_down_seconds, 5}, + {socket_options, [{keepalive, true}]} + ], + + ClientConfig = case SaslConfig of + {Mechanism, Username, Password} -> + [{sasl, {Mechanism, Username, Password}}|BaseConfig]; + undefined -> + BaseConfig + end, + ClientId = brod_client_test, + _ = catch brod:stop_client(ClientId), + + case catch brod:start_link_client(BootstrapServers, ClientId, ClientConfig) of + {ok, _ClientPid} -> + case brod:start_producer(ClientId, Topic, _ProducerConfig = []) of + ok -> + ok = brod:stop_client(ClientId), + ok; + {error, Reason} -> + logger:debug("[endpint_handler] start_producer: ~p, get error: ~p", [ClientId, Reason]), + _ = catch brod:stop_client(ClientId), + {error, <<"config kafka server failed">>} + end; + Error -> + logger:debug("[endpint_handler] start_client: ~p, get error: ~p", [ClientId, Error]), + {error, <<"config kafka server failed">>} + end. \ No newline at end of file diff --git a/apps/iot/src/http/endpoint_handler.erl b/apps/iot/src/http/endpoint_handler.erl index 25b6f3c..bfe6dc9 100644 --- a/apps/iot/src/http/endpoint_handler.erl +++ b/apps/iot/src/http/endpoint_handler.erl @@ -100,10 +100,8 @@ handle_request("POST", "/endpoint/restart", _, #{<<"id">> := Id}) when is_intege %% http接口的测试 handle_request("POST", "/endpoint/test", _, #{<<"protocol">> := <<"http">>, <<"config">> := #{<<"url">> := Url, <<"pool_size">> := PoolSize}}) when is_integer(PoolSize), PoolSize > 0 -> - Body = <<"">>, - ContentType = "application/json", - case httpc:request(post, {Url, [], ContentType, Body}, [], []) of - {ok, _} -> + case endpoint_tester:test(#http_endpoint{url = Url, pool_size = PoolSize}) of + ok -> {ok, 200, iot_util:json_data(<<"ok">>)}; {error, Reason} -> logger:debug("[endpint_handler] test http: ~p, error: ~p", [Url, Reason]), @@ -113,44 +111,12 @@ handle_request("POST", "/endpoint/test", _, #{<<"protocol">> := <<"http">>, <<"c %% 测试mqtt handle_request("POST", "/endpoint/test", _, #{<<"protocol">> := <<"mqtt">>, <<"config">> := Config}) -> case endpoint:parse_config(<<"mqtt">>, Config) of - {ok, #mqtt_endpoint{host = Host, port = Port, username = Username, password = Password}} -> - %% 用户测试的client - ClientId = "mqtt_client_test:" ++ iot_util:rand_bytes(16), - Opts = [ - {owner, self()}, - {clientid, ClientId}, - {host, binary_to_list(Host)}, - {port, Port}, - {tcp_opts, []}, - {username, binary_to_list(Username)}, - {password, binary_to_list(Password)}, - {keepalive, 86400}, - {auto_ack, true}, - {connect_timeout, 5000}, - {proto_ver, v5}, - {retry_interval, 5000} - ], - - case emqtt:start_link(Opts) of - {ok, ConnPid} -> - logger:debug("[endpint_handler] start connect, options: ~p", [Opts]), - case catch emqtt:connect(ConnPid, 5000) of - {ok, _} -> - logger:debug("[endpint_handler] connect success, pid: ~p", [ConnPid]), - emqtt:stop(ConnPid), - {ok, 200, iot_util:json_data(<<"ok">>)}; - {error, Reason} -> - logger:warning("[endpint_handler] connect get error: ~p", [Reason]), - emqtt:stop(ConnPid), - {ok, 200, iot_util:json_error(-1, <<"connect mqtt server failed">>)}; - Error -> - logger:warning("[endpint_handler] connect get error: ~p", [Error]), - emqtt:stop(ConnPid), - {ok, 200, iot_util:json_error(-1, <<"connect mqtt server failed">>)} - end; - Other -> - logger:warning("[endpint_handler] test connect mqtt with options: ~p, get error: ~p", [Opts, Other]), - {ok, 200, iot_util:json_error(-1, <<"connect mqtt server failed">>)} + {ok, MqttEndpoint = #mqtt_endpoint{}} -> + case endpoint_tester:test(MqttEndpoint) of + ok -> + {ok, 200, iot_util:json_data(<<"ok">>)}; + {error, Reason} -> + {ok, 200, iot_util:json_error(-1, Reason)} end; {error, Errors} -> {ok, 200, iot_util:json_error(-1, Errors)} @@ -159,35 +125,12 @@ handle_request("POST", "/endpoint/test", _, #{<<"protocol">> := <<"mqtt">>, <<"c %% 测试kafka handle_request("POST", "/endpoint/test", _, #{<<"protocol">> := <<"kafka">>, <<"config">> := Config}) -> case endpoint:parse_config(<<"kafka">>, Config) of - {ok, #kafka_endpoint{sasl_config = SaslConfig, bootstrap_servers = BootstrapServers, topic = Topic}} -> - BaseConfig = [ - {reconnect_cool_down_seconds, 5}, - {socket_options, [{keepalive, true}]} - ], - - ClientConfig = case SaslConfig of - {Mechanism, Username, Password} -> - [{sasl, {Mechanism, Username, Password}}|BaseConfig]; - undefined -> - BaseConfig - end, - ClientId = brod_client_test, - _ = catch brod:stop_client(ClientId), - - case catch brod:start_link_client(BootstrapServers, ClientId, ClientConfig) of - {ok, _ClientPid} -> - case brod:start_producer(ClientId, Topic, _ProducerConfig = []) of - ok -> - ok = brod:stop_client(ClientId), - {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 -> - logger:debug("[endpint_handler] start_client: ~p, get error: ~p", [ClientId, Error]), - {ok, 200, iot_util:json_error(-1, <<"config kafka server failed">>)} + {ok, KafkaEndpoint = #kafka_endpoint{}} -> + case endpoint_tester:test(KafkaEndpoint) of + ok -> + {ok, 200, iot_util:json_data(<<"ok">>)}; + {error, Reason} -> + {ok, 200, iot_util:json_error(-1, Reason)} end; {error, Errors} -> {ok, 200, iot_util:json_error(-1, Errors)} diff --git a/apps/iot/src/iot.app.src b/apps/iot/src/iot.app.src index d74f17d..f99b31a 100644 --- a/apps/iot/src/iot.app.src +++ b/apps/iot/src/iot.app.src @@ -6,6 +6,7 @@ {applications, [ sync, + endpoint, emqtt, eredis, ranch,