refactor(connnector): rename waiting_connect_complete -> wait_for_resource_ready
Rename the option to wait_for_resource_ready and defaults to 5s.
This commit is contained in:
parent
a38cac0233
commit
1d023b541f
|
@ -134,7 +134,7 @@ create(#{method := Method,
|
||||||
emqx_connector_http,
|
emqx_connector_http,
|
||||||
Config#{base_url => maps:remove(query, URIMap),
|
Config#{base_url => maps:remove(query, URIMap),
|
||||||
pool_type => random},
|
pool_type => random},
|
||||||
#{waiting_connect_complete => 5000}) of
|
#{}) of
|
||||||
{ok, already_created} ->
|
{ok, already_created} ->
|
||||||
{ok, State};
|
{ok, State};
|
||||||
{ok, _} ->
|
{ok, _} ->
|
||||||
|
|
|
@ -116,7 +116,7 @@ create(#{selector := Selector} = Config) ->
|
||||||
?RESOURCE_GROUP,
|
?RESOURCE_GROUP,
|
||||||
emqx_connector_mongo,
|
emqx_connector_mongo,
|
||||||
Config,
|
Config,
|
||||||
#{waiting_connect_complete => 5000}) of
|
#{}) of
|
||||||
{ok, already_created} ->
|
{ok, already_created} ->
|
||||||
{ok, NState};
|
{ok, NState};
|
||||||
{ok, _} ->
|
{ok, _} ->
|
||||||
|
|
|
@ -85,7 +85,7 @@ create(#{password_hash_algorithm := Algorithm,
|
||||||
?RESOURCE_GROUP,
|
?RESOURCE_GROUP,
|
||||||
emqx_connector_mysql,
|
emqx_connector_mysql,
|
||||||
Config,
|
Config,
|
||||||
#{waiting_connect_complete => 5000}) of
|
#{}) of
|
||||||
{ok, already_created} ->
|
{ok, already_created} ->
|
||||||
{ok, State};
|
{ok, State};
|
||||||
{ok, _} ->
|
{ok, _} ->
|
||||||
|
|
|
@ -81,7 +81,7 @@ create(#{query := Query0,
|
||||||
resource_id => ResourceId},
|
resource_id => ResourceId},
|
||||||
case emqx_resource:create_local(ResourceId, ?RESOURCE_GROUP, emqx_connector_pgsql,
|
case emqx_resource:create_local(ResourceId, ?RESOURCE_GROUP, emqx_connector_pgsql,
|
||||||
Config#{named_queries => #{ResourceId => Query}},
|
Config#{named_queries => #{ResourceId => Query}},
|
||||||
#{waiting_connect_complete => 5000}) of
|
#{}) of
|
||||||
{ok, already_created} ->
|
{ok, already_created} ->
|
||||||
{ok, State};
|
{ok, State};
|
||||||
{ok, _} ->
|
{ok, _} ->
|
||||||
|
|
|
@ -93,7 +93,7 @@ create(#{cmd := Cmd,
|
||||||
resource_id => ResourceId},
|
resource_id => ResourceId},
|
||||||
case emqx_resource:create_local(ResourceId, ?RESOURCE_GROUP,
|
case emqx_resource:create_local(ResourceId, ?RESOURCE_GROUP,
|
||||||
emqx_connector_redis, Config,
|
emqx_connector_redis, Config,
|
||||||
#{waiting_connect_complete => 5000}) of
|
#{}) of
|
||||||
{ok, already_created} ->
|
{ok, already_created} ->
|
||||||
{ok, NState};
|
{ok, NState};
|
||||||
{ok, _} ->
|
{ok, _} ->
|
||||||
|
|
|
@ -63,7 +63,7 @@ init_per_suite(Config) ->
|
||||||
?RESOURCE_GROUP,
|
?RESOURCE_GROUP,
|
||||||
emqx_connector_mysql,
|
emqx_connector_mysql,
|
||||||
mysql_config(),
|
mysql_config(),
|
||||||
#{waiting_connect_complete => 5000}),
|
#{}),
|
||||||
Config;
|
Config;
|
||||||
false ->
|
false ->
|
||||||
{skip, no_mysql}
|
{skip, no_mysql}
|
||||||
|
|
|
@ -64,7 +64,7 @@ init_per_suite(Config) ->
|
||||||
?RESOURCE_GROUP,
|
?RESOURCE_GROUP,
|
||||||
emqx_connector_pgsql,
|
emqx_connector_pgsql,
|
||||||
pgsql_config(),
|
pgsql_config(),
|
||||||
#{waiting_connect_complete => 5000}),
|
#{}),
|
||||||
Config;
|
Config;
|
||||||
false ->
|
false ->
|
||||||
{skip, no_pgsql}
|
{skip, no_pgsql}
|
||||||
|
|
|
@ -63,7 +63,7 @@ init_per_suite(Config) ->
|
||||||
?RESOURCE_GROUP,
|
?RESOURCE_GROUP,
|
||||||
emqx_connector_redis,
|
emqx_connector_redis,
|
||||||
redis_config(),
|
redis_config(),
|
||||||
#{waiting_connect_complete => 5000}),
|
#{}),
|
||||||
Config;
|
Config;
|
||||||
false ->
|
false ->
|
||||||
{skip, no_redis}
|
{skip, no_redis}
|
||||||
|
|
|
@ -56,7 +56,7 @@ init(#{query := SQL0} = Source) ->
|
||||||
?RESOURCE_GROUP,
|
?RESOURCE_GROUP,
|
||||||
emqx_connector_pgsql,
|
emqx_connector_pgsql,
|
||||||
Source#{named_queries => #{ResourceID => SQL}},
|
Source#{named_queries => #{ResourceID => SQL}},
|
||||||
#{waiting_connect_complete => 5000}) of
|
#{}) of
|
||||||
{ok, _} ->
|
{ok, _} ->
|
||||||
Source#{annotations =>
|
Source#{annotations =>
|
||||||
#{id => ResourceID,
|
#{id => ResourceID,
|
||||||
|
|
|
@ -38,7 +38,7 @@ create_resource(Module, Config) ->
|
||||||
case emqx_resource:create_local(ResourceID,
|
case emqx_resource:create_local(ResourceID,
|
||||||
?RESOURCE_GROUP,
|
?RESOURCE_GROUP,
|
||||||
Module, Config,
|
Module, Config,
|
||||||
#{waiting_connect_complete => 5000}) of
|
#{}) of
|
||||||
{ok, already_created} -> {ok, ResourceID};
|
{ok, already_created} -> {ok, ResourceID};
|
||||||
{ok, _} -> {ok, ResourceID};
|
{ok, _} -> {ok, ResourceID};
|
||||||
{error, Reason} -> {error, Reason}
|
{error, Reason} -> {error, Reason}
|
||||||
|
|
|
@ -45,7 +45,7 @@ init_per_suite(Config) ->
|
||||||
?RESOURCE_GROUP,
|
?RESOURCE_GROUP,
|
||||||
emqx_connector_mysql,
|
emqx_connector_mysql,
|
||||||
mysql_config(),
|
mysql_config(),
|
||||||
#{waiting_connect_complete => 5000}),
|
#{}),
|
||||||
Config;
|
Config;
|
||||||
false ->
|
false ->
|
||||||
{skip, no_mysql}
|
{skip, no_mysql}
|
||||||
|
|
|
@ -45,7 +45,7 @@ init_per_suite(Config) ->
|
||||||
?RESOURCE_GROUP,
|
?RESOURCE_GROUP,
|
||||||
emqx_connector_pgsql,
|
emqx_connector_pgsql,
|
||||||
pgsql_config(),
|
pgsql_config(),
|
||||||
#{waiting_connect_complete => 5000}),
|
#{}),
|
||||||
Config;
|
Config;
|
||||||
false ->
|
false ->
|
||||||
{skip, no_pgsql}
|
{skip, no_pgsql}
|
||||||
|
|
|
@ -46,7 +46,7 @@ init_per_suite(Config) ->
|
||||||
?RESOURCE_GROUP,
|
?RESOURCE_GROUP,
|
||||||
emqx_connector_redis,
|
emqx_connector_redis,
|
||||||
redis_config(),
|
redis_config(),
|
||||||
#{waiting_connect_complete => 5000}),
|
#{}),
|
||||||
Config;
|
Config;
|
||||||
false ->
|
false ->
|
||||||
{skip, no_redis}
|
{skip, no_redis}
|
||||||
|
|
|
@ -225,7 +225,7 @@ create(Type, Name, Conf) ->
|
||||||
<<"emqx_bridge">>,
|
<<"emqx_bridge">>,
|
||||||
emqx_bridge:resource_type(Type),
|
emqx_bridge:resource_type(Type),
|
||||||
parse_confs(Type, Name, Conf),
|
parse_confs(Type, Name, Conf),
|
||||||
#{waiting_connect_complete => 5000}) of
|
#{}) of
|
||||||
{ok, already_created} -> maybe_disable_bridge(Type, Name, Conf);
|
{ok, already_created} -> maybe_disable_bridge(Type, Name, Conf);
|
||||||
{ok, _} -> maybe_disable_bridge(Type, Name, Conf);
|
{ok, _} -> maybe_disable_bridge(Type, Name, Conf);
|
||||||
{error, Reason} -> {error, Reason}
|
{error, Reason} -> {error, Reason}
|
||||||
|
@ -272,7 +272,7 @@ recreate(Type, Name, Conf) ->
|
||||||
emqx_resource:recreate_local(resource_id(Type, Name),
|
emqx_resource:recreate_local(resource_id(Type, Name),
|
||||||
emqx_bridge:resource_type(Type),
|
emqx_bridge:resource_type(Type),
|
||||||
parse_confs(Type, Name, Conf),
|
parse_confs(Type, Name, Conf),
|
||||||
#{waiting_connect_complete => 5000}).
|
#{}).
|
||||||
|
|
||||||
create_dry_run(Type, Conf) ->
|
create_dry_run(Type, Conf) ->
|
||||||
Conf0 = Conf#{<<"ingress">> => #{<<"remote_topic">> => <<"t">>}},
|
Conf0 = Conf#{<<"ingress">> => #{<<"remote_topic">> => <<"t">>}},
|
||||||
|
|
|
@ -79,8 +79,14 @@ init_per_testcase(_, Config) ->
|
||||||
{ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
|
{ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
|
||||||
Config.
|
Config.
|
||||||
end_per_testcase(_, _Config) ->
|
end_per_testcase(_, _Config) ->
|
||||||
|
clear_resources(),
|
||||||
ok.
|
ok.
|
||||||
|
|
||||||
|
clear_resources() ->
|
||||||
|
lists:foreach(fun(#{type := Type, name := Name}) ->
|
||||||
|
ok = emqx_bridge:remove(Type, Name)
|
||||||
|
end, emqx_bridge:list()).
|
||||||
|
|
||||||
%%------------------------------------------------------------------------------
|
%%------------------------------------------------------------------------------
|
||||||
%% HTTP server for testing
|
%% HTTP server for testing
|
||||||
%%------------------------------------------------------------------------------
|
%%------------------------------------------------------------------------------
|
||||||
|
|
|
@ -71,7 +71,7 @@ perform_lifecycle_check(PoolName, InitialConfig) ->
|
||||||
?CONNECTOR_RESOURCE_GROUP,
|
?CONNECTOR_RESOURCE_GROUP,
|
||||||
?MYSQL_RESOURCE_MOD,
|
?MYSQL_RESOURCE_MOD,
|
||||||
CheckedConfig,
|
CheckedConfig,
|
||||||
#{waiting_connect_complete => 5000}
|
#{}
|
||||||
),
|
),
|
||||||
?assertEqual(InitialStatus, connected),
|
?assertEqual(InitialStatus, connected),
|
||||||
% Instance should match the state and status of the just started resource
|
% Instance should match the state and status of the just started resource
|
||||||
|
|
|
@ -72,7 +72,7 @@ perform_lifecycle_check(PoolName, InitialConfig) ->
|
||||||
?CONNECTOR_RESOURCE_GROUP,
|
?CONNECTOR_RESOURCE_GROUP,
|
||||||
?PGSQL_RESOURCE_MOD,
|
?PGSQL_RESOURCE_MOD,
|
||||||
CheckedConfig,
|
CheckedConfig,
|
||||||
#{waiting_connect_complete => 5000}
|
#{}
|
||||||
),
|
),
|
||||||
?assertEqual(InitialStatus, connected),
|
?assertEqual(InitialStatus, connected),
|
||||||
% Instance should match the state and status of the just started resource
|
% Instance should match the state and status of the just started resource
|
||||||
|
|
|
@ -86,7 +86,7 @@ perform_lifecycle_check(PoolName, InitialConfig, RedisCommand) ->
|
||||||
?CONNECTOR_RESOURCE_GROUP,
|
?CONNECTOR_RESOURCE_GROUP,
|
||||||
?REDIS_RESOURCE_MOD,
|
?REDIS_RESOURCE_MOD,
|
||||||
CheckedConfig,
|
CheckedConfig,
|
||||||
#{waiting_connect_complete => 5000}
|
#{}
|
||||||
),
|
),
|
||||||
?assertEqual(InitialStatus, connected),
|
?assertEqual(InitialStatus, connected),
|
||||||
% Instance should match the state and status of the just started resource
|
% Instance should match the state and status of the just started resource
|
||||||
|
|
|
@ -157,6 +157,8 @@ init(Name) ->
|
||||||
persistent_term:put(?CntrRef(Name), #{}),
|
persistent_term:put(?CntrRef(Name), #{}),
|
||||||
{ok, #state{}}.
|
{ok, #state{}}.
|
||||||
|
|
||||||
|
handle_call({get_rate, _Id}, _From, State = #state{rates = undefined}) ->
|
||||||
|
{reply, #{}, State};
|
||||||
handle_call({get_rate, Id}, _From, State = #state{rates = Rates}) ->
|
handle_call({get_rate, Id}, _From, State = #state{rates = Rates}) ->
|
||||||
{reply, case maps:get(Id, Rates, undefined) of
|
{reply, case maps:get(Id, Rates, undefined) of
|
||||||
undefined -> #{};
|
undefined -> #{};
|
||||||
|
|
|
@ -178,14 +178,17 @@ do_recreate(InstId, ResourceType, NewConfig, Opts) ->
|
||||||
{error, not_found}
|
{error, not_found}
|
||||||
end.
|
end.
|
||||||
|
|
||||||
wait_for_resource_ready(InstId, 0) ->
|
wait_for_resource_ready(InstId, WaitTime) ->
|
||||||
force_lookup(InstId);
|
do_wait_for_resource_ready(InstId, WaitTime div 100).
|
||||||
wait_for_resource_ready(InstId, Retry) ->
|
|
||||||
|
do_wait_for_resource_ready(_InstId, 0) ->
|
||||||
|
timeout;
|
||||||
|
do_wait_for_resource_ready(InstId, Retry) ->
|
||||||
case force_lookup(InstId) of
|
case force_lookup(InstId) of
|
||||||
#{status := connected} = Data -> Data;
|
#{status := connected} -> ok;
|
||||||
_ ->
|
_ ->
|
||||||
timer:sleep(100),
|
timer:sleep(100),
|
||||||
wait_for_resource_ready(InstId, Retry-1)
|
do_wait_for_resource_ready(InstId, Retry-1)
|
||||||
end.
|
end.
|
||||||
|
|
||||||
do_create(InstId, Group, ResourceType, Config, Opts) ->
|
do_create(InstId, Group, ResourceType, Config, Opts) ->
|
||||||
|
@ -197,8 +200,7 @@ do_create(InstId, Group, ResourceType, Config, Opts) ->
|
||||||
ok ->
|
ok ->
|
||||||
ok = emqx_plugin_libs_metrics:create_metrics(resource_metrics, InstId,
|
ok = emqx_plugin_libs_metrics:create_metrics(resource_metrics, InstId,
|
||||||
[matched, success, failed, exception], [matched]),
|
[matched, success, failed, exception], [matched]),
|
||||||
WaitTime = maps:get(waiting_connect_complete , Opts, 0),
|
{ok, force_lookup(InstId)};
|
||||||
{ok, wait_for_resource_ready(InstId, WaitTime div 100)};
|
|
||||||
Error ->
|
Error ->
|
||||||
Error
|
Error
|
||||||
end
|
end
|
||||||
|
@ -252,6 +254,7 @@ do_start(InstId, Group, ResourceType, Config, Opts) when is_binary(InstId) ->
|
||||||
spawn(fun() ->
|
spawn(fun() ->
|
||||||
start_and_check(InstId, Group, ResourceType, Config, Opts, InitData)
|
start_and_check(InstId, Group, ResourceType, Config, Opts, InitData)
|
||||||
end),
|
end),
|
||||||
|
_ = wait_for_resource_ready(InstId, maps:get(wait_for_resource_ready, Opts, 5000)),
|
||||||
ok.
|
ok.
|
||||||
|
|
||||||
start_and_check(InstId, Group, ResourceType, Config, Opts, Data) ->
|
start_and_check(InstId, Group, ResourceType, Config, Opts, Data) ->
|
||||||
|
|
|
@ -361,7 +361,7 @@ create_resource(Context, #{type := DB} = Config) ->
|
||||||
<<"emqx_retainer">>,
|
<<"emqx_retainer">>,
|
||||||
list_to_existing_atom(io_lib:format("~ts_~ts", [emqx_connector, DB])),
|
list_to_existing_atom(io_lib:format("~ts_~ts", [emqx_connector, DB])),
|
||||||
Config,
|
Config,
|
||||||
#{waiting_connect_complete => 5000}) of
|
#{}) of
|
||||||
{ok, already_created} ->
|
{ok, already_created} ->
|
||||||
Context#{resource_id => ResourceID};
|
Context#{resource_id => ResourceID};
|
||||||
{ok, _} ->
|
{ok, _} ->
|
||||||
|
|
Loading…
Reference in New Issue