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

Merge pull request #6869 from EMQ-YangM/fix_start_pool 

fix(emqx_connector): when start_pool fails, return error and reason
This commit is contained in:
zhongwencool 2022-01-29 10:45:45 +08:00 committed by GitHub
parent 5e01961dba e9f3fa7b6b
commit 928555546d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 27 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
apps/emqx_connector/src/emqx_connector_ldap.erl
View File

@ -77,8 +77,10 @@ on_start(InstId, #{servers := Servers0,
{auto_reconnect, reconn_interval(AutoReconn)},
{servers, Servers}],
PoolName = emqx_plugin_libs_pool:pool_name(InstId),
_ = emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Opts ++ SslOpts),
{ok, #{poolname => PoolName}}.
case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Opts ++ SslOpts) of
ok -> {ok, #{poolname => PoolName}};
{error, Reason} -> {error, Reason}
end.
on_stop(InstId, #{poolname := PoolName}) ->
?SLOG(info, #{msg => "stopping_ldap_connector",

6
apps/emqx_connector/src/emqx_connector_mongo.erl
View File

@ -130,8 +130,10 @@ on_start(InstId, Config = #{mongo_type := Type,
{options, init_topology_options(maps:to_list(Topology), [])},
{worker_options, init_worker_options(maps:to_list(NConfig), SslOpts)}],
PoolName = emqx_plugin_libs_pool:pool_name(InstId),
ok = emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Opts),
{ok, #{poolname => PoolName, type => Type}}.
case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Opts) of
ok -> {ok, #{poolname => PoolName, type => Type}};
{error, Reason} -> {error, Reason}
end.
on_stop(InstId, #{poolname := PoolName}) ->
?SLOG(info, #{msg => "stopping_mongodb_connector",

6
apps/emqx_connector/src/emqx_connector_mysql.erl
View File

@ -71,8 +71,10 @@ on_start(InstId, #{server := {Host, Port},
{auto_reconnect, reconn_interval(AutoReconn)},
{pool_size, PoolSize}],
PoolName = emqx_plugin_libs_pool:pool_name(InstId),
_ = emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Options ++ SslOpts),
{ok, #{poolname => PoolName}}.
case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Options ++ SslOpts) of
ok -> {ok, #{poolname => PoolName}};
{error, Reason} -> {error, Reason}
end.
on_stop(InstId, #{poolname := PoolName}) ->
?SLOG(info, #{msg => "stopping_mysql_connector",

6
apps/emqx_connector/src/emqx_connector_pgsql.erl
View File

@ -83,8 +83,10 @@ on_start(InstId, #{server := {Host, Port},
{pool_size, PoolSize},
{named_queries, maps:to_list(maps:get(named_queries, Config, #{}))}],
PoolName = emqx_plugin_libs_pool:pool_name(InstId),
_ = emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Options ++ SslOpts),
{ok, #{poolname => PoolName}}.
case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Options ++ SslOpts) of
ok -> {ok, #{poolname => PoolName}};
{error, Reason} -> {error, Reason}
end.
on_stop(InstId, #{poolname := PoolName}) ->
?SLOG(info, #{msg => "stopping postgresql connector",

14
apps/emqx_connector/src/emqx_connector_redis.erl
View File

@ -117,14 +117,16 @@ on_start(InstId, #{redis_type := Type,
case Type of
cluster ->
case eredis_cluster:start_pool(PoolName, Opts ++ [{options, Options}]) of
{ok, _} -> ok;
{ok, _, _} -> ok;
{error, Reason} -> error(connect_redis_cluster_failed, Reason)
{ok, _} -> {ok, #{poolname => PoolName, type => Type}};
{ok, _, _} -> {ok, #{poolname => PoolName, type => Type}};
{error, Reason} -> {error, Reason}
end;
_ ->
_ = emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Opts ++ [{options, Options}])
end,
{ok, #{poolname => PoolName, type => Type}}.
case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Opts ++ [{options, Options}]) of
ok -> {ok, #{poolname => PoolName, type => Type}};
{error, Reason} -> {error, Reason}
end
end.
on_stop(InstId, #{poolname := PoolName}) ->
?SLOG(info, #{msg => "stopping_redis_connector",

5
apps/emqx_plugin_libs/src/emqx_plugin_libs_pool.erl
View File

@ -30,14 +30,15 @@ pool_name(ID) when is_binary(ID) ->
start_pool(Name, Mod, Options) ->
case ecpool:start_sup_pool(Name, Mod, Options) of
{ok, _} ->
?SLOG(info, #{msg => "start_ecpool_ok", pool_name => Name});
?SLOG(info, #{msg => "start_ecpool_ok", pool_name => Name}),
ok;
{error, {already_started, _Pid}} ->
stop_pool(Name),
start_pool(Name, Mod, Options);
{error, Reason} ->
?SLOG(error, #{msg => "start_ecpool_error", pool_name => Name,
reason => Reason}),
error({start_pool_failed, Name, Reason})
{error, {start_pool_failed, Name, Reason}}
end.
stop_pool(Name) ->