f199a0f24a
This commit: * refactors the MQTT V1 bridge into connector, source and action * Extends the compatibility layer so it works for sources * Fixes the MQTT bridge test suite so that all test cases passes We still need to add a HTTP API handling sources. Also, we still need to add HTTP API example schemes and examples for the MQTT connector/action/source. We should also make sure that we handle the corner cases of the MQTT V1 bridge automatic upgrade downgrade in a sufficiently good way: * An error is currently thrown when converting an MQTT V1 bridge without egress or ingress config. * If there is a source and action with the same name we will currently throw an error in the compatibility layer. * We will also throw an error when converting an MQTT V1 bridge with both ingress and egress. The above is probably the right thing to do but we have to make sure that we return a reasonable error to the user when this happens. (partly) Fixes: https://emqx.atlassian.net/browse/EMQX-11489
100 lines
4.0 KiB
Erlang
100 lines
4.0 KiB
Erlang
%%--------------------------------------------------------------------
|
|
%% Copyright (c) 2020-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
%%
|
|
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
%% you may not use this file except in compliance with the License.
|
|
%% You may obtain a copy of the License at
|
|
%%
|
|
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
%%
|
|
%% Unless required by applicable law or agreed to in writing, software
|
|
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
%% See the License for the specific language governing permissions and
|
|
%% limitations under the License.
|
|
%%--------------------------------------------------------------------
|
|
-module(emqx_bridge_lib).
|
|
|
|
-export([
|
|
maybe_withdraw_rule_action/3,
|
|
upgrade_type/1,
|
|
downgrade_type/2,
|
|
get_conf/2
|
|
]).
|
|
|
|
%% @doc A bridge can be used as a rule action.
|
|
%% The bridge-ID in rule-engine's world is the action-ID.
|
|
%% This function is to remove a bridge (action) from all rules
|
|
%% using it if the `rule_actions' is included in `DeleteDeps' list
|
|
maybe_withdraw_rule_action(BridgeType, BridgeName, DeleteDeps) ->
|
|
BridgeIds = external_ids(BridgeType, BridgeName),
|
|
DeleteActions = lists:member(rule_actions, DeleteDeps),
|
|
maybe_withdraw_rule_action_loop(BridgeIds, DeleteActions).
|
|
|
|
maybe_withdraw_rule_action_loop([], _DeleteActions) ->
|
|
ok;
|
|
maybe_withdraw_rule_action_loop([BridgeId | More], DeleteActions) ->
|
|
case emqx_rule_engine:get_rule_ids_by_action(BridgeId) of
|
|
[] ->
|
|
maybe_withdraw_rule_action_loop(More, DeleteActions);
|
|
RuleIds when DeleteActions ->
|
|
lists:foreach(
|
|
fun(R) ->
|
|
emqx_rule_engine:ensure_action_removed(R, BridgeId)
|
|
end,
|
|
RuleIds
|
|
),
|
|
maybe_withdraw_rule_action_loop(More, DeleteActions);
|
|
RuleIds ->
|
|
{error, #{
|
|
reason => rules_depending_on_this_bridge,
|
|
bridge_id => BridgeId,
|
|
rule_ids => RuleIds
|
|
}}
|
|
end.
|
|
|
|
%% @doc Kafka producer bridge renamed from 'kafka' to 'kafka_bridge' since 5.3.1.
|
|
upgrade_type(Type) when is_atom(Type) ->
|
|
emqx_bridge_v2:bridge_v1_type_to_bridge_v2_type(Type);
|
|
upgrade_type(Type) when is_binary(Type) ->
|
|
atom_to_binary(emqx_bridge_v2:bridge_v1_type_to_bridge_v2_type(Type));
|
|
upgrade_type(Type) when is_list(Type) ->
|
|
atom_to_list(emqx_bridge_v2:bridge_v1_type_to_bridge_v2_type(list_to_binary(Type))).
|
|
|
|
%% @doc Kafka producer bridge type renamed from 'kafka' to 'kafka_bridge' since 5.3.1
|
|
downgrade_type(Type, Conf) when is_atom(Type) ->
|
|
emqx_bridge_v2:bridge_v2_type_to_bridge_v1_type(Type, Conf);
|
|
downgrade_type(Type, Conf) when is_binary(Type) ->
|
|
atom_to_binary(emqx_bridge_v2:bridge_v2_type_to_bridge_v1_type(Type, Conf));
|
|
downgrade_type(Type, Conf) when is_list(Type) ->
|
|
atom_to_list(emqx_bridge_v2:bridge_v2_type_to_bridge_v1_type(list_to_binary(Type), Conf)).
|
|
|
|
%% A rule might be referencing an old version bridge type name
|
|
%% i.e. 'kafka' instead of 'kafka_producer' so we need to try both
|
|
external_ids(Type, Name) ->
|
|
case downgrade_type(Type, get_conf(Type, Name)) of
|
|
Type ->
|
|
[external_id(Type, Name)];
|
|
Type0 ->
|
|
[external_id(Type0, Name), external_id(Type, Name)]
|
|
end.
|
|
|
|
get_conf(BridgeType, BridgeName) ->
|
|
case emqx_bridge_v2:is_bridge_v2_type(BridgeType) of
|
|
true ->
|
|
ConfRootName = emqx_bridge_v2:get_conf_root_key_if_only_one(BridgeType, BridgeName),
|
|
emqx_conf:get_raw([ConfRootName, BridgeType, BridgeName]);
|
|
false ->
|
|
undefined
|
|
end.
|
|
|
|
%% Creates the external id for the bridge_v2 that is used by the rule actions
|
|
%% to refer to the bridge_v2
|
|
external_id(BridgeType, BridgeName) ->
|
|
Name = bin(BridgeName),
|
|
Type = bin(BridgeType),
|
|
<<Type/binary, ":", Name/binary>>.
|
|
|
|
bin(Bin) when is_binary(Bin) -> Bin;
|
|
bin(Atom) when is_atom(Atom) -> atom_to_binary(Atom, utf8).
|