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

feat: refactored some bridges to avoid leaking resources during crashes at creation

This commit is contained in:
firest 2023-05-31 16:48:13 +08:00
parent e6049df724
commit 6081ce8b00
8 changed files with 64 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt_connector.erl
View File

@ -83,6 +83,7 @@ start_ingress(ResourceId, Ingress, ClientOpts) ->
{ingress, Ingress},
{client_opts, ClientOpts}
],
ok = emqx_resource:allocate_resource(ResourceId, ingress_pool_name, PoolName),
case emqx_resource_pool:start(PoolName, emqx_bridge_mqtt_ingress, Options) of
ok ->
{ok, #{ingress_pool_name => PoolName}};
@ -132,6 +133,7 @@ start_egress(ResourceId, Egress, ClientOpts) ->
{pool_size, PoolSize},
{client_opts, ClientOpts}
],
ok = emqx_resource:allocate_resource(ResourceId, egress_pool_name, PoolName),
case emqx_resource_pool:start(PoolName, emqx_bridge_mqtt_egress, Options) of
ok ->
{ok, #{
@ -142,13 +144,14 @@ start_egress(ResourceId, Egress, ClientOpts) ->
{error, Reason}
end.
on_stop(ResourceId, State) ->
on_stop(ResourceId, _State) ->
?SLOG(info, #{
msg => "stopping_mqtt_connector",
connector => ResourceId
}),
ok = stop_ingress(State),
ok = stop_egress(State).
Allocated = emqx_resource:get_allocated_resources(ResourceId),
ok = stop_ingress(Allocated),
ok = stop_egress(Allocated).
stop_ingress(#{ingress_pool_name := PoolName}) ->
emqx_resource_pool:stop(PoolName);

10
apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine_connector.erl
View File

@ -108,6 +108,7 @@ on_start(
Prepares = parse_prepare_sql(Config),
State = Prepares#{pool_name => InstanceId, query_opts => query_opts(Config)},
ok = emqx_resource:allocate_resource(InstanceId, pool_name, InstanceId),
case emqx_resource_pool:start(InstanceId, ?MODULE, Options) of
ok ->
{ok, State};
@ -115,12 +116,17 @@ on_start(
Error
end.
on_stop(InstanceId, #{pool_name := PoolName}) ->
on_stop(InstanceId, _State) ->
?SLOG(info, #{
msg => "stopping_tdengine_connector",
connector => InstanceId
}),
emqx_resource_pool:stop(PoolName).
case emqx_resource:get_allocated_resources(InstanceId) of
#{pool_name := PoolName} ->
emqx_resource_pool:stop(PoolName);
_ ->
ok
end.
on_query(InstanceId, {query, SQL}, State) ->
do_query(InstanceId, SQL, State);

10
apps/emqx_connector/src/emqx_connector_http.erl
View File

@ -219,18 +219,24 @@ on_start(
base_path => BasePath,
request => preprocess_request(maps:get(request, Config, undefined))
},
ok = emqx_resource:allocate_resource(InstId, pool_name, InstId),
case ehttpc_sup:start_pool(InstId, PoolOpts) of
{ok, _} -> {ok, State};
{error, {already_started, _}} -> {ok, State};
{error, Reason} -> {error, Reason}
end.
on_stop(InstId, #{pool_name := PoolName}) ->
on_stop(InstId, _State) ->
?SLOG(info, #{
msg => "stopping_http_connector",
connector => InstId
}),
ehttpc_sup:stop_pool(PoolName).
case emqx_resource:get_allocated_resources(InstId) of
#{pool_name := PoolName} ->
ehttpc_sup:stop_pool(PoolName);
_ ->
ok
end.
on_query(InstId, {send_message, Msg}, State) ->
case maps:get(request, State, undefined) of

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

@ -97,17 +97,23 @@ on_start(
{pool_size, PoolSize},
{auto_reconnect, ?AUTO_RECONNECT_INTERVAL}
],
ok = emqx_resource:allocate_resource(InstId, pool_name, InstId),
case emqx_resource_pool:start(InstId, ?MODULE, Opts ++ SslOpts) of
ok -> {ok, #{pool_name => InstId}};
{error, Reason} -> {error, Reason}
end.
on_stop(InstId, #{pool_name := PoolName}) ->
on_stop(InstId, _State) ->
?SLOG(info, #{
msg => "stopping_ldap_connector",
connector => InstId
}),
emqx_resource_pool:stop(PoolName).
case emqx_resource:get_allocated_resources(InstId) of
#{pool_name := PoolName} ->
emqx_resource_pool:stop(PoolName);
_ ->
ok
end.
on_query(InstId, {search, Base, Filter, Attributes}, #{pool_name := PoolName} = State) ->
Request = {Base, Filter, Attributes},

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

@ -183,6 +183,7 @@ on_start(
{worker_options, init_worker_options(maps:to_list(NConfig), SslOpts)}
],
Collection = maps:get(collection, Config, <<"mqtt">>),
ok = emqx_resource:allocate_resource(InstId, pool_name, InstId),
case emqx_resource_pool:start(InstId, ?MODULE, Opts) of
ok ->
{ok, #{
@ -194,12 +195,17 @@ on_start(
{error, Reason}
end.
on_stop(InstId, #{pool_name := PoolName}) ->
on_stop(InstId, _State) ->
?SLOG(info, #{
msg => "stopping_mongodb_connector",
connector => InstId
}),
emqx_resource_pool:stop(PoolName).
case emqx_resource:get_allocated_resources(InstId) of
#{pool_name := PoolName} ->
emqx_resource_pool:stop(PoolName);
_ ->
ok
end.
on_query(
InstId,

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

@ -124,6 +124,7 @@ on_start(
]
),
State = parse_prepare_sql(Config),
ok = emqx_resource:allocate_resource(InstId, pool_name, InstId),
case emqx_resource_pool:start(InstId, ?MODULE, Options ++ SslOpts) of
ok ->
{ok, init_prepare(State#{pool_name => InstId})};
@ -140,12 +141,17 @@ maybe_add_password_opt(undefined, Options) ->
maybe_add_password_opt(Password, Options) ->
[{password, Password} | Options].
on_stop(InstId, #{pool_name := PoolName}) ->
on_stop(InstId, _State) ->
?SLOG(info, #{
msg => "stopping_mysql_connector",
connector => InstId
}),
emqx_resource_pool:stop(PoolName).
case emqx_resource:get_allocated_resources(InstId) of
#{pool_name := PoolName} ->
emqx_resource_pool:stop(PoolName);
_ ->
ok
end.
on_query(InstId, {TypeOrKey, SQLOrKey}, State) ->
on_query(InstId, {TypeOrKey, SQLOrKey, [], default_timeout}, State);

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

@ -121,6 +121,7 @@ on_start(
{pool_size, PoolSize}
],
State = parse_prepare_sql(Config),
ok = emqx_resource:allocate_resource(InstId, pool_name, InstId),
case emqx_resource_pool:start(InstId, ?MODULE, Options ++ SslOpts) of
ok ->
{ok, init_prepare(State#{pool_name => InstId, prepare_statement => #{}})};
@ -132,12 +133,17 @@ on_start(
{error, Reason}
end.
on_stop(InstId, #{pool_name := PoolName}) ->
on_stop(InstId, _State) ->
?SLOG(info, #{
msg => "stopping postgresql connector",
connector => InstId
}),
emqx_resource_pool:stop(PoolName).
case emqx_resource:get_allocated_resources(InstId) of
#{pool_name := PoolName} ->
emqx_resource_pool:stop(PoolName);
_ ->
ok
end.
on_query(InstId, {TypeOrKey, NameOrSQL}, State) ->
on_query(InstId, {TypeOrKey, NameOrSQL, []}, State);

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

@ -160,6 +160,8 @@ on_start(
[{ssl, false}]
end ++ [{sentinel, maps:get(sentinel, Config, undefined)}],
State = #{pool_name => InstId, type => Type},
ok = emqx_resource:allocate_resource(InstId, type, Type),
ok = emqx_resource:allocate_resource(InstId, pool_name, InstId),
case Type of
cluster ->
case eredis_cluster:start_pool(InstId, Opts ++ [{options, Options}]) of
@ -177,14 +179,18 @@ on_start(
end
end.
on_stop(InstId, #{pool_name := PoolName, type := Type}) ->
on_stop(InstId, _State) ->
?SLOG(info, #{
msg => "stopping_redis_connector",
connector => InstId
}),
case Type of
cluster -> eredis_cluster:stop_pool(PoolName);
_ -> emqx_resource_pool:stop(PoolName)
case emqx_resource:get_allocated_resources(InstId) of
#{pool_name := PoolName, type := cluster} ->
eredis_cluster:stop_pool(PoolName);
#{pool_name := PoolName, type := _} ->
emqx_resource_pool:stop(PoolName);
_ ->
ok
end.
on_query(InstId, {cmd, _} = Query, State) ->