yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: delete emqx_resource:create/4 and /5 

the functions are only used in tests,
now the tests are rewritten to call create_local/4 and /5 instead
This commit is contained in:
Zaiming (Stone) Shi 2024-02-20 19:28:24 +01:00
parent 0cbe626aed
commit 1878fded5a
3 changed files with 68 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
apps/emqx_resource/src/emqx_resource.erl
View File

@ -28,8 +28,6 @@
-export([
check_config/2,
check_and_create/4,
check_and_create/5,
check_and_create_local/4,
check_and_create_local/5,
check_and_recreate/4,
@ -42,12 +40,8 @@
%% store the config and start the instance
-export([
create/4,
create/5,
create_local/4,
create_local/5,
%% run start/2, health_check/2 and stop/1 sequentially
create_dry_run/2,
create_dry_run_local/2,
create_dry_run_local/3,
create_dry_run_local/4,
@ -275,16 +269,6 @@ is_resource_mod(Module) ->
%% =================================================================================
%% APIs for resource instances
%% =================================================================================
-spec create(resource_id(), resource_group(), resource_type(), resource_config()) ->
{ok, resource_data() | 'already_created'} | {error, Reason :: term()}.
create(ResId, Group, ResourceType, Config) ->
create(ResId, Group, ResourceType, Config, #{}).
-spec create(resource_id(), resource_group(), resource_type(), resource_config(), creation_opts()) ->
{ok, resource_data() | 'already_created'} | {error, Reason :: term()}.
create(ResId, Group, ResourceType, Config, Opts) ->
emqx_resource_proto_v1:create(ResId, Group, ResourceType, Config, Opts).
% --------------------------------------------
-spec create_local(resource_id(), resource_group(), resource_type(), resource_config()) ->
{ok, resource_data() | 'already_created'} | {error, Reason :: term()}.
@ -302,11 +286,6 @@ create_local(ResId, Group, ResourceType, Config) ->
create_local(ResId, Group, ResourceType, Config, Opts) ->
emqx_resource_manager:ensure_resource(ResId, Group, ResourceType, Config, Opts).
-spec create_dry_run(resource_type(), resource_config()) ->
ok | {error, Reason :: term()}.
create_dry_run(ResourceType, Config) ->
emqx_resource_proto_v1:create_dry_run(ResourceType, Config).
-spec create_dry_run_local(resource_type(), resource_config()) ->
ok | {error, Reason :: term()}.
create_dry_run_local(ResourceType, Config) ->
@ -616,31 +595,6 @@ query_mode(Mod, Config, Opts) ->
check_config(ResourceType, Conf) ->
emqx_hocon:check(ResourceType, Conf).
-spec check_and_create(
resource_id(),
resource_group(),
resource_type(),
raw_resource_config()
) ->
{ok, resource_data() | 'already_created'} | {error, term()}.
check_and_create(ResId, Group, ResourceType, RawConfig) ->
check_and_create(ResId, Group, ResourceType, RawConfig, #{}).
-spec check_and_create(
resource_id(),
resource_group(),
resource_type(),
raw_resource_config(),
creation_opts()
) ->
{ok, resource_data() | 'already_created'} | {error, term()}.
check_and_create(ResId, Group, ResourceType, RawConfig, Opts) ->
check_and_do(
ResourceType,
RawConfig,
fun(ResConf) -> create(ResId, Group, ResourceType, ResConf, Opts) end
).
-spec check_and_create_local(
resource_id(),
resource_group(),

6
apps/emqx_resource/src/proto/emqx_resource_proto_v1.erl
View File

@ -20,6 +20,7 @@
-export([
introduced_in/0,
deprecated_since/0,
create/5,
create_dry_run/2,
recreate/4,
@ -33,6 +34,10 @@
introduced_in() ->
"5.0.0".
deprecated_since() ->
"5.6.0".
%% deprecated, delete after 5.7.0
-spec create(
resource_id(),
resource_group(),
@ -46,6 +51,7 @@ create(ResId, Group, ResourceType, Config, Opts) ->
ResId, Group, ResourceType, Config, Opts
]).
%% deprecated, delete after 5.7.0
-spec create_dry_run(
resource_type(),
resource_config()

118
apps/emqx_resource/test/emqx_resource_SUITE.erl
View File

@ -88,7 +88,7 @@ t_create_remove(_) ->
?assertMatch(
{ok, _},
emqx_resource:create(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -136,7 +136,7 @@ t_create_remove_local(_) ->
?assertMatch(
{ok, _},
emqx_resource:create_local(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -185,7 +185,7 @@ t_do_not_start_after_created(_) ->
begin
?assertMatch(
{ok, _},
emqx_resource:create_local(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -226,7 +226,7 @@ t_do_not_start_after_created(_) ->
).
t_query(_) ->
{ok, _} = emqx_resource:create_local(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -243,7 +243,7 @@ t_query(_) ->
ok = emqx_resource:remove_local(?ID).
t_query_counter(_) ->
{ok, _} = emqx_resource:create_local(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -260,7 +260,7 @@ t_query_counter(_) ->
t_batch_query_counter(_) ->
BatchSize = 100,
{ok, _} = emqx_resource:create_local(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -313,7 +313,7 @@ t_batch_query_counter(_) ->
ok = emqx_resource:remove_local(?ID).
t_query_counter_async_query(_) ->
{ok, _} = emqx_resource:create_local(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -371,7 +371,7 @@ t_query_counter_async_callback(_) ->
ets:insert(Tab, {make_ref(), Result})
end,
ReqOpts = #{async_reply_fun => {Insert, [Tab0]}},
{ok, _} = emqx_resource:create_local(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -452,7 +452,7 @@ t_query_counter_async_inflight(_) ->
end,
ReqOpts = fun() -> #{async_reply_fun => {Insert0, [Tab0, make_ref()]}} end,
WindowSize = 15,
{ok, _} = emqx_resource:create_local(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -635,7 +635,7 @@ t_query_counter_async_inflight_batch(_) ->
ReqOpts = fun() -> #{async_reply_fun => {Insert0, [Tab0, make_ref()]}} end,
BatchSize = 2,
WindowSize = 15,
{ok, _} = emqx_resource:create_local(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -837,7 +837,7 @@ t_healthy_timeout(_) ->
begin
?assertMatch(
{ok, _},
emqx_resource:create_local(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -866,7 +866,7 @@ t_healthy(_) ->
begin
?assertMatch(
{ok, _},
emqx_resource:create_local(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -904,7 +904,7 @@ t_unhealthy_target(_) ->
HealthCheckError = {unhealthy_target, "some message"},
?assertMatch(
{ok, _},
emqx_resource:create_local(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -937,7 +937,7 @@ t_stop_start(_) ->
begin
?assertMatch(
{error, _},
emqx_resource:check_and_create(
emqx_resource:check_and_create_local(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -947,7 +947,7 @@ t_stop_start(_) ->
?assertMatch(
{ok, _},
emqx_resource:check_and_create(
emqx_resource:check_and_create_local(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1071,13 +1071,13 @@ t_stop_start_local(_) ->
).
t_list_filter(_) ->
{ok, _} = emqx_resource:create_local(
{ok, _} = create(
emqx_resource:generate_id(<<"a">>),
<<"group1">>,
?TEST_RESOURCE,
#{name => a}
),
{ok, _} = emqx_resource:create_local(
{ok, _} = create(
emqx_resource:generate_id(<<"a">>),
<<"group2">>,
?TEST_RESOURCE,
@ -1208,7 +1208,7 @@ t_test_func(_) ->
).
t_reset_metrics(_) ->
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1222,7 +1222,7 @@ t_reset_metrics(_) ->
?assertNot(is_process_alive(Pid)).
t_auto_retry(_) ->
{Res, _} = emqx_resource:create_local(
{Res, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1239,7 +1239,7 @@ t_start_throw_error(_Config) ->
?assertMatch(
{{ok, _}, {ok, _}},
?wait_async_action(
emqx_resource:create_local(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1257,7 +1257,7 @@ t_start_throw_error(_Config) ->
t_health_check_disconnected(_) ->
?check_trace(
begin
_ = emqx_resource:create_local(
_ = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1276,7 +1276,7 @@ t_health_check_disconnected(_) ->
).
t_unblock_only_required_buffer_workers(_) ->
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1292,7 +1292,7 @@ t_unblock_only_required_buffer_workers(_) ->
fun emqx_resource_buffer_worker:block/1,
emqx_resource_buffer_worker_sup:worker_pids(?ID)
),
emqx_resource:create(
create(
?ID1,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1323,7 +1323,7 @@ t_unblock_only_required_buffer_workers(_) ->
).
t_retry_batch(_Config) ->
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1422,7 +1422,7 @@ t_retry_batch(_Config) ->
t_delete_and_re_create_with_same_name(_Config) ->
NumBufferWorkers = 2,
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1489,7 +1489,7 @@ t_delete_and_re_create_with_same_name(_Config) ->
%% re-create the resource with the *same name*
{{ok, _}, {ok, _Events}} =
?wait_async_action(
emqx_resource:create(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1521,7 +1521,7 @@ t_delete_and_re_create_with_same_name(_Config) ->
%% check that, if we configure a max queue size too small, then we
%% never send requests and always overflow.
t_always_overflow(_Config) ->
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1560,7 +1560,7 @@ t_always_overflow(_Config) ->
t_retry_sync_inflight(_Config) ->
ResumeInterval = 1_000,
emqx_connector_demo:set_callback_mode(always_sync),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1610,7 +1610,7 @@ t_retry_sync_inflight(_Config) ->
t_retry_sync_inflight_batch(_Config) ->
ResumeInterval = 1_000,
emqx_connector_demo:set_callback_mode(always_sync),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1662,7 +1662,7 @@ t_retry_sync_inflight_batch(_Config) ->
t_retry_async_inflight(_Config) ->
ResumeInterval = 1_000,
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1706,7 +1706,7 @@ t_retry_async_inflight_full(_Config) ->
ResumeInterval = 1_000,
AsyncInflightWindow = 5,
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1769,7 +1769,7 @@ t_async_reply_multi_eval(_Config) ->
AsyncInflightWindow = 3,
TotalQueries = AsyncInflightWindow * 5,
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1818,7 +1818,7 @@ t_async_reply_multi_eval(_Config) ->
t_retry_async_inflight_batch(_Config) ->
ResumeInterval = 1_000,
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1865,7 +1865,7 @@ t_async_pool_worker_death(_Config) ->
ResumeInterval = 1_000,
NumBufferWorkers = 2,
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1956,7 +1956,7 @@ t_async_pool_worker_death(_Config) ->
t_expiration_sync_before_sending(_Config) ->
emqx_connector_demo:set_callback_mode(always_sync),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1973,7 +1973,7 @@ t_expiration_sync_before_sending(_Config) ->
t_expiration_sync_batch_before_sending(_Config) ->
emqx_connector_demo:set_callback_mode(always_sync),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -1991,7 +1991,7 @@ t_expiration_sync_batch_before_sending(_Config) ->
t_expiration_async_before_sending(_Config) ->
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2008,7 +2008,7 @@ t_expiration_async_before_sending(_Config) ->
t_expiration_async_batch_before_sending(_Config) ->
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2089,7 +2089,7 @@ do_t_expiration_before_sending(QueryMode) ->
t_expiration_sync_before_sending_partial_batch(_Config) ->
emqx_connector_demo:set_callback_mode(always_sync),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2108,7 +2108,7 @@ t_expiration_sync_before_sending_partial_batch(_Config) ->
t_expiration_async_before_sending_partial_batch(_Config) ->
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2265,7 +2265,7 @@ do_t_expiration_before_sending_partial_batch(QueryMode) ->
t_expiration_async_after_reply(_Config) ->
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2282,7 +2282,7 @@ t_expiration_async_after_reply(_Config) ->
t_expiration_async_batch_after_reply(_Config) ->
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2407,7 +2407,7 @@ do_t_expiration_async_after_reply(IsBatch) ->
t_expiration_batch_all_expired_after_reply(_Config) ->
ResumeInterval = 300,
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2496,7 +2496,7 @@ t_expiration_batch_all_expired_after_reply(_Config) ->
t_expiration_retry(_Config) ->
emqx_connector_demo:set_callback_mode(always_sync),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2512,7 +2512,7 @@ t_expiration_retry(_Config) ->
t_expiration_retry_batch(_Config) ->
emqx_connector_demo:set_callback_mode(always_sync),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2612,7 +2612,7 @@ do_t_expiration_retry() ->
t_expiration_retry_batch_multiple_times(_Config) ->
ResumeInterval = 300,
emqx_connector_demo:set_callback_mode(always_sync),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2680,7 +2680,7 @@ t_expiration_retry_batch_multiple_times(_Config) ->
t_recursive_flush(_Config) ->
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2695,7 +2695,7 @@ t_recursive_flush(_Config) ->
t_recursive_flush_batch(_Config) ->
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2750,7 +2750,7 @@ t_call_mode_uncoupled_from_query_mode(_Config) ->
%% calls, even if the underlying connector itself only
%% supports sync calls.
emqx_connector_demo:set_callback_mode(always_sync),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2775,7 +2775,7 @@ t_call_mode_uncoupled_from_query_mode(_Config) ->
%% calls can be called synchronously, but the underlying
%% call should be async.
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
{ok, _} = create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2829,7 +2829,7 @@ t_volatile_offload_mode(_Config) ->
%% default to equal max bytes.
?assertMatch(
{ok, _},
emqx_resource:create(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2842,7 +2842,7 @@ t_volatile_offload_mode(_Config) ->
%% Create with segment bytes < max bytes
?assertMatch(
{ok, _},
emqx_resource:create(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2857,7 +2857,7 @@ t_volatile_offload_mode(_Config) ->
%% Create with segment bytes = max bytes
?assertMatch(
{ok, _},
emqx_resource:create(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2874,7 +2874,7 @@ t_volatile_offload_mode(_Config) ->
%% to max bytes.
?assertMatch(
{ok, _},
emqx_resource:create(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2931,7 +2931,7 @@ t_late_call_reply(_Config) ->
RequestTTL = 500,
?assertMatch(
{ok, _},
emqx_resource:create(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -2987,7 +2987,7 @@ do_t_resource_activate_alarm_once(ResourceConfig, SubscribeEvent) ->
?check_trace(
begin
?wait_async_action(
emqx_resource:create_local(
create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
@ -3288,3 +3288,9 @@ gauge_metric_set_fns() ->
_ -> false
end
].
create(Id, Group, Type, Config) ->
emqx_resource:create_local(Id, Group, Type, Config).
create(Id, Group, Type, Config, Opts) ->
emqx_resource:create_local(Id, Group, Type, Config, Opts).