diff --git a/apps/emqx_bridge/src/emqx_bridge_v2.erl b/apps/emqx_bridge/src/emqx_bridge_v2.erl index 947c5e06c..f2459799b 100644 --- a/apps/emqx_bridge/src/emqx_bridge_v2.erl +++ b/apps/emqx_bridge/src/emqx_bridge_v2.erl @@ -70,7 +70,8 @@ -export([ bridge_v2_type_to_connector_type/1, id/2, - id/3 + id/3, + is_valid_bridge_v1/2 ]). %% Config Update Handler API @@ -752,8 +753,30 @@ unpack_bridge_conf(Type, PackedConf, TopLevelConf) -> %%==================================================================== %% Compatibility API %%==================================================================== + +%% Check if the bridge can be converted to a valid bridge v1 +%% +%% * The corresponding bridge v2 should exist +%% * The connector for the bridge v2 should have exactly on channel +is_valid_bridge_v1(BridgeV1Type, BridgeName) -> + BridgeV2Type = ?MODULE:bridge_v1_type_to_bridge_v2_type(BridgeV1Type), + case lookup_raw_conf(BridgeV2Type, BridgeName) of + {error, _} -> + false; + #{connector := ConnectorName} -> + ConnectorType = ?MODULE:bridge_v2_type_to_connector_type(BridgeV2Type), + ConnectorResourceId = emqx_connector_resource:resource_id(ConnectorType, ConnectorName), + {ok, Channels} = emqx_resource:get_channels(ConnectorResourceId), + case Channels of + [_Channel] -> + true; + _ -> + false + end + end. + bridge_v1_type_to_bridge_v2_type(Bin) when is_binary(Bin) -> - bridge_v1_type_to_bridge_v2_type(binary_to_existing_atom(Bin)); + ?MODULE:bridge_v1_type_to_bridge_v2_type(binary_to_existing_atom(Bin)); bridge_v1_type_to_bridge_v2_type(kafka) -> kafka. @@ -985,7 +1008,7 @@ connector_has_channels(BridgeV2Type, ConnectorName) -> bridge_v1_id_to_connector_resource_id(BridgeId) -> case binary:split(BridgeId, <<":">>) of [Type, Name] -> - BridgeV2Type = bin(bridge_v1_type_to_bridge_v2_type(Type)), + BridgeV2Type = bin(?MODULE:bridge_v1_type_to_bridge_v2_type(Type)), ConnectorName = case lookup_raw_conf(BridgeV2Type, Name) of #{connector := Con} -> diff --git a/apps/emqx_bridge/test/emqx_bridge_v2_SUITE.erl b/apps/emqx_bridge/test/emqx_bridge_v2_SUITE.erl index b64d7e92e..a0786d91a 100644 --- a/apps/emqx_bridge/test/emqx_bridge_v2_SUITE.erl +++ b/apps/emqx_bridge/test/emqx_bridge_v2_SUITE.erl @@ -66,7 +66,7 @@ bridge_schema() -> bridge_config() -> #{ - <<"connector">> => con_name() + <<"connector">> => atom_to_binary(con_name()) }. all() -> @@ -87,6 +87,7 @@ init_per_suite(Config) -> meck:new(emqx_bridge_v2, [passthrough, no_link]), meck:expect(emqx_bridge_v2, bridge_v2_type_to_connector_type, 1, con_type()), + meck:expect(emqx_bridge_v2, bridge_v1_type_to_bridge_v2_type, 1, bridge_type()), _ = application:load(emqx_conf), ok = emqx_common_test_helpers:start_apps(start_apps()), @@ -166,3 +167,15 @@ t_create_dry_run_fail_get_channel_status(_) -> t_create_dry_run_connector_does_not_exist(_) -> BridgeConf = (bridge_config())#{<<"connector">> => <<"connector_does_not_exist">>}, {error, _} = emqx_bridge_v2:create_dry_run(bridge_type(), BridgeConf). + +t_is_valid_bridge_v1(_) -> + {ok, _} = emqx_bridge_v2:create(bridge_type(), my_test_bridge, bridge_config()), + true = emqx_bridge_v2:is_valid_bridge_v1(bridge_v1_type, my_test_bridge), + %% Add another channel/bridge to the connector + {ok, _} = emqx_bridge_v2:create(bridge_type(), my_test_bridge_2, bridge_config()), + false = emqx_bridge_v2:is_valid_bridge_v1(bridge_v1_type, my_test_bridge), + {ok, _} = emqx_bridge_v2:remove(bridge_type(), my_test_bridge), + true = emqx_bridge_v2:is_valid_bridge_v1(bridge_v1_type, my_test_bridge_2), + {ok, _} = emqx_bridge_v2:remove(bridge_type(), my_test_bridge_2), + false = emqx_bridge_v2:is_valid_bridge_v1(bridge_v1_type, my_test_bridge), + ok. diff --git a/apps/emqx_resource/src/emqx_resource.erl b/apps/emqx_resource/src/emqx_resource.erl index eab64774d..af52cf17b 100644 --- a/apps/emqx_resource/src/emqx_resource.erl +++ b/apps/emqx_resource/src/emqx_resource.erl @@ -449,7 +449,7 @@ health_check(ResId) -> channel_health_check(ResId, ChannelId) -> emqx_resource_manager:channel_health_check(ResId, ChannelId). --spec get_channels(resource_id()) -> [{binary(), map()}]. +-spec get_channels(resource_id()) -> {ok, [{binary(), map()}]}. get_channels(ResId) -> emqx_resource_manager:get_channels(ResId).