Merge pull request #12486 from thalesmg/gcp-consumer-source-m-20240205

feat: convert `gcp_pubsub_consumer` to connector/source

apps/emqx_bridge/src/emqx_action_info.erl

@@ -88,6 +88,7 @@ hard_coded_action_info_modules_ee() ->
     [
         emqx_bridge_azure_event_hub_action_info,
         emqx_bridge_confluent_producer_action_info,
+        emqx_bridge_gcp_pubsub_consumer_action_info,
         emqx_bridge_gcp_pubsub_producer_action_info,
         emqx_bridge_kafka_action_info,
         emqx_bridge_kinesis_action_info,

apps/emqx_bridge/src/emqx_bridge_v2.erl

@@ -27,7 +27,9 @@
 %% Note: this is strange right now, because it lives in `emqx_bridge_v2', but it shall be
 %% refactored into a new module/application with appropriate name.
 -define(ROOT_KEY_ACTIONS, actions).
+-define(ROOT_KEY_ACTIONS_BIN, <<"actions">>).
 -define(ROOT_KEY_SOURCES, sources).
+-define(ROOT_KEY_SOURCES_BIN, <<"sources">>).
 
 %% Loading and unloading config when EMQX starts and stops
 -export([
@@ -91,6 +93,8 @@
 -export([
     id/2,
     id/3,
+    source_id/3,
+    source_hookpoint/1,
     bridge_v1_is_valid/2,
     bridge_v1_is_valid/3,
     extract_connector_id_from_bridge_v2_id/1
@@ -475,7 +479,7 @@ augment_channel_config(
         bridge_type => bin(BridgeV2Type),
         bridge_name => bin(BridgeName)
     },
-    case emqx_action_info:is_source(BridgeV2Type) of
+    case emqx_action_info:is_source(BridgeV2Type) andalso ConfigRoot =:= ?ROOT_KEY_SOURCES of
         true ->
             BId = emqx_bridge_resource:bridge_id(BridgeV2Type, BridgeName),
             BridgeHookpoint = emqx_bridge_resource:bridge_hookpoint(BId),
@@ -681,7 +685,8 @@ do_query_with_enabled_config(
 ) ->
     QueryMode = get_query_mode(BridgeType, Config),
     ConnectorName = maps:get(connector, Config),
-    ConnectorResId = emqx_connector_resource:resource_id(BridgeType, ConnectorName),
+    ConnectorType = emqx_action_info:action_type_to_connector_type(BridgeType),
+    ConnectorResId = emqx_connector_resource:resource_id(ConnectorType, ConnectorName),
     QueryOpts = maps:merge(
         emqx_bridge:query_opts(Config),
         QueryOpts0#{
@@ -750,7 +755,9 @@ create_dry_run(ConfRootKey, Type, Conf0) ->
             not_found ->
                 {error, iolist_to_binary(io_lib:format("Connector ~p not found", [ConnectorName]))};
             ConnectorRawConf ->
-                create_dry_run_helper(Type, ConnectorRawConf, Conf1)
+                create_dry_run_helper(
+                    ensure_atom_root_key(ConfRootKey), Type, ConnectorRawConf, Conf1
+                )
         end
     catch
         %% validation errors
@@ -758,9 +765,6 @@ create_dry_run(ConfRootKey, Type, Conf0) ->
             {error, Reason1}
     end.
 
-create_dry_run_helper(BridgeType, ConnectorRawConf, BridgeV2RawConf) ->
-    create_dry_run_helper(?ROOT_KEY_ACTIONS, BridgeType, ConnectorRawConf, BridgeV2RawConf).
-
 create_dry_run_helper(ConfRootKey, BridgeType, ConnectorRawConf, BridgeV2RawConf) ->
     BridgeName = iolist_to_binary([?TEST_ID_PREFIX, emqx_utils:gen_id(8)]),
     ConnectorType = connector_type(BridgeType),
@@ -969,6 +973,9 @@ id(BridgeType, BridgeName) ->
 id(BridgeType, BridgeName, ConnectorName) ->
     id_with_root_name(?ROOT_KEY_ACTIONS, BridgeType, BridgeName, ConnectorName).
 
+source_id(BridgeType, BridgeName, ConnectorName) ->
+    id_with_root_name(?ROOT_KEY_SOURCES, BridgeType, BridgeName, ConnectorName).
+
 id_with_root_name(RootName, BridgeType, BridgeName) ->
     case lookup_conf(RootName, BridgeType, BridgeName) of
         #{connector := ConnectorName} ->
@@ -1637,6 +1644,7 @@ bridge_v1_create_dry_run(BridgeType, RawConfig0) ->
     PreviousRawConf = undefined,
     try
         #{
+            conf_root_key := ConfRootKey,
             connector_type := _ConnectorType,
             connector_name := _NewConnectorName,
             connector_conf := ConnectorRawConf,
@@ -1644,7 +1652,9 @@ bridge_v1_create_dry_run(BridgeType, RawConfig0) ->
             bridge_v2_name := _BridgeName,
             bridge_v2_conf := BridgeV2RawConf
         } = split_and_validate_bridge_v1_config(BridgeType, TmpName, RawConf, PreviousRawConf),
-        create_dry_run_helper(BridgeV2Type, ConnectorRawConf, BridgeV2RawConf)
+        create_dry_run_helper(
+            ensure_atom_root_key(ConfRootKey), BridgeV2Type, ConnectorRawConf, BridgeV2RawConf
+        )
     catch
         throw:Reason ->
             {error, Reason}
@@ -1862,6 +1872,13 @@ extract_connector_id_from_bridge_v2_id(Id) ->
             error({error, iolist_to_binary(io_lib:format("Invalid action ID: ~p", [Id]))})
     end.
 
+ensure_atom_root_key(ConfRootKey) when is_atom(ConfRootKey) ->
+    ConfRootKey;
+ensure_atom_root_key(?ROOT_KEY_ACTIONS_BIN) ->
+    ?ROOT_KEY_ACTIONS;
+ensure_atom_root_key(?ROOT_KEY_SOURCES_BIN) ->
+    ?ROOT_KEY_SOURCES.
+
 to_existing_atom(X) ->
     case emqx_utils:safe_to_existing_atom(X, utf8) of
         {ok, A} -> A;

apps/emqx_bridge/src/schema/emqx_bridge_v2_schema.erl

@@ -64,8 +64,9 @@
     make_producer_action_schema/1, make_producer_action_schema/2,
     make_consumer_action_schema/1, make_consumer_action_schema/2,
     top_level_common_action_keys/0,
+    top_level_common_source_keys/0,
     project_to_actions_resource_opts/1,
-    project_to_sources_resource_opts/1
+    project_to_sources_resource_opts/1, project_to_sources_resource_opts/2
 ]).
 
 -export([actions_convert_from_connectors/1]).
@@ -422,6 +423,16 @@ top_level_common_action_keys() ->
         <<"resource_opts">>
     ].
 
+top_level_common_source_keys() ->
+    [
+        <<"connector">>,
+        <<"tags">>,
+        <<"description">>,
+        <<"enable">>,
+        <<"parameters">>,
+        <<"resource_opts">>
+    ].
+
 %%======================================================================================
 %% Helper functions for making HOCON Schema
 %%======================================================================================
@@ -474,7 +485,9 @@ project_to_actions_resource_opts(OldResourceOpts) ->
     maps:with(Subfields, OldResourceOpts).
 
 project_to_sources_resource_opts(OldResourceOpts) ->
-    Subfields = common_source_resource_opts_subfields_bin(),
+    project_to_sources_resource_opts(OldResourceOpts, common_source_resource_opts_subfields_bin()).
+
+project_to_sources_resource_opts(OldResourceOpts, Subfields) ->
     maps:with(Subfields, OldResourceOpts).
 
 actions_convert_from_connectors(RawConf = #{<<"actions">> := Actions}) ->

apps/emqx_bridge/test/emqx_bridge_testlib.erl

@@ -105,9 +105,15 @@ parse_and_check(BridgeType, BridgeName, ConfigString) ->
     BridgeConfig.
 
 resource_id(Config) ->
+    BridgeKind = proplists:get_value(bridge_kind, Config, action),
+    ConfRootKey =
+        case BridgeKind of
+            action -> actions;
+            source -> sources
+        end,
     BridgeType = ?config(bridge_type, Config),
     BridgeName = ?config(bridge_name, Config),
-    emqx_bridge_resource:resource_id(BridgeType, BridgeName).
+    emqx_bridge_resource:resource_id(ConfRootKey, BridgeType, BridgeName).
 
 create_bridge(Config) ->
     create_bridge(Config, _Overrides = #{}).

apps/emqx_bridge/test/emqx_bridge_v2_testlib.erl

@@ -9,6 +9,7 @@
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
+-include_lib("emqx_resource/include/emqx_resource.hrl").
 
 -import(emqx_common_test_helpers, [on_exit/1]).
 
@@ -120,13 +121,29 @@ delete_all_connectors() ->
 
 %% test helpers
 parse_and_check(Type, Name, InnerConfigMap0) ->
+    parse_and_check(action, Type, Name, InnerConfigMap0).
+
+parse_and_check(Kind, Type, Name, InnerConfigMap0) ->
+    RootBin =
+        case Kind of
+            action -> <<"actions">>;
+            source -> <<"sources">>
+        end,
     TypeBin = emqx_utils_conv:bin(Type),
-    RawConf = #{<<"actions">> => #{TypeBin => #{Name => InnerConfigMap0}}},
+    RawConf = #{RootBin => #{TypeBin => #{Name => InnerConfigMap0}}},
-    #{<<"actions">> := #{TypeBin := #{Name := InnerConfigMap}}} = hocon_tconf:check_plain(
+    #{RootBin := #{TypeBin := #{Name := InnerConfigMap}}} = hocon_tconf:check_plain(
         emqx_bridge_v2_schema, RawConf, #{required => false, atom_key => false}
     ),
     InnerConfigMap.
 
+parse_and_check_connector(Type, Name, InnerConfigMap0) ->
+    TypeBin = emqx_utils_conv:bin(Type),
+    RawConf = #{<<"connectors">> => #{TypeBin => #{Name => InnerConfigMap0}}},
+    #{<<"connectors">> := #{TypeBin := #{Name := InnerConfigMap}}} = hocon_tconf:check_plain(
+        emqx_connector_schema, RawConf, #{required => false, atom_key => false}
+    ),
+    InnerConfigMap.
+
 bridge_id(Config) ->
     BridgeType = ?config(bridge_type, Config),
     BridgeName = ?config(bridge_name, Config),
@@ -134,10 +151,30 @@ bridge_id(Config) ->
     ConnectorId = emqx_bridge_resource:resource_id(BridgeType, BridgeName),
     <<"action:", BridgeId/binary, ":", ConnectorId/binary>>.
 
+source_hookpoint(Config) ->
+    #{kind := source, type := Type, name := Name} = get_common_values(Config),
+    BridgeId = emqx_bridge_resource:bridge_id(Type, Name),
+    emqx_bridge_v2:source_hookpoint(BridgeId).
+
+add_source_hookpoint(Config) ->
+    Hookpoint = source_hookpoint(Config),
+    ok = emqx_hooks:add(Hookpoint, {?MODULE, source_hookpoint_callback, [self()]}, 1000),
+    on_exit(fun() -> emqx_hooks:del(Hookpoint, {?MODULE, source_hookpoint_callback}) end),
+    ok.
+
 resource_id(Config) ->
-    BridgeType = ?config(bridge_type, Config),
+    #{
-    BridgeName = ?config(bridge_name, Config),
+        kind := Kind,
-    emqx_bridge_resource:resource_id(BridgeType, BridgeName).
+        type := Type,
+        name := Name,
+        connector_name := ConnectorName
+    } = get_common_values(Config),
+    case Kind of
+        source ->
+            emqx_bridge_v2:source_id(Type, Name, ConnectorName);
+        action ->
+            emqx_bridge_resource:resource_id(Type, Name)
+    end.
 
 create_bridge(Config) ->
     create_bridge(Config, _Overrides = #{}).
@@ -502,6 +539,54 @@ bridges_api_spec_schemas() ->
 actions_api_spec_schemas() ->
     api_spec_schemas("actions").
 
+get_value(Key, Config) ->
+    case proplists:get_value(Key, Config, undefined) of
+        undefined ->
+            error({missing_required_config, Key, Config});
+        Value ->
+            Value
+    end.
+
+get_common_values(Config) ->
+    Kind = proplists:get_value(bridge_kind, Config, action),
+    case Kind of
+        action ->
+            #{
+                conf_root_key => actions,
+                kind => Kind,
+                type => get_ct_config_with_fallback(Config, [action_type, bridge_type]),
+                name => get_ct_config_with_fallback(Config, [action_name, bridge_name]),
+                connector_type => get_value(connector_type, Config),
+                connector_name => get_value(connector_name, Config)
+            };
+        source ->
+            #{
+                conf_root_key => sources,
+                kind => Kind,
+                type => get_value(source_type, Config),
+                name => get_value(source_name, Config),
+                connector_type => get_value(connector_type, Config),
+                connector_name => get_value(connector_name, Config)
+            }
+    end.
+
+connector_resource_id(Config) ->
+    #{connector_type := Type, connector_name := Name} = get_common_values(Config),
+    emqx_connector_resource:resource_id(Type, Name).
+
+health_check_channel(Config) ->
+    ConnectorResId = connector_resource_id(Config),
+    ChannelResId = resource_id(Config),
+    emqx_resource_manager:channel_health_check(ConnectorResId, ChannelResId).
+
+%%------------------------------------------------------------------------------
+%% Internal export
+%%------------------------------------------------------------------------------
+
+source_hookpoint_callback(Message, TestPid) ->
+    TestPid ! {consumed_message, Message},
+    ok.
+
 %%------------------------------------------------------------------------------
 %% Testcases
 %%------------------------------------------------------------------------------
@@ -570,6 +655,55 @@ t_async_query(Config, MakeMessageFun, IsSuccessCheck, TracePoint) ->
     end,
     ok.
 
+%% - `ProduceFn': produces a message in the remote system that shall be consumed.
+%% - `Tracepoint': marks the end of consumed message processing.
+t_consume(Config, Opts) ->
+    #{
+        consumer_ready_tracepoint := ConsumerReadyTPFn,
+        produce_fn := ProduceFn,
+        check_fn := CheckFn,
+        produce_tracepoint := TracePointFn
+    } = Opts,
+    ?check_trace(
+        begin
+            ?assertMatch(
+                {{ok, _}, {ok, _}},
+                snabbkaffe:wait_async_action(
+                    fun() -> create_bridge_api(Config) end,
+                    ConsumerReadyTPFn,
+                    15_000
+                )
+            ),
+            ok = add_source_hookpoint(Config),
+            ResourceId = resource_id(Config),
+            ?retry(
+                _Sleep = 200,
+                _Attempts = 20,
+                ?assertMatch(
+                    #{status := ?status_connected},
+                    health_check_channel(Config)
+                )
+            ),
+            ?assertMatch(
+                {_, {ok, _}},
+                snabbkaffe:wait_async_action(
+                    ProduceFn,
+                    TracePointFn,
+                    15_000
+                )
+            ),
+            receive
+                {consumed_message, Message} ->
+                    CheckFn(Message)
+            after 5_000 ->
+                error({timeout, process_info(self(), messages)})
+            end,
+            ok
+        end,
+        []
+    ),
+    ok.
+
 t_create_via_http(Config) ->
     ?check_trace(
         begin
@@ -604,13 +738,15 @@ t_start_stop(Config, StopTracePoint) ->
 
     ?check_trace(
         begin
-            ProbeRes0 = probe_bridge_api(
+            ?assertMatch(
-                Kind,
+                {ok, {{_, 204, _}, _Headers, _Body}},
-                Type,
+                probe_bridge_api(
-                Name,
+                    Kind,
-                BridgeConfig
+                    Type,
+                    Name,
+                    BridgeConfig
+                )
             ),
-            ?assertMatch({ok, {{_, 204, _}, _Headers, _Body}}, ProbeRes0),
             %% Check that the bridge probe API doesn't leak atoms.
             AtomsBefore = erlang:system_info(atom_count),
             %% Probe again; shouldn't have created more atoms.

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub.app.src

@@ -8,7 +8,12 @@
         emqx_resource,
         ehttpc
     ]},
-    {env, [{emqx_action_info_modules, [emqx_bridge_gcp_pubsub_producer_action_info]}]},
+    {env, [
+        {emqx_action_info_modules, [
+            emqx_bridge_gcp_pubsub_producer_action_info,
+            emqx_bridge_gcp_pubsub_consumer_action_info
+        ]}
+    ]},
     {modules, []},
     {links, []}
 ]}.

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_client.erl

@@ -180,17 +180,17 @@ query_async(
     ),
     do_send_requests_async(State, {prepared_request, PreparedRequest, ReqOpts}, ReplyFunAndArgs).
 
--spec get_status(state()) -> connected | disconnected.
+-spec get_status(state()) -> ?status_connected | ?status_disconnected.
 get_status(#{connect_timeout := Timeout, pool_name := PoolName} = State) ->
     case do_get_status(PoolName, Timeout) of
         true ->
-            connected;
+            ?status_connected;
         false ->
             ?SLOG(error, #{
                 msg => "gcp_pubsub_bridge_get_status_failed",
                 state => State
             }),
-            disconnected
+            ?status_disconnected
     end.
 
 %%-------------------------------------------------------------------------------------------------

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_consumer_action_info.erl

@@ -0,0 +1,96 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_bridge_gcp_pubsub_consumer_action_info).
+
+-behaviour(emqx_action_info).
+
+-export([
+    is_source/0,
+    is_action/0,
+    bridge_v1_type_name/0,
+    action_type_name/0,
+    connector_type_name/0,
+    schema_module/0,
+    bridge_v1_config_to_action_config/2,
+    connector_action_config_to_bridge_v1_config/2
+]).
+
+is_source() -> true.
+
+is_action() -> false.
+
+bridge_v1_type_name() -> gcp_pubsub_consumer.
+
+action_type_name() -> gcp_pubsub_consumer.
+
+connector_type_name() -> gcp_pubsub_consumer.
+
+schema_module() -> emqx_bridge_gcp_pubsub_consumer_schema.
+
+bridge_v1_config_to_action_config(BridgeV1Config, ConnectorName) ->
+    CommonSourceKeys = emqx_bridge_v2_schema:top_level_common_source_keys(),
+    ParamsKeys = source_action_parameters_field_keys(),
+    Config1 = maps:with(CommonSourceKeys, BridgeV1Config),
+    ConsumerCfg = maps:get(<<"consumer">>, BridgeV1Config, #{}),
+    Params0 = maps:with(ParamsKeys, ConsumerCfg),
+    Params = maybe_set_pubsub_topic(Params0),
+    {source, gcp_pubsub_consumer,
+        emqx_utils_maps:update_if_present(
+            <<"resource_opts">>,
+            fun(Cfg) ->
+                emqx_bridge_v2_schema:project_to_sources_resource_opts(
+                    Cfg,
+                    resource_opts_fields()
+                )
+            end,
+            Config1#{
+                <<"connector">> => ConnectorName,
+                <<"parameters">> => Params
+            }
+        )}.
+
+connector_action_config_to_bridge_v1_config(ConnectorConfig, SourceConfig) ->
+    BridgeV1Config1 = maps:remove(<<"connector">>, SourceConfig),
+    BridgeV1Config2 = emqx_utils_maps:deep_merge(ConnectorConfig, BridgeV1Config1),
+    BridgeV1Config3 =
+        emqx_utils_maps:update_if_present(
+            <<"resource_opts">>,
+            fun(RO) -> maps:with(bridge_v1_resource_opts_fields(), RO) end,
+            BridgeV1Config2
+        ),
+    BridgeV1Config4 = emqx_utils_maps:deep_remove([<<"parameters">>, <<"topic">>], BridgeV1Config3),
+    emqx_utils_maps:rename(<<"parameters">>, <<"consumer">>, BridgeV1Config4).
+
+%%------------------------------------------------------------------------------------------
+%% Internal helper fns
+%%------------------------------------------------------------------------------------------
+
+%% The new schema has a single pubsub topic, so we take it from topic mapping when
+%% converting from v1.
+maybe_set_pubsub_topic(#{<<"topic_mapping">> := [#{<<"pubsub_topic">> := Topic} | _]} = Params) ->
+    Params#{<<"topic">> => Topic};
+maybe_set_pubsub_topic(Params) ->
+    Params.
+
+resource_opts_fields() ->
+    [
+        to_bin(K)
+     || {K, _} <- emqx_bridge_gcp_pubsub_consumer_schema:fields(source_resource_opts)
+    ].
+
+bridge_v1_resource_opts_fields() ->
+    [
+        to_bin(K)
+     || {K, _} <- emqx_bridge_gcp_pubsub:fields("consumer_resource_opts")
+    ].
+
+source_action_parameters_field_keys() ->
+    [
+        to_bin(K)
+     || {K, _} <- emqx_bridge_gcp_pubsub_consumer_schema:fields(source_parameters)
+    ].
+
+to_bin(L) when is_list(L) -> list_to_binary(L);
+to_bin(A) when is_atom(A) -> atom_to_binary(A, utf8).

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_consumer_schema.erl

@@ -0,0 +1,279 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_bridge_gcp_pubsub_consumer_schema).
+
+-import(hoconsc, [mk/2, ref/2]).
+
+-include_lib("typerefl/include/types.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+
+%% `hocon_schema' API
+-export([
+    namespace/0,
+    roots/0,
+    fields/1,
+    desc/1
+]).
+
+%% `emqx_bridge_v2_schema' "unofficial" API
+-export([
+    source_examples/1,
+    conn_bridge_examples/1,
+    connector_examples/1
+]).
+
+-define(CONNECTOR_TYPE, gcp_pubsub_consumer).
+-define(SOURCE_TYPE, gcp_pubsub_consumer).
+
+%%-------------------------------------------------------------------------------------------------
+%% `hocon_schema' API
+%%-------------------------------------------------------------------------------------------------
+
+namespace() ->
+    "gcp_pubsub_consumer".
+
+roots() ->
+    [].
+
+%%=========================================
+%% Action fields
+%%=========================================
+fields(source) ->
+    {gcp_pubsub_consumer,
+        mk(
+            hoconsc:map(name, ref(?MODULE, consumer_source)),
+            #{
+                desc => <<"GCP PubSub Consumer Source Config">>,
+                required => false
+            }
+        )};
+fields(consumer_source) ->
+    emqx_bridge_v2_schema:make_consumer_action_schema(
+        mk(
+            ref(?MODULE, source_parameters),
+            #{
+                required => true,
+                desc => ?DESC(consumer_source)
+            }
+        ),
+        #{resource_opts_ref => ref(?MODULE, source_resource_opts)}
+    );
+fields(source_parameters) ->
+    Fields0 = emqx_bridge_gcp_pubsub:fields(consumer),
+    Fields = lists:map(
+        fun
+            ({topic_mapping = Name, Sc}) ->
+                %% to please dialyzer...
+                Override = #{
+                    type => hocon_schema:field_schema(Sc, type),
+                    required => false,
+                    default => [],
+                    validator => fun(_) -> ok end,
+                    importance => ?IMPORTANCE_HIDDEN
+                },
+                {Name, hocon_schema:override(Sc, Override)};
+            (FieldSchema) ->
+                FieldSchema
+        end,
+        Fields0
+    ),
+    [
+        {topic,
+            mk(
+                binary(),
+                #{
+                    required => true,
+                    desc => ?DESC(emqx_bridge_gcp_pubsub, "pubsub_topic")
+                }
+            )}
+        | Fields
+    ];
+fields(source_resource_opts) ->
+    Fields = [
+        health_check_interval,
+        %% the workers pull the messages
+        request_ttl,
+        resume_interval
+    ],
+    lists:filter(
+        fun({Key, _Sc}) -> lists:member(Key, Fields) end,
+        emqx_resource_schema:create_opts(
+            _Overrides = [
+                {health_check_interval, #{default => <<"30s">>}}
+            ]
+        )
+    );
+%%=========================================
+%% Connector fields
+%%=========================================
+fields("config_connector") ->
+    %% FIXME
+    emqx_connector_schema:common_fields() ++
+        connector_config_fields();
+fields(connector_resource_opts) ->
+    emqx_connector_schema:resource_opts_fields();
+%%=========================================
+%% HTTP API fields: source
+%%=========================================
+fields(Field) when
+    Field == "get_source";
+    Field == "post_source";
+    Field == "put_source"
+->
+    emqx_bridge_v2_schema:api_fields(Field, ?SOURCE_TYPE, fields(consumer_source));
+%%=========================================
+%% HTTP API fields: connector
+%%=========================================
+fields(Field) when
+    Field == "get_connector";
+    Field == "put_connector";
+    Field == "post_connector"
+->
+    emqx_connector_schema:api_fields(Field, ?CONNECTOR_TYPE, connector_config_fields()).
+
+connector_config_fields() ->
+    emqx_bridge_gcp_pubsub:fields(connector_config) ++
+        emqx_connector_schema:resource_opts_ref(?MODULE, connector_resource_opts).
+
+desc("config_connector") ->
+    ?DESC("config_connector");
+desc(source_parameters) ->
+    ?DESC(source_parameters);
+desc(consumer_source) ->
+    ?DESC(consumer_source);
+desc(connector_resource_opts) ->
+    ?DESC(emqx_resource_schema, "resource_opts");
+desc(source_resource_opts) ->
+    ?DESC(emqx_resource_schema, "resource_opts");
+desc(_Name) ->
+    undefined.
+
+%%-------------------------------------------------------------------------------------------------
+%% `emqx_bridge_v2_schema' "unofficial" API
+%%-------------------------------------------------------------------------------------------------
+
+source_examples(Method) ->
+    [
+        #{
+            <<"gcp_pubsub_consumer">> => #{
+                summary => <<"GCP PubSub Consumer Source">>,
+                value => source_example(Method)
+            }
+        }
+    ].
+
+connector_examples(Method) ->
+    [
+        #{
+            <<"gcp_pubsub_consumer">> => #{
+                summary => <<"GCP PubSub Consumer Connector">>,
+                value => connector_example(Method)
+            }
+        }
+    ].
+
+conn_bridge_examples(Method) ->
+    emqx_bridge_gcp_pubsub:conn_bridge_examples(Method).
+
+source_example(post) ->
+    maps:merge(
+        source_example(put),
+        #{
+            type => <<"gcp_pubsub_consumer">>,
+            name => <<"my_action">>
+        }
+    );
+source_example(get) ->
+    maps:merge(
+        source_example(put),
+        #{
+            status => <<"connected">>,
+            node_status => [
+                #{
+                    node => <<"emqx@localhost">>,
+                    status => <<"connected">>
+                }
+            ]
+        }
+    );
+source_example(put) ->
+    #{
+        enable => true,
+        description => <<"my source">>,
+        connector => <<"my_connector">>,
+        parameters =>
+            #{
+                topic => <<"my-topic">>,
+                pull_max_messages => 100
+            },
+        resource_opts =>
+            #{
+                request_ttl => <<"45s">>,
+                health_check_interval => <<"30s">>
+            }
+    }.
+
+connector_example(get) ->
+    maps:merge(
+        connector_example(post),
+        #{
+            status => <<"connected">>,
+            node_status => [
+                #{
+                    node => <<"emqx@localhost">>,
+                    status => <<"connected">>
+                }
+            ],
+            actions => [<<"my_action">>]
+        }
+    );
+connector_example(post) ->
+    maps:merge(
+        connector_example(put),
+        #{
+            type => <<"gcp_pubsub_producer">>,
+            name => <<"my_connector">>
+        }
+    );
+connector_example(put) ->
+    #{
+        enable => true,
+        description => <<"my connector">>,
+        connect_timeout => <<"15s">>,
+        pool_size => 8,
+        resource_opts =>
+            #{
+                start_after_created => true,
+                health_check_interval => <<"30s">>,
+                start_timeout => <<"5s">>
+            },
+        max_retries => 2,
+        service_account_json =>
+            #{
+                auth_provider_x509_cert_url =>
+                    <<"https://www.googleapis.com/oauth2/v1/certs">>,
+                auth_uri =>
+                    <<"https://accounts.google.com/o/oauth2/auth">>,
+                client_email =>
+                    <<"test@myproject.iam.gserviceaccount.com">>,
+                client_id => <<"123812831923812319190">>,
+                client_x509_cert_url =>
+                    <<
+                        "https://www.googleapis.com/robot/v1/"
+                        "metadata/x509/test%40myproject.iam.gserviceaccount.com"
+                    >>,
+                private_key =>
+                    <<
+                        "-----BEGIN PRIVATE KEY-----\n"
+                        "MIIEvQI..."
+                    >>,
+                private_key_id => <<"kid">>,
+                project_id => <<"myproject">>,
+                token_uri =>
+                    <<"https://oauth2.googleapis.com/token">>,
+                type => <<"service_account">>
+            },
+        pipelining => 100
+    }.

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_consumer_worker.erl

@@ -37,8 +37,9 @@
     client := emqx_bridge_gcp_pubsub_client:state(),
     ecpool_worker_id => non_neg_integer(),
     forget_interval := duration(),
-    hookpoint := binary(),
+    hookpoints := [binary()],
-    instance_id := binary(),
+    connector_resource_id := binary(),
+    source_resource_id := binary(),
     mqtt_config => emqx_bridge_gcp_pubsub_impl_consumer:mqtt_config(),
     project_id := emqx_bridge_gcp_pubsub_client:project_id(),
     pull_max_messages := non_neg_integer(),
@@ -55,9 +56,10 @@
     client := emqx_bridge_gcp_pubsub_client:state(),
     ecpool_worker_id := non_neg_integer(),
     forget_interval := duration(),
-    hookpoint := binary(),
+    hookpoints := [binary()],
-    instance_id := binary(),
+    connector_resource_id := binary(),
-    mqtt_config := emqx_bridge_gcp_pubsub_impl_consumer:mqtt_config(),
+    source_resource_id := binary(),
+    mqtt_config := #{} | emqx_bridge_gcp_pubsub_impl_consumer:mqtt_config(),
     pending_acks := #{message_id() => ack_id()},
     project_id := emqx_bridge_gcp_pubsub_client:project_id(),
     pull_max_messages := non_neg_integer(),
@@ -97,7 +99,7 @@ ensure_subscription(WorkerPid) ->
     gen_server:cast(WorkerPid, ensure_subscription).
 
 -spec reply_delegator(pid(), pull_async, binary(), {ok, map()} | {error, timeout | term()}) -> ok.
-reply_delegator(WorkerPid, pull_async = _Action, InstanceId, Result) ->
+reply_delegator(WorkerPid, pull_async = _Action, SourceResId, Result) ->
     ?tp(gcp_pubsub_consumer_worker_reply_delegator, #{result => Result}),
     case Result of
         {error, timeout} ->
@@ -107,7 +109,7 @@ reply_delegator(WorkerPid, pull_async = _Action, InstanceId, Result) ->
                 warning,
                 "gcp_pubsub_consumer_worker_pull_error",
                 #{
-                    instance_id => InstanceId,
+                    instance_id => SourceResId,
                     reason => Reason
                 }
             ),
@@ -144,7 +146,7 @@ health_check(WorkerPid) ->
     end.
 
 %%-------------------------------------------------------------------------------------------------
-%% `emqx_resource' API
+%% `ecpool' API
 %%-------------------------------------------------------------------------------------------------
 
 connect(Opts0) ->
@@ -156,8 +158,9 @@ connect(Opts0) ->
         client := Client,
         ecpool_worker_id := WorkerId,
         forget_interval := ForgetInterval,
-        hookpoint := Hookpoint,
+        hookpoints := Hookpoints,
-        instance_id := InstanceId,
+        connector_resource_id := ConnectorResId,
+        source_resource_id := SourceResId,
         project_id := ProjectId,
         pull_max_messages := PullMaxMessages,
         pull_retry_interval := PullRetryInterval,
@@ -175,8 +178,9 @@ connect(Opts0) ->
         %% workers.
         client => Client,
         forget_interval => ForgetInterval,
-        hookpoint => Hookpoint,
+        hookpoints => Hookpoints,
-        instance_id => InstanceId,
+        connector_resource_id => ConnectorResId,
+        source_resource_id => SourceResId,
         mqtt_config => MQTTConfig,
         project_id => ProjectId,
         pull_max_messages => PullMaxMessages,
@@ -185,6 +189,7 @@ connect(Opts0) ->
         topic => Topic,
         subscription_id => subscription_id(BridgeName, Topic)
     },
+    ?tp(gcp_pubsub_consumer_worker_about_to_spawn, #{}),
     start_link(Config).
 
 %%-------------------------------------------------------------------------------------------------
@@ -209,13 +214,13 @@ handle_continue(?ensure_subscription, State0) ->
         already_exists ->
             {noreply, State0, {continue, ?patch_subscription}};
         continue ->
-            #{instance_id := InstanceId} = State0,
+            #{source_resource_id := SourceResId} = State0,
             ?MODULE:pull_async(self()),
             optvar:set(?OPTVAR_SUB_OK(self()), subscription_ok),
             ?tp(
                 debug,
                 "gcp_pubsub_consumer_worker_subscription_ready",
-                #{instance_id => InstanceId}
+                #{instance_id => SourceResId}
             ),
             {noreply, State0};
         retry ->
@@ -232,13 +237,13 @@ handle_continue(?patch_subscription, State0) ->
     ?tp(gcp_pubsub_consumer_worker_patch_subscription_enter, #{}),
     case patch_subscription(State0) of
         ok ->
-            #{instance_id := InstanceId} = State0,
+            #{source_resource_id := SourceResId} = State0,
             ?MODULE:pull_async(self()),
             optvar:set(?OPTVAR_SUB_OK(self()), subscription_ok),
             ?tp(
                 debug,
                 "gcp_pubsub_consumer_worker_subscription_ready",
-                #{instance_id => InstanceId}
+                #{instance_id => SourceResId}
             ),
             {noreply, State0};
         error ->
@@ -292,13 +297,13 @@ handle_info({forget_message_ids, MsgIds}, State0) ->
     {noreply, State};
 handle_info(Msg, State0) ->
     #{
-        instance_id := InstanceId,
+        source_resource_id := SoureceResId,
         topic := Topic
     } = State0,
     ?SLOG(debug, #{
         msg => "gcp_pubsub_consumer_worker_unexpected_message",
         unexpected_msg => Msg,
-        instance_id => InstanceId,
+        instance_id => SoureceResId,
         topic => Topic
     }),
     {noreply, State0}.
@@ -309,11 +314,11 @@ terminate({error, Reason}, State) when
     Reason =:= permission_denied
 ->
     #{
-        instance_id := InstanceId,
+        source_resource_id := SourceResId,
         topic := _Topic
     } = State,
     optvar:unset(?OPTVAR_SUB_OK(self())),
-    emqx_bridge_gcp_pubsub_impl_consumer:mark_as_unhealthy(InstanceId, Reason),
+    emqx_bridge_gcp_pubsub_impl_consumer:mark_as_unhealthy(SourceResId, Reason),
     ?tp(gcp_pubsub_consumer_worker_terminate, #{reason => {error, Reason}, topic => _Topic}),
     ok;
 terminate(_Reason, _State) ->
@@ -351,7 +356,7 @@ ensure_subscription_exists(State) ->
     ?tp(gcp_pubsub_consumer_worker_create_subscription_enter, #{}),
     #{
         client := Client,
-        instance_id := InstanceId,
+        source_resource_id := SourceResId,
         request_ttl := RequestTTL,
         subscription_id := SubscriptionId,
         topic := Topic
@@ -369,7 +374,7 @@ ensure_subscription_exists(State) ->
                 debug,
                 "gcp_pubsub_consumer_worker_subscription_already_exists",
                 #{
-                    instance_id => InstanceId,
+                    instance_id => SourceResId,
                     topic => Topic,
                     subscription_id => SubscriptionId
                 }
@@ -381,7 +386,7 @@ ensure_subscription_exists(State) ->
                 warning,
                 "gcp_pubsub_consumer_worker_nonexistent_topic",
                 #{
-                    instance_id => InstanceId,
+                    instance_id => SourceResId,
                     topic => Topic
                 }
             ),
@@ -392,7 +397,7 @@ ensure_subscription_exists(State) ->
                 warning,
                 "gcp_pubsub_consumer_worker_permission_denied",
                 #{
-                    instance_id => InstanceId,
+                    instance_id => SourceResId,
                     topic => Topic
                 }
             ),
@@ -403,7 +408,7 @@ ensure_subscription_exists(State) ->
                 warning,
                 "gcp_pubsub_consumer_worker_bad_credentials",
                 #{
-                    instance_id => InstanceId,
+                    instance_id => SourceResId,
                     topic => Topic
                 }
             ),
@@ -413,7 +418,7 @@ ensure_subscription_exists(State) ->
                 debug,
                 "gcp_pubsub_consumer_worker_subscription_created",
                 #{
-                    instance_id => InstanceId,
+                    instance_id => SourceResId,
                     topic => Topic,
                     subscription_id => SubscriptionId
                 }
@@ -424,7 +429,7 @@ ensure_subscription_exists(State) ->
                 error,
                 "gcp_pubsub_consumer_worker_subscription_error",
                 #{
-                    instance_id => InstanceId,
+                    instance_id => SourceResId,
                     topic => Topic,
                     reason => Reason
                 }
@@ -436,7 +441,7 @@ ensure_subscription_exists(State) ->
 patch_subscription(State) ->
     #{
         client := Client,
-        instance_id := InstanceId,
+        source_resource_id := SourceResId,
         subscription_id := SubscriptionId,
         request_ttl := RequestTTL,
         topic := Topic
@@ -453,7 +458,7 @@ patch_subscription(State) ->
                 debug,
                 "gcp_pubsub_consumer_worker_subscription_patched",
                 #{
-                    instance_id => InstanceId,
+                    instance_id => SourceResId,
                     topic => Topic,
                     subscription_id => SubscriptionId,
                     result => Res
@@ -465,7 +470,7 @@ patch_subscription(State) ->
                 warning,
                 "gcp_pubsub_consumer_worker_subscription_patch_error",
                 #{
-                    instance_id => InstanceId,
+                    instance_id => SourceResId,
                     topic => Topic,
                     subscription_id => SubscriptionId,
                     reason => Reason
@@ -483,7 +488,7 @@ do_pull_async(State0) ->
         begin
             #{
                 client := Client,
-                instance_id := InstanceId,
+                source_resource_id := SourceResId,
                 request_ttl := RequestTTL
             } = State0,
             Method = post,
@@ -491,7 +496,7 @@ do_pull_async(State0) ->
             Body = body(State0, pull),
             ReqOpts = #{request_ttl => RequestTTL},
             PreparedRequest = {prepared_request, {Method, Path, Body}, ReqOpts},
-            ReplyFunAndArgs = {fun ?MODULE:reply_delegator/4, [self(), pull_async, InstanceId]},
+            ReplyFunAndArgs = {fun ?MODULE:reply_delegator/4, [self(), pull_async, SourceResId]},
             Res = emqx_bridge_gcp_pubsub_client:query_async(
                 PreparedRequest,
                 ReplyFunAndArgs,
@@ -734,13 +739,9 @@ handle_message(State, #{<<"ackId">> := AckId, <<"message">> := InnerMsg} = _Mess
         #{message_id => maps:get(<<"messageId">>, InnerMsg), message => _Message, ack_id => AckId},
         begin
             #{
-                instance_id := InstanceId,
+                source_resource_id := SourceResId,
-                hookpoint := Hookpoint,
+                hookpoints := Hookpoints,
-                mqtt_config := #{
+                mqtt_config := MQTTConfig,
-                    payload_template := PayloadTemplate,
-                    qos := MQTTQoS,
-                    mqtt_topic := MQTTTopic
-                },
                 topic := Topic
             } = State,
             #{
@@ -764,15 +765,32 @@ handle_message(State, #{<<"ackId">> := AckId, <<"message">> := InnerMsg} = _Mess
                         {<<"orderingKey">>, ordering_key}
                     ]
                 ),
-            Payload = render(FullMessage, PayloadTemplate),
+            legacy_maybe_publish_mqtt_message(MQTTConfig, SourceResId, FullMessage),
-            MQTTMessage = emqx_message:make(InstanceId, MQTTQoS, MQTTTopic, Payload),
+            lists:foreach(
-            _ = emqx:publish(MQTTMessage),
+                fun(Hookpoint) -> emqx_hooks:run(Hookpoint, [FullMessage]) end,
-            emqx_hooks:run(Hookpoint, [FullMessage]),
+                Hookpoints
-            emqx_resource_metrics:received_inc(InstanceId),
+            ),
+            emqx_resource_metrics:received_inc(SourceResId),
             ok
         end
     ).
 
+legacy_maybe_publish_mqtt_message(
+    _MQTTConfig = #{
+        payload_template := PayloadTemplate,
+        qos := MQTTQoS,
+        mqtt_topic := MQTTTopic
+    },
+    SourceResId,
+    FullMessage
+) ->
+    Payload = render(FullMessage, PayloadTemplate),
+    MQTTMessage = emqx_message:make(SourceResId, MQTTQoS, MQTTTopic, Payload),
+    _ = emqx:publish(MQTTMessage),
+    ok;
+legacy_maybe_publish_mqtt_message(_MQTTConfig, _SourceResId, _FullMessage) ->
+    ok.
+
 -spec add_if_present(any(), map(), any(), map()) -> map().
 add_if_present(FromKey, Message, ToKey, Map) ->
     case maps:get(FromKey, Message, undefined) of

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_impl_consumer.erl

@@ -12,7 +12,12 @@
     query_mode/1,
     on_start/2,
     on_stop/2,
-    on_get_status/2
+    on_get_status/2,
+
+    on_add_channel/4,
+    on_remove_channel/3,
+    on_get_channels/1,
+    on_get_channel_status/3
 ]).
 
 %% health check API
@@ -31,7 +36,7 @@
     qos := emqx_types:qos(),
     payload_template := string()
 }.
--type config() :: #{
+-type connector_config() :: #{
     connect_timeout := emqx_schema:duration_ms(),
     max_retries := non_neg_integer(),
     pool_size := non_neg_integer(),
@@ -39,9 +44,28 @@
     service_account_json := emqx_bridge_gcp_pubsub_client:service_account_json(),
     any() => term()
 }.
--type state() :: #{
+-type connector_state() :: #{
-    client := emqx_bridge_gcp_pubsub_client:state()
+    client := emqx_bridge_gcp_pubsub_client:state(),
+    installed_sources := #{source_resource_id() => source_state()},
+    project_id := binary()
 }.
+-type topic_mapping() :: #{
+    pubsub_topic := binary(),
+    mqtt_topic := binary(),
+    qos := emqx_types:qos(),
+    payload_template := binary()
+}.
+-type source_config() :: #{
+    bridge_name := binary(),
+    hookpoints := [binary()],
+    parameters := #{
+        consumer_workers_per_topic := pos_integer(),
+        topic_mapping := [topic_mapping(), ...]
+    },
+    resource_opts := #{request_ttl := infinity | emqx_schema:duration_ms(), any() => term()},
+    topic := binary()
+}.
+-type source_state() :: #{}.
 
 -export_type([mqtt_config/0]).
 
@@ -67,68 +91,140 @@ callback_mode() -> async_if_possible.
 -spec query_mode(any()) -> query_mode().
 query_mode(_Config) -> no_queries.
 
--spec on_start(resource_id(), config()) -> {ok, state()} | {error, term()}.
+-spec on_start(connector_resource_id(), connector_config()) ->
-on_start(InstanceId, Config0) ->
+    {ok, connector_state()} | {error, term()}.
+on_start(ConnectorResId, Config0) ->
     %% ensure it's a binary key map
     Config = maps:update_with(service_account_json, fun emqx_utils_maps:binary_key_map/1, Config0),
-    case emqx_bridge_gcp_pubsub_client:start(InstanceId, Config) of
+    #{service_account_json := #{<<"project_id">> := ProjectId}} = Config,
+    case emqx_bridge_gcp_pubsub_client:start(ConnectorResId, Config) of
         {ok, Client} ->
-            start_consumers(InstanceId, Client, Config);
+            ConnectorState = #{
+                client => Client,
+                installed_sources => #{},
+                project_id => ProjectId
+            },
+            {ok, ConnectorState};
         Error ->
             Error
     end.
 
--spec on_stop(resource_id(), state()) -> ok | {error, term()}.
+-spec on_stop(resource_id(), connector_state()) -> ok | {error, term()}.
-on_stop(InstanceId, _State) ->
+on_stop(ConnectorResId, ConnectorState) ->
     ?tp(gcp_pubsub_consumer_stop_enter, #{}),
-    clear_unhealthy(InstanceId),
+    clear_unhealthy(ConnectorState),
-    ok = stop_consumers(InstanceId),
+    ok = stop_consumers(ConnectorState),
-    emqx_bridge_gcp_pubsub_client:stop(InstanceId).
+    emqx_bridge_gcp_pubsub_client:stop(ConnectorResId).
 
--spec on_get_status(resource_id(), state()) -> connected | connecting | {disconnected, state(), _}.
+-spec on_get_status(resource_id(), connector_state()) ->
-on_get_status(InstanceId, State) ->
+    ?status_connected | ?status_connecting.
+on_get_status(_ConnectorResId, ConnectorState) ->
+    #{client := Client} = ConnectorState,
+    get_client_status(Client).
+
+-spec on_add_channel(
+    connector_resource_id(),
+    connector_state(),
+    source_resource_id(),
+    source_config()
+) ->
+    {ok, connector_state()}.
+on_add_channel(ConnectorResId, ConnectorState0, SourceResId, SourceConfig) ->
+    #{
+        client := Client,
+        installed_sources := InstalledSources0,
+        project_id := ProjectId
+    } = ConnectorState0,
+    case start_consumers(ConnectorResId, SourceResId, Client, ProjectId, SourceConfig) of
+        {ok, SourceState} ->
+            InstalledSources = InstalledSources0#{SourceResId => SourceState},
+            ConnectorState = ConnectorState0#{installed_sources := InstalledSources},
+            {ok, ConnectorState};
+        Error = {error, _} ->
+            Error
+    end.
+
+-spec on_remove_channel(
+    connector_resource_id(),
+    connector_state(),
+    source_resource_id()
+) ->
+    {ok, connector_state()}.
+on_remove_channel(_ConnectorResId, ConnectorState0, SourceResId) ->
+    #{installed_sources := InstalledSources0} = ConnectorState0,
+    case maps:take(SourceResId, InstalledSources0) of
+        {SourceState, InstalledSources} ->
+            stop_consumers1(SourceState),
+            ok;
+        error ->
+            InstalledSources = InstalledSources0
+    end,
+    ConnectorState = ConnectorState0#{installed_sources := InstalledSources},
+    {ok, ConnectorState}.
+
+-spec on_get_channels(connector_resource_id()) ->
+    [{action_resource_id(), source_config()}].
+on_get_channels(ConnectorResId) ->
+    emqx_bridge_v2:get_channels_for_connector(ConnectorResId).
+
+-spec on_get_channel_status(connector_resource_id(), source_resource_id(), connector_state()) ->
+    health_check_status().
+on_get_channel_status(
+    _ConnectorResId,
+    SourceResId,
+    ConnectorState = #{installed_sources := InstalledSources}
+) when is_map_key(SourceResId, InstalledSources) ->
     %% We need to check this flag separately because the workers might be gone when we
     %% check them.
-    case check_if_unhealthy(InstanceId) of
+    case check_if_unhealthy(SourceResId) of
         {error, topic_not_found} ->
-            {disconnected, State, {unhealthy_target, ?TOPIC_MESSAGE}};
+            {?status_disconnected, {unhealthy_target, ?TOPIC_MESSAGE}};
         {error, permission_denied} ->
-            {disconnected, State, {unhealthy_target, ?PERMISSION_MESSAGE}};
+            {?status_disconnected, {unhealthy_target, ?PERMISSION_MESSAGE}};
         {error, bad_credentials} ->
-            {disconnected, State, {unhealthy_target, ?PERMISSION_MESSAGE}};
+            {?status_disconnected, {unhealthy_target, ?PERMISSION_MESSAGE}};
         ok ->
-            #{client := Client} = State,
+            #{client := Client} = ConnectorState,
-            check_workers(InstanceId, Client)
+            #{SourceResId := #{pool_name := PoolName}} = InstalledSources,
-    end.
+            check_workers(PoolName, Client)
+    end;
+on_get_channel_status(_ConnectorResId, _SourceResId, _ConnectorState) ->
+    ?status_disconnected.
 
 %%-------------------------------------------------------------------------------------------------
 %% Health check API (signalled by consumer worker)
 %%-------------------------------------------------------------------------------------------------
 
 -spec mark_as_unhealthy(
-    resource_id(),
+    source_resource_id(),
     topic_not_found
     | permission_denied
     | bad_credentials
 ) -> ok.
-mark_as_unhealthy(InstanceId, Reason) ->
+mark_as_unhealthy(SourceResId, Reason) ->
-    optvar:set(?OPTVAR_UNHEALTHY(InstanceId), Reason),
+    optvar:set(?OPTVAR_UNHEALTHY(SourceResId), Reason),
     ok.
 
--spec clear_unhealthy(resource_id()) -> ok.
+-spec clear_unhealthy(connector_state()) -> ok.
-clear_unhealthy(InstanceId) ->
+clear_unhealthy(ConnectorState) ->
-    optvar:unset(?OPTVAR_UNHEALTHY(InstanceId)),
+    #{installed_sources := InstalledSources} = ConnectorState,
+    maps:foreach(
+        fun(SourceResId, _SourceState) ->
+            optvar:unset(?OPTVAR_UNHEALTHY(SourceResId))
+        end,
+        InstalledSources
+    ),
     ?tp(gcp_pubsub_consumer_clear_unhealthy, #{}),
     ok.
 
--spec check_if_unhealthy(resource_id()) ->
+-spec check_if_unhealthy(source_resource_id()) ->
     ok
     | {error,
         topic_not_found
         | permission_denied
         | bad_credentials}.
-check_if_unhealthy(InstanceId) ->
+check_if_unhealthy(SourceResId) ->
-    case optvar:peek(?OPTVAR_UNHEALTHY(InstanceId)) of
+    case optvar:peek(?OPTVAR_UNHEALTHY(SourceResId)) of
         {ok, Reason} ->
             {error, Reason};
         undefined ->
@@ -139,15 +235,14 @@ check_if_unhealthy(InstanceId) ->
 %% Internal fns
 %%-------------------------------------------------------------------------------------------------
 
-start_consumers(InstanceId, Client, Config) ->
+start_consumers(ConnectorResId, SourceResId, Client, ProjectId, SourceConfig) ->
     #{
         bridge_name := BridgeName,
-        consumer := ConsumerConfig0,
+        parameters := ConsumerConfig0,
-        hookpoint := Hookpoint,
+        hookpoints := Hookpoints,
-        resource_opts := #{request_ttl := RequestTTL},
+        resource_opts := #{request_ttl := RequestTTL}
-        service_account_json := #{<<"project_id">> := ProjectId}
+    } = SourceConfig,
-    } = Config,
+    ConsumerConfig1 = ensure_topic_mapping(ConsumerConfig0),
-    ConsumerConfig1 = maps:update_with(topic_mapping, fun convert_topic_mapping/1, ConsumerConfig0),
     TopicMapping = maps:get(topic_mapping, ConsumerConfig1),
     ConsumerWorkersPerTopic = maps:get(consumer_workers_per_topic, ConsumerConfig1),
     PoolSize = map_size(TopicMapping) * ConsumerWorkersPerTopic,
@@ -156,8 +251,9 @@ start_consumers(InstanceId, Client, Config) ->
         bridge_name => BridgeName,
         client => Client,
         forget_interval => forget_interval(RequestTTL),
-        hookpoint => Hookpoint,
+        hookpoints => Hookpoints,
-        instance_id => InstanceId,
+        connector_resource_id => ConnectorResId,
+        source_resource_id => SourceResId,
         pool_size => PoolSize,
         project_id => ProjectId,
         pull_retry_interval => RequestTTL,
@@ -169,17 +265,14 @@ start_consumers(InstanceId, Client, Config) ->
         ok ->
             ok;
         {error, not_found} ->
-            _ = emqx_bridge_gcp_pubsub_client:stop(InstanceId),
             throw(
                 {unhealthy_target, ?TOPIC_MESSAGE}
             );
         {error, permission_denied} ->
-            _ = emqx_bridge_gcp_pubsub_client:stop(InstanceId),
             throw(
                 {unhealthy_target, ?PERMISSION_MESSAGE}
             );
         {error, bad_credentials} ->
-            _ = emqx_bridge_gcp_pubsub_client:stop(InstanceId),
             throw(
                 {unhealthy_target, ?PERMISSION_MESSAGE}
             );
@@ -190,31 +283,46 @@ start_consumers(InstanceId, Client, Config) ->
             ok
     end,
     case
-        emqx_resource_pool:start(InstanceId, emqx_bridge_gcp_pubsub_consumer_worker, ConsumerOpts)
+        emqx_resource_pool:start(SourceResId, emqx_bridge_gcp_pubsub_consumer_worker, ConsumerOpts)
     of
         ok ->
-            State = #{
+            State = #{pool_name => SourceResId},
-                client => Client,
-                pool_name => InstanceId
-            },
             {ok, State};
         {error, Reason} ->
-            _ = emqx_bridge_gcp_pubsub_client:stop(InstanceId),
             {error, Reason}
     end.
 
-stop_consumers(InstanceId) ->
+stop_consumers(ConnectorState) ->
+    #{installed_sources := InstalledSources} = ConnectorState,
+    maps:foreach(
+        fun(_SourceResId, SourceState) ->
+            stop_consumers1(SourceState)
+        end,
+        InstalledSources
+    ).
+
+stop_consumers1(SourceState) ->
+    #{pool_name := PoolName} = SourceState,
     _ = log_when_error(
         fun() ->
-            ok = emqx_resource_pool:stop(InstanceId)
+            ok = emqx_resource_pool:stop(PoolName)
         end,
         #{
             msg => "failed_to_stop_pull_worker_pool",
-            instance_id => InstanceId
+            pool_name => PoolName
         }
     ),
     ok.
 
+%% This is to ensure backwards compatibility with the deprectated topic mapping.
+ensure_topic_mapping(ConsumerConfig0 = #{topic_mapping := [_ | _]}) ->
+    %% There is an existing topic mapping: legacy config.  We use it and ignore the single
+    %% pubsub topic so that the bridge keeps working as before.
+    maps:update_with(topic_mapping, fun convert_topic_mapping/1, ConsumerConfig0);
+ensure_topic_mapping(ConsumerConfig0 = #{topic := PubsubTopic}) ->
+    %% No topic mapping: generate one without MQTT templates.
+    maps:put(topic_mapping, #{PubsubTopic => #{}}, ConsumerConfig0).
+
 convert_topic_mapping(TopicMappingList) ->
     lists:foldl(
         fun(Fields, Acc) ->
@@ -268,35 +376,37 @@ check_for_topic_existence(Topic, Client, ReqOpts) ->
             {error, Reason}
     end.
 
--spec get_client_status(emqx_bridge_gcp_pubsub_client:state()) -> connected | connecting.
+-spec get_client_status(emqx_bridge_gcp_pubsub_client:state()) ->
+    ?status_connected | ?status_connecting.
 get_client_status(Client) ->
     case emqx_bridge_gcp_pubsub_client:get_status(Client) of
-        disconnected -> connecting;
+        ?status_disconnected -> ?status_connecting;
-        connected -> connected
+        ?status_connected -> ?status_connected
     end.
 
--spec check_workers(resource_id(), emqx_bridge_gcp_pubsub_client:state()) -> connected | connecting.
+-spec check_workers(source_resource_id(), emqx_bridge_gcp_pubsub_client:state()) ->
-check_workers(InstanceId, Client) ->
+    ?status_connected | ?status_connecting.
+check_workers(SourceResId, Client) ->
     case
         emqx_resource_pool:health_check_workers(
-            InstanceId,
+            SourceResId,
             fun emqx_bridge_gcp_pubsub_consumer_worker:health_check/1,
             emqx_resource_pool:health_check_timeout(),
             #{return_values => true}
         )
     of
         {ok, []} ->
-            connecting;
+            ?status_connecting;
         {ok, Values} ->
             AllOk = lists:all(fun(S) -> S =:= subscription_ok end, Values),
             case AllOk of
                 true ->
                     get_client_status(Client);
                 false ->
-                    connecting
+                    ?status_connecting
             end;
         {error, _} ->
-            connecting
+            ?status_connecting
     end.
 
 log_when_error(Fun, Log) ->

apps/emqx_bridge_gcp_pubsub/test/emqx_bridge_gcp_pubsub_consumer_SUITE.erl

@@ -13,6 +13,8 @@
 
 -define(BRIDGE_TYPE, gcp_pubsub_consumer).
 -define(BRIDGE_TYPE_BIN, <<"gcp_pubsub_consumer">>).
+-define(CONNECTOR_TYPE_BIN, <<"gcp_pubsub_consumer">>).
+-define(SOURCE_TYPE_BIN, <<"gcp_pubsub_consumer">>).
 -define(REPUBLISH_TOPIC, <<"republish/t">>).
 -define(PREPARED_REQUEST(METHOD, PATH, BODY),
     {prepared_request, {METHOD, PATH, BODY}, #{request_ttl => 1_000}}
@@ -147,6 +149,7 @@ common_init_per_testcase(TestCase, Config0) ->
     ensure_topics(Config),
     ok = snabbkaffe:start_trace(),
     [
+        {bridge_kind, source},
         {bridge_type, ?BRIDGE_TYPE},
         {bridge_name, Name},
         {bridge_config, ConsumerConfig},
@@ -157,18 +160,13 @@ common_init_per_testcase(TestCase, Config0) ->
     ].
 
 end_per_testcase(_Testcase, Config) ->
-    case proplists:get_bool(skip_does_not_apply, Config) of
+    ProxyHost = ?config(proxy_host, Config),
-        true ->
+    ProxyPort = ?config(proxy_port, Config),
-            ok;
+    emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
-        false ->
+    emqx_bridge_v2_testlib:delete_all_bridges_and_connectors(),
-            ProxyHost = ?config(proxy_host, Config),
+    emqx_common_test_helpers:call_janitor(60_000),
-            ProxyPort = ?config(proxy_port, Config),
+    ok = snabbkaffe:stop(),
-            emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
+    ok.
-            emqx_bridge_testlib:delete_all_bridges(),
-            emqx_common_test_helpers:call_janitor(60_000),
-            ok = snabbkaffe:stop(),
-            ok
-    end.
 
 %%------------------------------------------------------------------------------
 %% Helper fns
@@ -421,9 +419,22 @@ start_and_subscribe_mqtt(Config) ->
     ok.
 
 resource_id(Config) ->
-    Type = ?BRIDGE_TYPE_BIN,
     Name = ?config(consumer_name, Config),
-    emqx_bridge_resource:resource_id(Type, Name).
+    emqx_bridge_v2:source_id(?SOURCE_TYPE_BIN, Name, Name).
+
+connector_resource_id(Config) ->
+    Name = ?config(consumer_name, Config),
+    emqx_connector_resource:resource_id(?CONNECTOR_TYPE_BIN, Name).
+
+health_check(Config) ->
+    #{status := Status} = health_check_channel(Config),
+    {ok, Status}.
+
+health_check_channel(Config) ->
+    Name = ?config(consumer_name, Config),
+    ConnectorResId = emqx_connector_resource:resource_id(?CONNECTOR_TYPE_BIN, Name),
+    SourceResId = resource_id(Config),
+    emqx_resource_manager:channel_health_check(ConnectorResId, SourceResId).
 
 bridge_id(Config) ->
     Type = ?BRIDGE_TYPE_BIN,
@@ -564,7 +575,7 @@ get_pull_worker_pids(Config) ->
     Pids.
 
 get_async_worker_pids(Config) ->
-    ResourceId = resource_id(Config),
+    ResourceId = connector_resource_id(Config),
     Pids =
         [
             AsyncWorkerPid
@@ -902,7 +913,6 @@ unauthenticated_response() ->
 %%------------------------------------------------------------------------------
 
 t_start_stop(Config) ->
-    ResourceId = resource_id(Config),
     [#{pubsub_topic := PubSubTopic} | _] = ?config(topic_mapping, Config),
     ?check_trace(
         begin
@@ -913,7 +923,7 @@ t_start_stop(Config) ->
                 ),
             ?assertMatch({ok, _}, create_bridge(Config)),
             {ok, _} = snabbkaffe:receive_events(SRef0),
-            ?assertMatch({ok, connected}, emqx_resource_manager:health_check(ResourceId)),
+            ?assertMatch({ok, connected}, health_check(Config)),
 
             ?assertMatch(ok, remove_bridge(Config)),
             ok
@@ -1045,8 +1055,8 @@ t_consume_ok(Config) ->
             ),
 
             %% Check that the bridge probe API doesn't leak atoms.
-            ProbeRes0 = probe_bridge_api(Config),
+            ?assertMatch({ok, {{_, 204, _}, _Headers, _Body}}, probe_bridge_api(Config)),
-            ?assertMatch({ok, {{_, 204, _}, _Headers, _Body}}, ProbeRes0),
+            ?assertMatch({ok, {{_, 204, _}, _Headers, _Body}}, probe_bridge_api(Config)),
             AtomsBefore = erlang:system_info(atom_count),
             %% Probe again; shouldn't have created more atoms.
             ProbeRes1 = probe_bridge_api(Config),
@@ -1143,13 +1153,12 @@ t_bridge_rule_action_source(Config) ->
     ok.
 
 t_on_get_status(Config) ->
-    ResourceId = resource_id(Config),
     emqx_bridge_testlib:t_on_get_status(Config, #{failure_status => connecting}),
     %% no workers alive
     ?retry(
         _Interval0 = 200,
         _NAttempts0 = 20,
-        ?assertMatch({ok, connected}, emqx_resource_manager:health_check(ResourceId))
+        ?assertMatch({ok, connected}, health_check(Config))
     ),
     WorkerPids = get_pull_worker_pids(Config),
     emqx_utils:pmap(
@@ -1163,7 +1172,7 @@ t_on_get_status(Config) ->
         end,
         WorkerPids
     ),
-    ?assertMatch({ok, connecting}, emqx_resource_manager:health_check(ResourceId)),
+    ?assertMatch({ok, connecting}, health_check(Config)),
     ok.
 
 t_create_update_via_http_api(Config) ->
@@ -1331,7 +1340,6 @@ t_multiple_pull_workers(Config) ->
 t_nonexistent_topic(Config) ->
     BridgeName = ?config(bridge_name, Config),
     [Mapping0] = ?config(topic_mapping, Config),
-    ResourceId = resource_id(Config),
     PubSubTopic = <<"nonexistent-", (emqx_guid:to_hexstr(emqx_guid:gen()))/binary>>,
     TopicMapping0 = [Mapping0#{pubsub_topic := PubSubTopic}],
     TopicMapping = emqx_utils_maps:binary_key_map(TopicMapping0),
@@ -1347,11 +1355,14 @@ t_nonexistent_topic(Config) ->
                 ),
             ?assertMatch(
                 {ok, disconnected},
-                emqx_resource_manager:health_check(ResourceId)
+                health_check(Config)
             ),
             ?assertMatch(
-                {ok, _Group, #{error := {unhealthy_target, "GCP PubSub topics are invalid" ++ _}}},
+                #{
-                emqx_resource_manager:lookup_cached(ResourceId)
+                    status := disconnected,
+                    error := {unhealthy_target, "GCP PubSub topics are invalid" ++ _}
+                },
+                health_check_channel(Config)
             ),
             %% now create the topic and restart the bridge
             ensure_topic(Config, PubSubTopic),
@@ -1362,11 +1373,14 @@ t_nonexistent_topic(Config) ->
             ?retry(
                 _Interval0 = 200,
                 _NAttempts0 = 20,
-                ?assertMatch({ok, connected}, emqx_resource_manager:health_check(ResourceId))
+                ?assertMatch({ok, connected}, health_check(Config))
             ),
             ?assertMatch(
-                {ok, _Group, #{error := undefined}},
+                #{
-                emqx_resource_manager:lookup_cached(ResourceId)
+                    status := connected,
+                    error := undefined
+                },
+                health_check_channel(Config)
             ),
             ok
         end,
@@ -1383,7 +1397,6 @@ t_nonexistent_topic(Config) ->
 t_topic_deleted_while_consumer_is_running(Config) ->
     TopicMapping = [#{pubsub_topic := PubSubTopic} | _] = ?config(topic_mapping, Config),
     NTopics = length(TopicMapping),
-    ResourceId = resource_id(Config),
     ?check_trace(
         begin
             {ok, SRef0} =
@@ -1395,7 +1408,7 @@ t_topic_deleted_while_consumer_is_running(Config) ->
             {ok, _} = create_bridge(Config),
             {ok, _} = snabbkaffe:receive_events(SRef0),
 
-            ?assertMatch({ok, connected}, emqx_resource_manager:health_check(ResourceId)),
+            ?assertMatch({ok, connected}, health_check(Config)),
 
             %% curiously, gcp pubsub doesn't seem to return any errors from the
             %% subscription if the topic is deleted while the subscription still exists...
@@ -1421,26 +1434,39 @@ t_connection_down_before_starting(Config) ->
     ProxyName = ?config(proxy_name, Config),
     ProxyHost = ?config(proxy_host, Config),
     ProxyPort = ?config(proxy_port, Config),
-    ResourceId = resource_id(Config),
     ?check_trace(
         begin
-            emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
+            ?force_ordering(
-                ?assertMatch(
+                #{?snk_kind := gcp_pubsub_consumer_worker_about_to_spawn},
-                    {{ok, _}, {ok, _}},
+                #{?snk_kind := will_cut_connection}
-                    ?wait_async_action(
+            ),
-                        create_bridge(Config),
+            ?force_ordering(
-                        #{?snk_kind := gcp_pubsub_consumer_worker_init},
+                #{?snk_kind := connection_down},
-                        10_000
+                #{?snk_kind := gcp_pubsub_consumer_worker_create_subscription_enter}
-                    )
+            ),
-                ),
+            spawn_link(fun() ->
-                ?assertMatch({ok, connecting}, emqx_resource_manager:health_check(ResourceId)),
+                ?tp(notice, will_cut_connection, #{}),
-                ok
+                emqx_common_test_helpers:enable_failure(down, ProxyName, ProxyHost, ProxyPort),
+                ?tp(notice, connection_down, #{})
             end),
+            %% check retries
+            {ok, SRef0} =
+                snabbkaffe:subscribe(
+                    ?match_event(#{?snk_kind := "gcp_pubsub_consumer_worker_subscription_error"}),
+                    _NEvents0 = 2,
+                    10_000
+                ),
+            {ok, _} = create_bridge(Config),
+            {ok, _} = snabbkaffe:receive_events(SRef0),
+            ?assertMatch({ok, connecting}, health_check(Config)),
+
+            emqx_common_test_helpers:heal_failure(down, ProxyName, ProxyHost, ProxyPort),
             ?retry(
                 _Interval0 = 200,
                 _NAttempts0 = 20,
-                ?assertMatch({ok, connected}, emqx_resource_manager:health_check(ResourceId))
+                ?assertMatch({ok, connected}, health_check(Config))
             ),
+
             ok
         end,
         []
@@ -1451,7 +1477,6 @@ t_connection_timeout_before_starting(Config) ->
     ProxyName = ?config(proxy_name, Config),
     ProxyHost = ?config(proxy_host, Config),
     ProxyPort = ?config(proxy_port, Config),
-    ResourceId = resource_id(Config),
     ?check_trace(
         begin
             emqx_common_test_helpers:with_failure(
@@ -1464,14 +1489,14 @@ t_connection_timeout_before_starting(Config) ->
                             10_000
                         )
                     ),
-                    ?assertMatch({ok, connecting}, emqx_resource_manager:health_check(ResourceId)),
+                    ?assertMatch({ok, connecting}, health_check(Config)),
                     ok
                 end
             ),
             ?retry(
                 _Interval0 = 200,
                 _NAttempts0 = 20,
-                ?assertMatch({ok, connected}, emqx_resource_manager:health_check(ResourceId))
+                ?assertMatch({ok, connected}, health_check(Config))
             ),
             ok
         end,
@@ -1480,7 +1505,6 @@ t_connection_timeout_before_starting(Config) ->
     ok.
 
 t_pull_worker_death(Config) ->
-    ResourceId = resource_id(Config),
     ?check_trace(
         begin
             ?assertMatch(
@@ -1500,13 +1524,13 @@ t_pull_worker_death(Config) ->
                     ok
             after 500 -> ct:fail("pull worker didn't die")
             end,
-            ?assertMatch({ok, connecting}, emqx_resource_manager:health_check(ResourceId)),
+            ?assertMatch({ok, connecting}, health_check(Config)),
 
             %% recovery
             ?retry(
                 _Interval0 = 200,
                 _NAttempts0 = 20,
-                ?assertMatch({ok, connected}, emqx_resource_manager:health_check(ResourceId))
+                ?assertMatch({ok, connected}, health_check(Config))
             ),
 
             ok
@@ -1641,19 +1665,32 @@ t_connection_error_while_creating_subscription(Config) ->
     ProxyPort = ?config(proxy_port, Config),
     ?check_trace(
         begin
-            emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
+            ?force_ordering(
-                %% check retries
+                #{?snk_kind := gcp_pubsub_consumer_worker_init},
-                {ok, SRef0} =
+                #{?snk_kind := will_cut_connection}
-                    snabbkaffe:subscribe(
+            ),
-                        ?match_event(#{?snk_kind := "gcp_pubsub_consumer_worker_subscription_error"}),
+            ?force_ordering(
-                        _NEvents0 = 2,
+                #{?snk_kind := connection_down},
-                        10_000
+                #{?snk_kind := gcp_pubsub_consumer_worker_create_subscription_enter}
-                    ),
+            ),
-                {ok, _} = create_bridge(Config),
+            spawn_link(fun() ->
-                {ok, _} = snabbkaffe:receive_events(SRef0),
+                ?tp(notice, will_cut_connection, #{}),
-                ok
+                emqx_common_test_helpers:enable_failure(down, ProxyName, ProxyHost, ProxyPort),
+                ?tp(notice, connection_down, #{})
             end),
-            %% eventually succeeds
+            %% check retries
+            {ok, SRef0} =
+                snabbkaffe:subscribe(
+                    ?match_event(#{?snk_kind := "gcp_pubsub_consumer_worker_subscription_error"}),
+                    _NEvents0 = 2,
+                    10_000
+                ),
+            {ok, _} = create_bridge(Config),
+            {ok, _} = snabbkaffe:receive_events(SRef0),
+            emqx_common_test_helpers:heal_failure(down, ProxyName, ProxyHost, ProxyPort),
+
+            %% should eventually succeed
+            ?tp(notice, "waiting for recovery", #{}),
             {ok, _} =
                 ?block_until(
                     #{?snk_kind := "gcp_pubsub_consumer_worker_subscription_created"},
@@ -1763,7 +1800,6 @@ t_subscription_patch_error(Config) ->
 
 t_topic_deleted_while_creating_subscription(Config) ->
     [#{pubsub_topic := PubSubTopic}] = ?config(topic_mapping, Config),
-    ResourceId = resource_id(Config),
     ?check_trace(
         begin
             ?force_ordering(
@@ -1787,7 +1823,7 @@ t_topic_deleted_while_creating_subscription(Config) ->
                     #{?snk_kind := gcp_pubsub_consumer_worker_terminate},
                     10_000
                 ),
-            ?assertMatch({ok, disconnected}, emqx_resource_manager:health_check(ResourceId)),
+            ?assertMatch({ok, disconnected}, health_check(Config)),
             ok
         end,
         []
@@ -1797,7 +1833,6 @@ t_topic_deleted_while_creating_subscription(Config) ->
 t_topic_deleted_while_patching_subscription(Config) ->
     BridgeName = ?config(bridge_name, Config),
     [#{pubsub_topic := PubSubTopic}] = ?config(topic_mapping, Config),
-    ResourceId = resource_id(Config),
     ?check_trace(
         begin
             {{ok, _}, {ok, _}} =
@@ -1832,7 +1867,7 @@ t_topic_deleted_while_patching_subscription(Config) ->
                     #{?snk_kind := "gcp_pubsub_consumer_worker_subscription_ready"},
                     10_000
                 ),
-            ?assertMatch({ok, connected}, emqx_resource_manager:health_check(ResourceId)),
+            ?assertMatch({ok, connected}, health_check(Config)),
             ok
         end,
         []
@@ -1840,7 +1875,6 @@ t_topic_deleted_while_patching_subscription(Config) ->
     ok.
 
 t_subscription_deleted_while_consumer_is_running(Config) ->
-    ResourceId = resource_id(Config),
     ?check_trace(
         begin
             {{ok, _}, {ok, #{subscription_id := SubscriptionId}}} =
@@ -1870,7 +1904,7 @@ t_subscription_deleted_while_consumer_is_running(Config) ->
             {ok, _} = snabbkaffe:receive_events(SRef0),
             {ok, _} = snabbkaffe:receive_events(SRef1),
 
-            ?assertMatch({ok, connected}, emqx_resource_manager:health_check(ResourceId)),
+            ?assertMatch({ok, connected}, health_check(Config)),
             ok
         end,
         []
@@ -1880,7 +1914,6 @@ t_subscription_deleted_while_consumer_is_running(Config) ->
 t_subscription_and_topic_deleted_while_consumer_is_running(Config) ->
     ct:timetrap({seconds, 90}),
     [#{pubsub_topic := PubSubTopic}] = ?config(topic_mapping, Config),
-    ResourceId = resource_id(Config),
     ?check_trace(
         begin
             {{ok, _}, {ok, #{subscription_id := SubscriptionId}}} =
@@ -1896,7 +1929,11 @@ t_subscription_and_topic_deleted_while_consumer_is_running(Config) ->
             delete_subscription(Config, SubscriptionId),
             {ok, _} = ?block_until(#{?snk_kind := gcp_pubsub_consumer_worker_terminate}, 60_000),
 
-            ?assertMatch({ok, disconnected}, emqx_resource_manager:health_check(ResourceId)),
+            ?retry(
+                _Sleep0 = 100,
+                _Retries = 20,
+                ?assertMatch({ok, disconnected}, health_check(Config))
+            ),
             ok
         end,
         []
@@ -2161,7 +2198,6 @@ t_get_subscription(Config) ->
 
 t_permission_denied_topic_check(Config) ->
     [#{pubsub_topic := PubSubTopic}] = ?config(topic_mapping, Config),
-    ResourceId = resource_id(Config),
     ?check_trace(
         begin
             %% the emulator does not check any credentials
@@ -2178,19 +2214,17 @@ t_permission_denied_topic_check(Config) ->
                     end
                 end,
                 fun() ->
-                    {{ok, _}, {ok, _}} =
+                    {ok, _} = create_bridge(Config),
-                        ?wait_async_action(
-                            create_bridge(Config),
-                            #{?snk_kind := gcp_pubsub_stop},
-                            5_000
-                        ),
                     ?assertMatch(
                         {ok, disconnected},
-                        emqx_resource_manager:health_check(ResourceId)
+                        health_check(Config)
                     ),
                     ?assertMatch(
-                        {ok, _Group, #{error := {unhealthy_target, "Permission denied" ++ _}}},
+                        #{
-                        emqx_resource_manager:lookup_cached(ResourceId)
+                            status := disconnected,
+                            error := {unhealthy_target, "Permission denied" ++ _}
+                        },
+                        health_check_channel(Config)
                     ),
                     ok
                 end
@@ -2236,7 +2270,6 @@ t_permission_denied_worker(Config) ->
 
 t_unauthenticated_topic_check(Config) ->
     [#{pubsub_topic := PubSubTopic}] = ?config(topic_mapping, Config),
-    ResourceId = resource_id(Config),
     ?check_trace(
         begin
             %% the emulator does not check any credentials
@@ -2253,19 +2286,17 @@ t_unauthenticated_topic_check(Config) ->
                     end
                 end,
                 fun() ->
-                    {{ok, _}, {ok, _}} =
+                    {ok, _} = create_bridge(Config),
-                        ?wait_async_action(
-                            create_bridge(Config),
-                            #{?snk_kind := gcp_pubsub_stop},
-                            5_000
-                        ),
                     ?assertMatch(
                         {ok, disconnected},
-                        emqx_resource_manager:health_check(ResourceId)
+                        health_check(Config)
                     ),
                     ?assertMatch(
-                        {ok, _Group, #{error := {unhealthy_target, "Permission denied" ++ _}}},
+                        #{
-                        emqx_resource_manager:lookup_cached(ResourceId)
+                            status := disconnected,
+                            error := {unhealthy_target, "Permission denied" ++ _}
+                        },
+                        health_check_channel(Config)
                     ),
                     ok
                 end

apps/emqx_bridge_gcp_pubsub/test/emqx_bridge_v2_gcp_pubsub_consumer_SUITE.erl

@@ -0,0 +1,167 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_bridge_v2_gcp_pubsub_consumer_SUITE).
+
+-compile(nowarn_export_all).
+-compile(export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+
+-define(CONNECTOR_TYPE_BIN, <<"gcp_pubsub_consumer">>).
+-define(SOURCE_TYPE_BIN, <<"gcp_pubsub_consumer">>).
+
+%%------------------------------------------------------------------------------
+%% CT boilerplate
+%%------------------------------------------------------------------------------
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+init_per_suite(Config) ->
+    emqx_bridge_gcp_pubsub_consumer_SUITE:init_per_suite(Config).
+
+end_per_suite(Config) ->
+    emqx_bridge_gcp_pubsub_consumer_SUITE:end_per_suite(Config).
+
+init_per_testcase(TestCase, Config) ->
+    common_init_per_testcase(TestCase, Config).
+
+common_init_per_testcase(TestCase, Config0) ->
+    ct:timetrap(timer:seconds(60)),
+    ServiceAccountJSON =
+        #{<<"project_id">> := ProjectId} =
+        emqx_bridge_gcp_pubsub_utils:generate_service_account_json(),
+    Name = atom_to_binary(TestCase),
+    ConnectorConfig = connector_config(Name, ServiceAccountJSON),
+    PubsubTopic = Name,
+    SourceConfig = source_config(#{
+        connector => Name,
+        parameters => #{topic => PubsubTopic}
+    }),
+    Config = [
+        {bridge_kind, source},
+        {source_type, ?SOURCE_TYPE_BIN},
+        {source_name, Name},
+        {source_config, SourceConfig},
+        {connector_name, Name},
+        {connector_type, ?CONNECTOR_TYPE_BIN},
+        {connector_config, ConnectorConfig},
+        {service_account_json, ServiceAccountJSON},
+        {project_id, ProjectId},
+        {pubsub_topic, PubsubTopic}
+        | Config0
+    ],
+    ok = emqx_bridge_gcp_pubsub_consumer_SUITE:ensure_topic(Config, PubsubTopic),
+    Config.
+
+end_per_testcase(_Testcase, Config) ->
+    ProxyHost = ?config(proxy_host, Config),
+    ProxyPort = ?config(proxy_port, Config),
+    emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
+    emqx_bridge_v2_testlib:delete_all_bridges_and_connectors(),
+    emqx_common_test_helpers:call_janitor(60_000),
+    ok = snabbkaffe:stop(),
+    ok.
+
+%%------------------------------------------------------------------------------
+%% Helper fns
+%%------------------------------------------------------------------------------
+
+connector_config(Name, ServiceAccountJSON) ->
+    InnerConfigMap0 =
+        #{
+            <<"enable">> => true,
+            <<"tags">> => [<<"bridge">>],
+            <<"description">> => <<"my cool bridge">>,
+            <<"connect_timeout">> => <<"5s">>,
+            <<"pool_size">> => 8,
+            <<"pipelining">> => <<"100">>,
+            <<"max_retries">> => <<"2">>,
+            <<"service_account_json">> => ServiceAccountJSON,
+            <<"resource_opts">> =>
+                #{
+                    <<"health_check_interval">> => <<"1s">>,
+                    <<"start_after_created">> => true,
+                    <<"start_timeout">> => <<"5s">>
+                }
+        },
+    emqx_bridge_v2_testlib:parse_and_check_connector(?SOURCE_TYPE_BIN, Name, InnerConfigMap0).
+
+source_config(Overrides0) ->
+    Overrides = emqx_utils_maps:binary_key_map(Overrides0),
+    CommonConfig =
+        #{
+            <<"enable">> => true,
+            <<"connector">> => <<"please override">>,
+            <<"parameters">> =>
+                #{
+                    <<"topic">> => <<"my-topic">>
+                },
+            <<"resource_opts">> => #{
+                <<"health_check_interval">> => <<"1s">>,
+                <<"request_ttl">> => <<"1s">>,
+                <<"resume_interval">> => <<"1s">>
+            }
+        },
+    maps:merge(CommonConfig, Overrides).
+
+%%------------------------------------------------------------------------------
+%% Testcases
+%%------------------------------------------------------------------------------
+
+t_start_stop(Config) ->
+    ok = emqx_bridge_v2_testlib:t_start_stop(Config, gcp_pubsub_stop),
+    ok.
+
+t_consume(Config) ->
+    Topic = ?config(pubsub_topic, Config),
+    Payload = #{<<"key">> => <<"value">>},
+    Attributes = #{<<"hkey">> => <<"hval">>},
+    ProduceFn = fun() ->
+        emqx_bridge_gcp_pubsub_consumer_SUITE:pubsub_publish(
+            Config,
+            Topic,
+            [
+                #{
+                    <<"data">> => Payload,
+                    <<"orderingKey">> => <<"ok">>,
+                    <<"attributes">> => Attributes
+                }
+            ]
+        )
+    end,
+    Encoded = emqx_utils_json:encode(Payload),
+    CheckFn = fun(Message) ->
+        ?assertMatch(
+            #{
+                attributes := Attributes,
+                message_id := _,
+                ordering_key := <<"ok">>,
+                publish_time := _,
+                topic := Topic,
+                value := Encoded
+            },
+            Message
+        )
+    end,
+    ok = emqx_bridge_v2_testlib:t_consume(
+        Config,
+        #{
+            consumer_ready_tracepoint => ?match_event(
+                #{?snk_kind := "gcp_pubsub_consumer_worker_subscription_ready"}
+            ),
+            produce_fn => ProduceFn,
+            check_fn => CheckFn,
+            produce_tracepoint => ?match_event(
+                #{
+                    ?snk_kind := "gcp_pubsub_consumer_worker_handle_message",
+                    ?snk_span := {complete, _}
+                }
+            )
+        }
+    ),
+    ok.

apps/emqx_connector/src/schema/emqx_connector_ee_schema.erl

@@ -26,6 +26,8 @@ resource_type(azure_event_hub_producer) ->
     emqx_bridge_kafka_impl_producer;
 resource_type(confluent_producer) ->
     emqx_bridge_kafka_impl_producer;
+resource_type(gcp_pubsub_consumer) ->
+    emqx_bridge_gcp_pubsub_impl_consumer;
 resource_type(gcp_pubsub_producer) ->
     emqx_bridge_gcp_pubsub_impl_producer;
 resource_type(hstreamdb) ->
@@ -120,6 +122,14 @@ connector_structs() ->
                     required => false
                 }
             )},
+        {gcp_pubsub_consumer,
+            mk(
+                hoconsc:map(name, ref(emqx_bridge_gcp_pubsub_consumer_schema, "config_connector")),
+                #{
+                    desc => <<"GCP PubSub Consumer Connector Config">>,
+                    required => false
+                }
+            )},
         {gcp_pubsub_producer,
             mk(
                 hoconsc:map(name, ref(emqx_bridge_gcp_pubsub_producer_schema, "config_connector")),
@@ -319,6 +329,7 @@ schema_modules() ->
     [
         emqx_bridge_azure_event_hub,
         emqx_bridge_confluent_producer,
+        emqx_bridge_gcp_pubsub_consumer_schema,
         emqx_bridge_gcp_pubsub_producer_schema,
         emqx_bridge_hstreamdb,
         emqx_bridge_kafka,
@@ -355,6 +366,11 @@ api_schemas(Method) ->
         api_ref(
             emqx_bridge_confluent_producer, <<"confluent_producer">>, Method ++ "_connector"
         ),
+        api_ref(
+            emqx_bridge_gcp_pubsub_consumer_schema,
+            <<"gcp_pubsub_consumer">>,
+            Method ++ "_connector"
+        ),
         api_ref(
             emqx_bridge_gcp_pubsub_producer_schema,
             <<"gcp_pubsub_producer">>,

apps/emqx_connector/src/schema/emqx_connector_schema.erl

@@ -126,6 +126,8 @@ connector_type_to_bridge_types(azure_event_hub_producer) ->
     [azure_event_hub_producer];
 connector_type_to_bridge_types(confluent_producer) ->
     [confluent_producer];
+connector_type_to_bridge_types(gcp_pubsub_consumer) ->
+    [gcp_pubsub_consumer];
 connector_type_to_bridge_types(gcp_pubsub_producer) ->
     [gcp_pubsub, gcp_pubsub_producer];
 connector_type_to_bridge_types(hstreamdb) ->
@@ -529,14 +531,6 @@ fields(connectors) ->
                     required => false
                 }
             )}
-        % {mqtt_subscriber,
-        %     mk(
-        %         hoconsc:map(name, ref(emqx_bridge_mqtt_connector_schema, "config_connector")),
-        %         #{
-        %             desc => <<"MQTT Subscriber Connector Config">>,
-        %             required => false
-        %         }
-        %     )}
     ] ++ enterprise_fields_connectors();
 fields("node_status") ->
     [

apps/emqx_resource/include/emqx_resource.hrl

@@ -115,6 +115,7 @@
     | {error, term()}.
 
 -type action_resource_id() :: resource_id().
+-type source_resource_id() :: resource_id().
 -type connector_resource_id() :: resource_id().
 -type message_tag() :: action_resource_id().
 

rel/i18n/emqx_bridge_gcp_pubsub_consumer_schema.hocon

@@ -0,0 +1,18 @@
+emqx_bridge_gcp_pubsub_consumer_schema {
+
+  source_parameters.desc:
+  """Source specific configs."""
+  source_parameters.label:
+  """Source Specific Configs"""
+
+  consumer_source.desc:
+  """Source configs."""
+  consumer_source.label:
+  """Source"""
+
+  config_connector.desc:
+  """Configuration for a GCP PubSub Consumer Client."""
+  config_connector.label:
+  """GCP PubSub Consumer Client Configuration"""
+
+}