refactor(bridge): rename the APIs for emqx_bridge

This commit is contained in:
Shawn 2021-10-26 14:34:23 +08:00
parent 1e6884ee7d
commit 7996d8f709
4 changed files with 51 additions and 45 deletions

View File

@ -20,26 +20,13 @@
-export([post_config_update/4]). -export([post_config_update/4]).
-export([reload_hook/0, unload_hook/0]). -export([ load_hook/0
, reload_hook/0
, unload_hook/0
]).
-export([on_message_publish/1]). -export([on_message_publish/1]).
-export([ load_bridges/0
, get_bridge/2
, get_bridge/3
, list_bridges/0
, create_bridge/3
, remove_bridge/3
, update_bridge/3
, start_bridge/2
, stop_bridge/2
, restart_bridge/2
, send_message/2
]).
-export([ config_key_path/0
]).
-export([ resource_type/1 -export([ resource_type/1
, bridge_type/1 , bridge_type/1
, resource_id/1 , resource_id/1
@ -47,9 +34,30 @@
, parse_bridge_id/1 , parse_bridge_id/1
]). ]).
-export([ load/0
, lookup/2
, lookup/3
, list/0
, create/3
, remove/3
, update/3
, start/2
, stop/2
, restart/2
]).
-export([ send_message/2
]).
-export([ config_key_path/0
]).
reload_hook() -> reload_hook() ->
unload_hook(), unload_hook(),
Bridges = emqx_conf:get([bridges], #{}), load_hook().
load_hook() ->
Bridges = emqx:get_config([bridges], #{}),
lists:foreach(fun({_Type, Bridge}) -> lists:foreach(fun({_Type, Bridge}) ->
lists:foreach(fun({_Name, BridgeConf}) -> lists:foreach(fun({_Name, BridgeConf}) ->
load_hook(BridgeConf) load_hook(BridgeConf)
@ -82,9 +90,7 @@ config_key_path() ->
[bridges]. [bridges].
resource_type(mqtt) -> emqx_connector_mqtt; resource_type(mqtt) -> emqx_connector_mqtt;
resource_type(<<"mqtt">>) -> emqx_connector_mqtt; resource_type(http) -> emqx_connector_http.
resource_type(http) -> emqx_connector_http;
resource_type(<<"http">>) -> emqx_connector_http.
bridge_type(emqx_connector_mqtt) -> mqtt; bridge_type(emqx_connector_mqtt) -> mqtt;
bridge_type(emqx_connector_http) -> http. bridge_type(emqx_connector_http) -> http.
@ -93,9 +99,9 @@ post_config_update(_Req, NewConf, OldConf, _AppEnv) ->
#{added := Added, removed := Removed, changed := Updated} #{added := Added, removed := Removed, changed := Updated}
= diff_confs(NewConf, OldConf), = diff_confs(NewConf, OldConf),
perform_bridge_changes([ perform_bridge_changes([
{fun remove_bridge/3, Removed}, {fun remove/3, Removed},
{fun create_bridge/3, Added}, {fun create/3, Added},
{fun update_bridge/3, Updated} {fun update/3, Updated}
]), ]),
reload_hook(). reload_hook().
@ -116,8 +122,8 @@ perform_bridge_changes([{Action, MapConfs} | Tasks], Result0) ->
end, Result0, MapConfs), end, Result0, MapConfs),
perform_bridge_changes(Tasks, Result). perform_bridge_changes(Tasks, Result).
load_bridges() -> load() ->
Bridges = emqx_conf:get([bridges], #{}), Bridges = emqx:get_config([bridges], #{}),
emqx_bridge_monitor:ensure_all_started(Bridges). emqx_bridge_monitor:ensure_all_started(Bridges).
resource_id(BridgeId) when is_binary(BridgeId) -> resource_id(BridgeId) when is_binary(BridgeId) ->
@ -138,36 +144,36 @@ parse_bridge_id(BridgeId) ->
_ -> error({invalid_bridge_id, BridgeId}) _ -> error({invalid_bridge_id, BridgeId})
end. end.
list_bridges() -> list() ->
lists:foldl(fun({Type, NameAndConf}, Bridges) -> lists:foldl(fun({Type, NameAndConf}, Bridges) ->
lists:foldl(fun({Name, RawConf}, Acc) -> lists:foldl(fun({Name, RawConf}, Acc) ->
case get_bridge(Type, Name, RawConf) of case lookup(Type, Name, RawConf) of
{error, not_found} -> Acc; {error, not_found} -> Acc;
{ok, Res} -> [Res | Acc] {ok, Res} -> [Res | Acc]
end end
end, Bridges, maps:to_list(NameAndConf)) end, Bridges, maps:to_list(NameAndConf))
end, [], maps:to_list(emqx:get_raw_config([bridges], #{}))). end, [], maps:to_list(emqx:get_raw_config([bridges], #{}))).
get_bridge(Type, Name) -> lookup(Type, Name) ->
RawConf = emqx:get_raw_config([bridges, Type, Name], #{}), RawConf = emqx:get_raw_config([bridges, Type, Name], #{}),
get_bridge(Type, Name, RawConf). lookup(Type, Name, RawConf).
get_bridge(Type, Name, RawConf) -> lookup(Type, Name, RawConf) ->
case emqx_resource:get_instance(resource_id(Type, Name)) of case emqx_resource:get_instance(resource_id(Type, Name)) of
{error, not_found} -> {error, not_found}; {error, not_found} -> {error, not_found};
{ok, Data} -> {ok, #{id => bridge_id(Type, Name), resource_data => Data, {ok, Data} -> {ok, #{id => bridge_id(Type, Name), resource_data => Data,
raw_config => RawConf}} raw_config => RawConf}}
end. end.
start_bridge(Type, Name) -> start(Type, Name) ->
restart_bridge(Type, Name). restart(Type, Name).
stop_bridge(Type, Name) -> stop(Type, Name) ->
emqx_resource:stop(resource_id(Type, Name)). emqx_resource:stop(resource_id(Type, Name)).
restart_bridge(Type, Name) -> restart(Type, Name) ->
emqx_resource:restart(resource_id(Type, Name)). emqx_resource:restart(resource_id(Type, Name)).
create_bridge(Type, Name, Conf) -> create(Type, Name, Conf) ->
?SLOG(info, #{msg => "create bridge", type => Type, name => Name, ?SLOG(info, #{msg => "create bridge", type => Type, name => Name,
config => Conf}), config => Conf}),
ResId = resource_id(Type, Name), ResId = resource_id(Type, Name),
@ -181,7 +187,7 @@ create_bridge(Type, Name, Conf) ->
{error, Reason} {error, Reason}
end. end.
update_bridge(Type, Name, {_OldConf, Conf}) -> update(Type, Name, {_OldConf, Conf}) ->
%% TODO: sometimes its not necessary to restart the bridge connection. %% TODO: sometimes its not necessary to restart the bridge connection.
%% %%
%% - if the connection related configs like `servers` is updated, we should restart/start %% - if the connection related configs like `servers` is updated, we should restart/start
@ -195,7 +201,7 @@ update_bridge(Type, Name, {_OldConf, Conf}) ->
emqx_resource:recreate(resource_id(Type, Name), emqx_resource:recreate(resource_id(Type, Name),
emqx_bridge:resource_type(Type), Conf, []). emqx_bridge:resource_type(Type), Conf, []).
remove_bridge(Type, Name, _Conf) -> remove(Type, Name, _Conf) ->
?SLOG(info, #{msg => "remove bridge", type => Type, name => Name}), ?SLOG(info, #{msg => "remove bridge", type => Type, name => Name}),
case emqx_resource:remove(resource_id(Type, Name)) of case emqx_resource:remove(resource_id(Type, Name)) of
ok -> ok; ok -> ok;

View File

@ -157,7 +157,7 @@ list_bridges(get, _Params) ->
{200, lists:append([list_local_bridges(Node) || Node <- mria_mnesia:running_nodes()])}. {200, lists:append([list_local_bridges(Node) || Node <- mria_mnesia:running_nodes()])}.
list_local_bridges(Node) when Node =:= node() -> list_local_bridges(Node) when Node =:= node() ->
[format_resp(Data) || Data <- emqx_bridge:list_bridges()]; [format_resp(Data) || Data <- emqx_bridge:list()];
list_local_bridges(Node) -> list_local_bridges(Node) ->
rpc_call(Node, list_local_bridges, [Node]). rpc_call(Node, list_local_bridges, [Node]).
@ -177,7 +177,7 @@ crud_bridges(Node, Method, Params) when Node =/= node() ->
rpc_call(Node, crud_bridges, [Node, Method, Params]); rpc_call(Node, crud_bridges, [Node, Method, Params]);
crud_bridges(_, get, #{bindings := #{id := Id}}) -> crud_bridges(_, get, #{bindings := #{id := Id}}) ->
?TRY_PARSE_ID(Id, case emqx_bridge:get_bridge(BridgeType, BridgeName) of ?TRY_PARSE_ID(Id, case emqx_bridge:lookup(BridgeType, BridgeName) of
{ok, Data} -> {200, format_resp(Data)}; {ok, Data} -> {200, format_resp(Data)};
{error, not_found} -> {error, not_found} ->
{404, #{code => 102, message => <<"not_found: ", Id/binary>>}} {404, #{code => 102, message => <<"not_found: ", Id/binary>>}}
@ -190,7 +190,7 @@ crud_bridges(_, put, #{bindings := #{id := Id}, body := Conf}) ->
{ok, #{raw_config := RawConf, post_config_update := #{emqx_bridge := Data}}} -> {ok, #{raw_config := RawConf, post_config_update := #{emqx_bridge := Data}}} ->
{200, format_resp(#{id => Id, raw_config => RawConf, resource_data => Data})}; {200, format_resp(#{id => Id, raw_config => RawConf, resource_data => Data})};
{ok, _} -> %% the bridge already exits {ok, _} -> %% the bridge already exits
{ok, Data} = emqx_bridge:get_bridge(BridgeType, BridgeName), {ok, Data} = emqx_bridge:lookup(BridgeType, BridgeName),
{200, format_resp(Data)}; {200, format_resp(Data)};
{error, Reason} -> {error, Reason} ->
{500, #{code => 102, message => emqx_resource_api:stringnify(Reason)}} {500, #{code => 102, message => emqx_resource_api:stringnify(Reason)}}

View File

@ -21,9 +21,9 @@
start(_StartType, _StartArgs) -> start(_StartType, _StartArgs) ->
{ok, Sup} = emqx_bridge_sup:start_link(), {ok, Sup} = emqx_bridge_sup:start_link(),
ok = emqx_bridge:load_bridges(), ok = emqx_bridge:load(),
ok = emqx_bridge:reload_hook(), ok = emqx_bridge:load_hook(),
emqx_conf:add_handler(emqx_bridge:config_key_path(), emqx_bridge), emqx_config_handler:add_handler(emqx_bridge:config_key_path(), emqx_bridge),
{ok, Sup}. {ok, Sup}.
stop(_State) -> stop(_State) ->

View File

@ -67,6 +67,6 @@ code_change(_OldVsn, State, _Extra) ->
load_bridges(Configs) -> load_bridges(Configs) ->
lists:foreach(fun({Type, NamedConf}) -> lists:foreach(fun({Type, NamedConf}) ->
lists:foreach(fun({Name, Conf}) -> lists:foreach(fun({Name, Conf}) ->
emqx_bridge:create_bridge(Type, Name, Conf) emqx_bridge:create(Type, Name, Conf)
end, maps:to_list(NamedConf)) end, maps:to_list(NamedConf))
end, maps:to_list(Configs)). end, maps:to_list(Configs)).