fix endpoint
This commit is contained in:
parent
1dd217b02f
commit
8345a5fbc0
@ -9,7 +9,6 @@
|
||||
]},
|
||||
{mod, {endpoint_app, []}},
|
||||
{applications, [
|
||||
iot,
|
||||
emqtt,
|
||||
brod,
|
||||
hackney,
|
||||
|
||||
93
apps/endpoint/src/endpoint_tester.erl
Normal file
93
apps/endpoint/src/endpoint_tester.erl
Normal file
@ -0,0 +1,93 @@
|
||||
%%%-------------------------------------------------------------------
|
||||
%%% @author anlicheng
|
||||
%%% @copyright (C) 2026, <COMPANY>
|
||||
%%% @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.
|
||||
@ -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)}
|
||||
|
||||
@ -6,6 +6,7 @@
|
||||
{applications,
|
||||
[
|
||||
sync,
|
||||
endpoint,
|
||||
emqtt,
|
||||
eredis,
|
||||
ranch,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user