721 lines
24 KiB
Erlang
721 lines
24 KiB
Erlang
%%--------------------------------------------------------------------
|
|
%% Copyright (c) 2022-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
%%--------------------------------------------------------------------
|
|
|
|
-module(emqx_bridge_hstreamdb_SUITE).
|
|
|
|
-compile(nowarn_export_all).
|
|
-compile(export_all).
|
|
|
|
-include_lib("emqx_bridge_hstreamdb.hrl").
|
|
-include_lib("eunit/include/eunit.hrl").
|
|
-include_lib("common_test/include/ct.hrl").
|
|
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
|
|
|
|
% SQL definitions
|
|
|
|
-define(STREAM, "demo_stream").
|
|
%% could not be "stream" in Production Environment
|
|
%% especially not in hstreamdb_sql CLI client
|
|
|
|
-define(REPLICATION_FACTOR, 1).
|
|
|
|
%% in seconds
|
|
-define(BACKLOG_RETENTION_SECOND, (24 * 60 * 60)).
|
|
-define(SHARD_COUNT, 1).
|
|
|
|
-define(BRIDGE_NAME, <<"hstreamdb_demo_bridge">>).
|
|
-define(RECORD_TEMPLATE,
|
|
"{ \"temperature\": ${payload.temperature}, \"humidity\": ${payload.humidity} }"
|
|
).
|
|
|
|
-define(POOL_SIZE, 8).
|
|
-define(BATCH_SIZE, 10).
|
|
-define(GRPC_TIMEOUT, "1s").
|
|
|
|
-define(WORKER_POOL_SIZE, 4).
|
|
|
|
-define(WITH_CLIENT(Process),
|
|
Client = connect_direct_hstream(_Name = test_c, Config),
|
|
Process,
|
|
ok = disconnect(Client)
|
|
).
|
|
|
|
%% How to run it locally (all commands are run in $PROJ_ROOT dir):
|
|
%% A: run ct on host
|
|
%% 1. Start all deps services
|
|
%% ```bash
|
|
%% sudo docker compose -f .ci/docker-compose-file/docker-compose.yaml \
|
|
%% -f .ci/docker-compose-file/docker-compose-hstreamdb.yaml \
|
|
%% -f .ci/docker-compose-file/docker-compose-toxiproxy.yaml \
|
|
%% up --build
|
|
%% ```
|
|
%%
|
|
%% 2. Run use cases with special environment variables
|
|
%% 6570 is toxiproxy exported port.
|
|
%% Local:
|
|
%% ```bash
|
|
%% HSTREAMDB_HOST=$REAL_TOXIPROXY_IP HSTREAMDB_PORT=6570 \
|
|
%% PROXY_HOST=$REAL_TOXIPROXY_IP PROXY_PORT=6570 \
|
|
%% ./rebar3 as test ct -c -v --readable true --name ct@127.0.0.1 \
|
|
%% --suite apps/emqx_bridge_hstreamdb/test/emqx_bridge_hstreamdb_SUITE.erl
|
|
%% ```
|
|
%%
|
|
%% B: run ct in docker container
|
|
%% run script:
|
|
%% ```bash
|
|
%% ./scripts/ct/run.sh --ci --app apps/emqx_bridge_hstreamdb/ -- \
|
|
%% --name 'test@127.0.0.1' -c -v --readable true \
|
|
%% --suite apps/emqx_bridge_hstreamdb/test/emqx_bridge_hstreamdb_SUITE.erl
|
|
%% ````
|
|
|
|
%%------------------------------------------------------------------------------
|
|
%% CT boilerplate
|
|
%%------------------------------------------------------------------------------
|
|
|
|
all() ->
|
|
[
|
|
{group, sync}
|
|
].
|
|
|
|
groups() ->
|
|
TCs = emqx_common_test_helpers:all(?MODULE),
|
|
NonBatchCases = [t_write_timeout],
|
|
BatchingGroups = [{group, with_batch}, {group, without_batch}],
|
|
[
|
|
{sync, BatchingGroups},
|
|
{with_batch, TCs -- NonBatchCases},
|
|
{without_batch, TCs}
|
|
].
|
|
|
|
init_per_group(sync, Config) ->
|
|
[{query_mode, sync} | Config];
|
|
init_per_group(with_batch, Config0) ->
|
|
Config = [{enable_batch, true} | Config0],
|
|
common_init(Config);
|
|
init_per_group(without_batch, Config0) ->
|
|
Config = [{enable_batch, false} | Config0],
|
|
common_init(Config);
|
|
init_per_group(_Group, Config) ->
|
|
Config.
|
|
|
|
end_per_group(Group, Config) when Group =:= with_batch; Group =:= without_batch ->
|
|
connect_and_delete_stream(Config),
|
|
ProxyHost = ?config(proxy_host, Config),
|
|
ProxyPort = ?config(proxy_port, Config),
|
|
emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
|
|
ok;
|
|
end_per_group(_Group, _Config) ->
|
|
ok.
|
|
|
|
init_per_suite(Config) ->
|
|
Config.
|
|
|
|
end_per_suite(_Config) ->
|
|
emqx_mgmt_api_test_util:end_suite(),
|
|
ok = emqx_common_test_helpers:stop_apps([emqx_bridge, emqx_resource, emqx_conf, hstreamdb_erl]),
|
|
ok.
|
|
|
|
init_per_testcase(t_to_hrecord_failed, Config) ->
|
|
init_per_testcase_common(),
|
|
meck:new([hstreamdb], [passthrough, no_history, no_link]),
|
|
meck:expect(hstreamdb, to_record, fun(_, _, _) -> error(trans_to_hrecord_failed) end),
|
|
Config;
|
|
init_per_testcase(_Testcase, Config) ->
|
|
init_per_testcase_common(),
|
|
%% drop stream and will create a new one in common_init/1
|
|
%% TODO: create a new stream for each test case
|
|
delete_bridge(Config),
|
|
snabbkaffe:start_trace(),
|
|
Config.
|
|
|
|
init_per_testcase_common() ->
|
|
emqx_bridge_v2_testlib:delete_all_bridges_and_connectors().
|
|
|
|
end_per_testcase(t_to_hrecord_failed, _Config) ->
|
|
meck:unload([hstreamdb]);
|
|
end_per_testcase(_Testcase, Config) ->
|
|
ProxyHost = ?config(proxy_host, Config),
|
|
ProxyPort = ?config(proxy_port, Config),
|
|
emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
|
|
ok = snabbkaffe:stop(),
|
|
delete_bridge(Config),
|
|
ok.
|
|
|
|
%%------------------------------------------------------------------------------
|
|
%% Testcases
|
|
%%------------------------------------------------------------------------------
|
|
|
|
t_setup_via_config_and_publish(Config) ->
|
|
?assertMatch(
|
|
{ok, _},
|
|
create_bridge(Config)
|
|
),
|
|
Data = rand_data(),
|
|
?check_trace(
|
|
begin
|
|
?wait_async_action(
|
|
?assertEqual(ok, send_message(Config, Data)),
|
|
#{?snk_kind := hstreamdb_connector_query_append_return},
|
|
10_000
|
|
),
|
|
ok
|
|
end,
|
|
fun(Trace0) ->
|
|
Trace = ?of_kind(hstreamdb_connector_query_append_return, Trace0),
|
|
lists:foreach(
|
|
fun(EachTrace) ->
|
|
case ?config(enable_batch, Config) of
|
|
true ->
|
|
?assertMatch(#{result := ok, is_batch := true}, EachTrace);
|
|
false ->
|
|
?assertMatch(
|
|
#{result := #{'batchId' := _}, is_batch := false}, EachTrace
|
|
)
|
|
end
|
|
end,
|
|
Trace
|
|
),
|
|
ok
|
|
end
|
|
),
|
|
ok.
|
|
|
|
t_setup_via_http_api_and_publish(Config) ->
|
|
BridgeType = ?config(hstreamdb_bridge_type, Config),
|
|
Name = ?config(hstreamdb_name, Config),
|
|
HStreamDBConfig0 = ?config(hstreamdb_config, Config),
|
|
HStreamDBConfig = HStreamDBConfig0#{
|
|
<<"name">> => Name,
|
|
<<"type">> => BridgeType
|
|
},
|
|
?assertMatch(
|
|
{ok, _},
|
|
create_bridge_http(HStreamDBConfig)
|
|
),
|
|
Data = rand_data(),
|
|
?check_trace(
|
|
begin
|
|
?wait_async_action(
|
|
?assertEqual(ok, send_message(Config, Data)),
|
|
#{?snk_kind := hstreamdb_connector_query_append_return},
|
|
10_000
|
|
),
|
|
ok
|
|
end,
|
|
fun(Trace) ->
|
|
lists:foreach(
|
|
fun(EachTrace) ->
|
|
case ?config(enable_batch, Config) of
|
|
true ->
|
|
?assertMatch(#{result := ok, is_batch := true}, EachTrace);
|
|
false ->
|
|
?assertMatch(
|
|
#{result := #{'batchId' := _}, is_batch := false}, EachTrace
|
|
)
|
|
end
|
|
end,
|
|
?of_kind(hstreamdb_connector_query_append_return, Trace)
|
|
),
|
|
ok
|
|
end
|
|
),
|
|
ok.
|
|
|
|
t_get_status(Config) ->
|
|
?assertMatch(
|
|
{ok, _},
|
|
create_bridge(Config)
|
|
),
|
|
ProxyPort = ?config(proxy_port, Config),
|
|
ProxyHost = ?config(proxy_host, Config),
|
|
ProxyName = ?config(proxy_name, Config),
|
|
|
|
health_check_resource_ok(Config),
|
|
emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
|
|
health_check_resource_down(Config)
|
|
end),
|
|
ok.
|
|
|
|
t_create_disconnected(Config) ->
|
|
ProxyPort = ?config(proxy_port, Config),
|
|
ProxyHost = ?config(proxy_host, Config),
|
|
ProxyName = ?config(proxy_name, Config),
|
|
|
|
?check_trace(
|
|
emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
|
|
?assertMatch({ok, _}, create_bridge(Config))
|
|
end),
|
|
fun(Trace) ->
|
|
?assertMatch(
|
|
[#{error := client_not_alive}],
|
|
?of_kind(hstreamdb_connector_start_failed, Trace)
|
|
),
|
|
ok
|
|
end
|
|
),
|
|
%% TODO: Investigate why reconnection takes at least 5 seconds during ct.
|
|
%% While in practical applications, recovers to the 'connected' state
|
|
%% within 3 seconds after toxiproxy being enabled.'"
|
|
%% timer:sleep(10000),
|
|
restart_resource(Config),
|
|
health_check_resource_ok(Config),
|
|
ok.
|
|
|
|
t_write_failure(Config) ->
|
|
ProxyName = ?config(proxy_name, Config),
|
|
ProxyPort = ?config(proxy_port, Config),
|
|
ProxyHost = ?config(proxy_host, Config),
|
|
QueryMode = ?config(query_mode, Config),
|
|
Data = rand_data(),
|
|
{{ok, _}, {ok, _}} =
|
|
?wait_async_action(
|
|
create_bridge(Config),
|
|
#{?snk_kind := resource_connected_enter},
|
|
20_000
|
|
),
|
|
emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
|
|
health_check_resource_down(Config),
|
|
case QueryMode of
|
|
sync ->
|
|
%% Error (call timeout) is expected for both with_batch and without_batch.
|
|
%% `health_check_resource_down(Config)` above calls health check and asserts
|
|
%% that resource is already down.
|
|
%% So, emqx_resource_manager updates it state to disconnected before returning health_check result.
|
|
%% After that, emqx_resource_buffer_worker reads resource state and doesn't even attempt calling
|
|
%% hstreamdb connector, since it is disconnected, see: emqx_resource_buffer_worker.erl:1163:
|
|
%% ```
|
|
%% do_call_query(_QM, _Id, _Index, _Ref, _Query, _QueryOpts, _Data) ->
|
|
%% ?RESOURCE_ERROR(not_connected, "resource not connected").
|
|
%% ```
|
|
?assertMatch({error, _}, send_message(Config, Data));
|
|
async ->
|
|
%% TODO: async mode is not supported yet,
|
|
%% but it will return ok if calling emqx_resource_buffer_worker:async_query/3,
|
|
?assertMatch(
|
|
ok,
|
|
send_message(Config, Data)
|
|
)
|
|
end
|
|
end),
|
|
ok.
|
|
|
|
t_simple_query(Config) ->
|
|
BatchSize = batch_size(Config),
|
|
?assertMatch(
|
|
{ok, _},
|
|
create_bridge(Config)
|
|
),
|
|
Type = ?config(hstreamdb_bridge_type, Config),
|
|
Name = ?config(hstreamdb_name, Config),
|
|
ActionId = emqx_bridge_v2:id(Type, Name),
|
|
Requests = gen_batch_req(BatchSize, ActionId),
|
|
?check_trace(
|
|
begin
|
|
?wait_async_action(
|
|
lists:foreach(
|
|
fun(Request) ->
|
|
?assertEqual(ok, query_resource(Config, Request))
|
|
end,
|
|
Requests
|
|
),
|
|
#{?snk_kind := hstreamdb_connector_query_append_return},
|
|
10_000
|
|
)
|
|
end,
|
|
fun(Trace) ->
|
|
lists:foreach(
|
|
fun(EachTrace) ->
|
|
case ?config(enable_batch, Config) of
|
|
true ->
|
|
?assertMatch(#{result := ok, is_batch := true}, EachTrace);
|
|
false ->
|
|
?assertMatch(
|
|
#{result := #{'batchId' := _}, is_batch := false}, EachTrace
|
|
)
|
|
end
|
|
end,
|
|
?of_kind(hstreamdb_connector_query_append_return, Trace)
|
|
),
|
|
ok
|
|
end
|
|
),
|
|
ok.
|
|
|
|
t_to_hrecord_failed(Config) ->
|
|
QueryMode = ?config(query_mode, Config),
|
|
?assertMatch(
|
|
{ok, _},
|
|
create_bridge(Config)
|
|
),
|
|
Result = send_message(Config, #{}),
|
|
case QueryMode of
|
|
sync ->
|
|
?assertMatch(
|
|
{error, {unrecoverable_error, failed_to_apply_hrecord_template}},
|
|
Result
|
|
)
|
|
%% TODO: async mode is not supported yet
|
|
end,
|
|
ok.
|
|
|
|
%% Connector Action Tests
|
|
|
|
t_action_on_get_status(Config) ->
|
|
emqx_bridge_v2_testlib:t_on_get_status(Config, #{failure_status => connecting}).
|
|
|
|
t_action_create_via_http(Config) ->
|
|
emqx_bridge_v2_testlib:t_create_via_http(Config).
|
|
|
|
t_action_sync_query(Config) ->
|
|
MakeMessageFun = fun() -> rand_data() end,
|
|
IsSuccessCheck = fun(Result) -> ?assertEqual(ok, Result) end,
|
|
TracePoint = hstreamdb_connector_query_append_return,
|
|
emqx_bridge_v2_testlib:t_sync_query(Config, MakeMessageFun, IsSuccessCheck, TracePoint).
|
|
|
|
t_action_start_stop(Config) ->
|
|
StopTracePoint = hstreamdb_connector_on_stop,
|
|
emqx_bridge_v2_testlib:t_start_stop(Config, StopTracePoint).
|
|
|
|
%%------------------------------------------------------------------------------
|
|
%% Helper fns
|
|
%%------------------------------------------------------------------------------
|
|
|
|
common_init(ConfigT) ->
|
|
Host = os:getenv("HSTREAMDB_HOST", "toxiproxy"),
|
|
RawPort = os:getenv("HSTREAMDB_PORT", str(?HSTREAMDB_DEFAULT_PORT)),
|
|
Port = list_to_integer(RawPort),
|
|
URL = "http://" ++ Host ++ ":" ++ RawPort,
|
|
|
|
Config0 = [
|
|
{bridge_type, <<"hstreamdb">>},
|
|
{bridge_name, <<"my_hstreamdb_action">>},
|
|
{connector_type, <<"hstreamdb">>},
|
|
{connector_name, <<"my_hstreamdb_connector">>},
|
|
{hstreamdb_host, Host},
|
|
{hstreamdb_port, Port},
|
|
{hstreamdb_url, URL},
|
|
%% see also for `proxy_name` : $PROJ_ROOT/.ci/docker-compose-file/toxiproxy.json
|
|
{proxy_name, "hstreamdb"},
|
|
{batch_size, batch_size(ConfigT)}
|
|
| ConfigT
|
|
],
|
|
|
|
BridgeType = proplists:get_value(bridge_type, Config0, <<"hstreamdb">>),
|
|
case emqx_common_test_helpers:is_tcp_server_available(Host, Port) of
|
|
true ->
|
|
% Setup toxiproxy
|
|
ProxyHost = os:getenv("PROXY_HOST", "toxiproxy"),
|
|
ProxyPort = list_to_integer(os:getenv("PROXY_PORT", "8474")),
|
|
emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
|
|
% Ensure EE bridge module is loaded
|
|
ok = emqx_common_test_helpers:start_apps([emqx_conf, emqx_resource, emqx_bridge]),
|
|
_ = application:ensure_all_started(hstreamdb_erl),
|
|
_ = emqx_bridge_enterprise:module_info(),
|
|
emqx_mgmt_api_test_util:init_suite(),
|
|
% Connect to hstreamdb directly
|
|
% drop old stream and then create new one
|
|
connect_and_delete_stream(Config0),
|
|
connect_and_create_stream(Config0),
|
|
{Name, HStreamDBConf} = hstreamdb_config(BridgeType, Config0),
|
|
Config =
|
|
[
|
|
{hstreamdb_config, HStreamDBConf},
|
|
{hstreamdb_bridge_type, BridgeType},
|
|
{hstreamdb_name, Name},
|
|
{bridge_config, action_config(Config0)},
|
|
{connector_config, connector_config(Config0)},
|
|
{proxy_host, ProxyHost},
|
|
{proxy_port, ProxyPort}
|
|
| Config0
|
|
],
|
|
Config;
|
|
false ->
|
|
case os:getenv("IS_CI") of
|
|
"yes" ->
|
|
throw(no_hstreamdb);
|
|
_ ->
|
|
{skip, no_hstreamdb}
|
|
end
|
|
end.
|
|
|
|
hstreamdb_config(BridgeType, Config) ->
|
|
Port = integer_to_list(?config(hstreamdb_port, Config)),
|
|
URL = "http://" ++ ?config(hstreamdb_host, Config) ++ ":" ++ Port,
|
|
Name = ?BRIDGE_NAME,
|
|
BatchSize = batch_size(Config),
|
|
ConfigString =
|
|
io_lib:format(
|
|
"bridges.~s.~s {\n"
|
|
" enable = true\n"
|
|
" url = ~p\n"
|
|
" stream = ~p\n"
|
|
" record_template = ~p\n"
|
|
" pool_size = ~p\n"
|
|
" grpc_timeout = ~p\n"
|
|
" resource_opts = {\n"
|
|
%% always sync
|
|
" query_mode = sync\n"
|
|
" request_ttl = 10000ms\n"
|
|
" batch_size = ~b\n"
|
|
" worker_pool_size = ~b\n"
|
|
" }\n"
|
|
"}",
|
|
[
|
|
BridgeType,
|
|
Name,
|
|
URL,
|
|
?STREAM,
|
|
?RECORD_TEMPLATE,
|
|
?POOL_SIZE,
|
|
?GRPC_TIMEOUT,
|
|
BatchSize,
|
|
?WORKER_POOL_SIZE
|
|
]
|
|
),
|
|
{Name, parse_and_check(ConfigString, BridgeType, Name)}.
|
|
|
|
action_config(Config) ->
|
|
ConnectorName = ?config(connector_name, Config),
|
|
BatchSize = batch_size(Config),
|
|
#{
|
|
<<"connector">> => ConnectorName,
|
|
<<"enable">> => true,
|
|
<<"parameters">> =>
|
|
#{
|
|
<<"aggregation_pool_size">> => ?POOL_SIZE,
|
|
<<"record_template">> => ?RECORD_TEMPLATE,
|
|
<<"stream">> => ?STREAM,
|
|
<<"writer_pool_size">> => ?POOL_SIZE
|
|
},
|
|
<<"resource_opts">> =>
|
|
#{
|
|
<<"batch_size">> => BatchSize,
|
|
<<"health_check_interval">> => <<"15s">>,
|
|
<<"inflight_window">> => 100,
|
|
<<"max_buffer_bytes">> => <<"256MB">>,
|
|
<<"query_mode">> => <<"sync">>,
|
|
<<"request_ttl">> => <<"45s">>,
|
|
<<"worker_pool_size">> => ?POOL_SIZE
|
|
}
|
|
}.
|
|
|
|
connector_config(Config) ->
|
|
Port = integer_to_list(?config(hstreamdb_port, Config)),
|
|
URL = "http://" ++ ?config(hstreamdb_host, Config) ++ ":" ++ Port,
|
|
#{
|
|
<<"url">> => URL,
|
|
<<"ssl">> =>
|
|
#{<<"enable">> => false, <<"verify">> => <<"verify_peer">>},
|
|
<<"grpc_timeout">> => <<"30s">>,
|
|
<<"resource_opts">> => #{
|
|
<<"health_check_interval">> => <<"15s">>,
|
|
<<"start_timeout">> => <<"5s">>
|
|
}
|
|
}.
|
|
|
|
parse_and_check(ConfigString, BridgeType, Name) ->
|
|
{ok, RawConf} = hocon:binary(ConfigString, #{format => map}),
|
|
hocon_tconf:check_plain(emqx_bridge_schema, RawConf, #{required => false, atom_key => false}),
|
|
#{<<"bridges">> := #{BridgeType := #{Name := Config}}} = RawConf,
|
|
Config.
|
|
|
|
-define(RPC_OPTIONS, #{pool_size => 4}).
|
|
|
|
-define(CONN_ATTEMPTS, 10).
|
|
|
|
default_options(Config) ->
|
|
#{
|
|
url => ?config(hstreamdb_url, Config),
|
|
rpc_options => ?RPC_OPTIONS
|
|
}.
|
|
|
|
connect_direct_hstream(Name, Config) ->
|
|
client(Name, Config, ?CONN_ATTEMPTS).
|
|
|
|
client(_Name, _Config, N) when N =< 0 -> error(cannot_connect);
|
|
client(Name, Config, N) ->
|
|
try
|
|
_ = hstreamdb:stop_client(Name),
|
|
{ok, Client} = hstreamdb:start_client(Name, default_options(Config)),
|
|
ok = hstreamdb_client:echo(Client),
|
|
Client
|
|
catch
|
|
Class:Error ->
|
|
ct:print("Error connecting: ~p", [{Class, Error}]),
|
|
ct:sleep(timer:seconds(1)),
|
|
client(Name, Config, N - 1)
|
|
end.
|
|
|
|
disconnect(Client) ->
|
|
hstreamdb:stop_client(Client).
|
|
|
|
create_bridge(Config) ->
|
|
create_bridge(Config, _Overrides = #{}).
|
|
|
|
create_bridge(Config, Overrides) ->
|
|
BridgeType = ?config(hstreamdb_bridge_type, Config),
|
|
Name = ?config(hstreamdb_name, Config),
|
|
HSDBConfig0 = ?config(hstreamdb_config, Config),
|
|
HSDBConfig = emqx_utils_maps:deep_merge(HSDBConfig0, Overrides),
|
|
emqx_bridge:create(BridgeType, Name, HSDBConfig).
|
|
|
|
delete_bridge(Config) ->
|
|
BridgeType = ?config(hstreamdb_bridge_type, Config),
|
|
Name = ?config(hstreamdb_name, Config),
|
|
emqx_bridge:remove(BridgeType, Name).
|
|
|
|
create_bridge_http(Params) ->
|
|
Path = emqx_mgmt_api_test_util:api_path(["bridges"]),
|
|
AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
|
|
case emqx_mgmt_api_test_util:request_api(post, Path, "", AuthHeader, Params) of
|
|
{ok, Res} -> {ok, emqx_utils_json:decode(Res, [return_maps])};
|
|
Error -> Error
|
|
end.
|
|
|
|
send_message(Config, Data) ->
|
|
Name = ?config(hstreamdb_name, Config),
|
|
BridgeType = ?config(hstreamdb_bridge_type, Config),
|
|
BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name),
|
|
emqx_bridge:send_message(BridgeID, Data).
|
|
|
|
query_resource(Config, Request) ->
|
|
Name = ?config(hstreamdb_name, Config),
|
|
BridgeType = ?config(hstreamdb_bridge_type, Config),
|
|
ID = emqx_bridge_v2:id(BridgeType, Name),
|
|
ResID = emqx_connector_resource:resource_id(BridgeType, Name),
|
|
emqx_resource:query(ID, Request, #{timeout => 1_000, connector_resource_id => ResID}).
|
|
|
|
restart_resource(Config) ->
|
|
BridgeName = ?config(hstreamdb_name, Config),
|
|
BridgeType = ?config(hstreamdb_bridge_type, Config),
|
|
emqx_bridge:disable_enable(disable, BridgeType, BridgeName),
|
|
timer:sleep(200),
|
|
emqx_bridge:disable_enable(enable, BridgeType, BridgeName).
|
|
|
|
resource_id(Config) ->
|
|
BridgeName = ?config(hstreamdb_name, Config),
|
|
BridgeType = ?config(hstreamdb_bridge_type, Config),
|
|
_ResourceID = emqx_bridge_resource:resource_id(BridgeType, BridgeName).
|
|
|
|
action_id(Config) ->
|
|
ActionName = ?config(hstreamdb_name, Config),
|
|
ActionType = ?config(hstreamdb_bridge_type, Config),
|
|
_ActionID = emqx_bridge_v2:id(ActionType, ActionName).
|
|
|
|
health_check_resource_ok(Config) ->
|
|
?assertEqual({ok, connected}, emqx_resource_manager:health_check(resource_id(Config))),
|
|
ActionName = ?config(hstreamdb_name, Config),
|
|
ActionType = ?config(hstreamdb_bridge_type, Config),
|
|
?assertMatch(#{status := connected}, emqx_bridge_v2:health_check(ActionType, ActionName)).
|
|
|
|
health_check_resource_down(Config) ->
|
|
case emqx_resource_manager:health_check(resource_id(Config)) of
|
|
{ok, Status} when Status =:= disconnected orelse Status =:= connecting ->
|
|
ok;
|
|
{error, timeout} ->
|
|
ok;
|
|
Other ->
|
|
?assert(
|
|
false, lists:flatten(io_lib:format("invalid health check result:~p~n", [Other]))
|
|
)
|
|
end,
|
|
ActionName = ?config(hstreamdb_name, Config),
|
|
ActionType = ?config(hstreamdb_bridge_type, Config),
|
|
#{status := StatusV2} = emqx_bridge_v2:health_check(ActionType, ActionName),
|
|
case StatusV2 of
|
|
disconnected ->
|
|
ok;
|
|
connecting ->
|
|
ok;
|
|
OtherV2 ->
|
|
?assert(
|
|
false, lists:flatten(io_lib:format("invalid health check result:~p~n", [OtherV2]))
|
|
)
|
|
end.
|
|
|
|
% These funs start and then stop the hstreamdb connection
|
|
connect_and_create_stream(Config) ->
|
|
?WITH_CLIENT(
|
|
_ = hstreamdb_client:create_stream(
|
|
Client, ?STREAM, ?REPLICATION_FACTOR, ?BACKLOG_RETENTION_SECOND, ?SHARD_COUNT
|
|
)
|
|
),
|
|
%% force write to stream to make it created and ready to be written data for test cases
|
|
ProducerOptions = #{
|
|
stream => ?STREAM,
|
|
buffer_options => #{
|
|
interval => 1000,
|
|
callback => {?MODULE, on_flush_result, [<<"WHAT">>]},
|
|
max_records => 1,
|
|
max_batches => 1
|
|
},
|
|
buffer_pool_size => 1,
|
|
writer_options => #{
|
|
grpc_timeout => 100
|
|
},
|
|
writer_pool_size => 1,
|
|
client_options => default_options(Config)
|
|
},
|
|
|
|
?WITH_CLIENT(
|
|
begin
|
|
ok = hstreamdb:start_producer(test_producer, ProducerOptions),
|
|
_ = hstreamdb:append_flush(test_producer, hstreamdb:to_record([], raw, rand_payload())),
|
|
_ = hstreamdb:stop_producer(test_producer)
|
|
end
|
|
).
|
|
|
|
on_flush_result({{flush, _Stream, _Records}, {ok, _Resp}}) ->
|
|
ok;
|
|
on_flush_result({{flush, _Stream, _Records}, {error, _Reason}}) ->
|
|
ok.
|
|
|
|
connect_and_delete_stream(Config) ->
|
|
?WITH_CLIENT(
|
|
_ = hstreamdb_client:delete_stream(Client, ?STREAM)
|
|
).
|
|
|
|
%%--------------------------------------------------------------------
|
|
%% help functions
|
|
%%--------------------------------------------------------------------
|
|
|
|
batch_size(Config) ->
|
|
case ?config(enable_batch, Config) of
|
|
true -> ?BATCH_SIZE;
|
|
false -> 1
|
|
end.
|
|
|
|
rand_data() ->
|
|
#{
|
|
%% Raw MTTT Payload in binary
|
|
payload => rand_payload(),
|
|
id => <<"0005F8F84FFFAFB9F44200000D810002">>,
|
|
topic => <<"test/topic">>,
|
|
qos => 0
|
|
}.
|
|
|
|
rand_payload() ->
|
|
emqx_utils_json:encode(#{
|
|
temperature => rand:uniform(40), humidity => rand:uniform(100)
|
|
}).
|
|
|
|
gen_batch_req(Count, ActionId) when
|
|
is_integer(Count) andalso Count > 0
|
|
->
|
|
[{ActionId, rand_data()} || _Val <- lists:seq(1, Count)];
|
|
gen_batch_req(Count, _ActionId) ->
|
|
ct:pal("Gen batch requests failed with unexpected Count: ~p", [Count]).
|
|
|
|
str(List) when is_list(List) ->
|
|
unicode:characters_to_list(List, utf8);
|
|
str(Bin) when is_binary(Bin) ->
|
|
unicode:characters_to_list(Bin, utf8);
|
|
str(Num) when is_number(Num) ->
|
|
number_to_list(Num).
|
|
|
|
number_to_list(Int) when is_integer(Int) ->
|
|
integer_to_list(Int);
|
|
number_to_list(Float) when is_float(Float) ->
|
|
float_to_list(Float, [{decimals, 10}, compact]).
|