feat(gcp_pubsub): implement GCP PubSub bridge (ee5.0)

apps/emqx/test/emqx_common_test_helpers.erl

@@ -66,6 +66,10 @@
 
 -export([clear_screen/0]).
 -export([with_mock/4]).
+-export([
+    on_exit/1,
+    run_on_exit_callbacks/0
+]).
 
 %% Toxiproxy API
 -export([
@@ -930,3 +934,24 @@ latency_up_proxy(off, Name, ProxyHost, ProxyPort) ->
         [],
         [{body_format, binary}]
     ).
+
+%%-------------------------------------------------------------------------------
+%% Testcase teardown utilities
+%%-------------------------------------------------------------------------------
+
+get_on_exit_callbacks() ->
+    persistent_term:get({?MODULE, on_exit}, []).
+
+put_on_exit_callbacks(Funs) ->
+    persistent_term:put({?MODULE, on_exit}, Funs).
+
+on_exit(Fun) ->
+    Callbacks = get_on_exit_callbacks(),
+    put_on_exit_callbacks([Fun | Callbacks]).
+
+%% should be called at `end_per_testcase'.
+%% TODO: scope per group and suite as well?
+run_on_exit_callbacks() ->
+    Callbacks = get_on_exit_callbacks(),
+    put_on_exit_callbacks([]),
+    lists:foreach(fun(Fun) -> Fun() end, Callbacks).

apps/emqx_connector/include/emqx_connector_tables.hrl

@@ -0,0 +1,17 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 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.
+%%--------------------------------------------------------------------
+
+-define(JWT_TABLE, emqx_connector_jwt).

apps/emqx_connector/rebar.config

@@ -7,6 +7,7 @@
 
 {deps, [
     {emqx, {path, "../emqx"}},
+    {emqx_resource, {path, "../emqx_resource"}},
     {eldap2, {git, "https://github.com/emqx/eldap2", {tag, "v0.2.2"}}},
     {mysql, {git, "https://github.com/emqx/mysql-otp", {tag, "1.7.1"}}},
     {epgsql, {git, "https://github.com/emqx/epgsql", {tag, "4.7-emqx.2"}}},

apps/emqx_connector/src/emqx_connector.app.src

@@ -16,6 +16,7 @@
         mysql,
         mongodb,
         ehttpc,
+        jose,
         emqx,
         emqtt
     ]},

apps/emqx_connector/src/emqx_connector_jwt.erl

@@ -0,0 +1,46 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 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_connector_jwt).
+
+-include("emqx_connector_tables.hrl").
+-include_lib("emqx_resource/include/emqx_resource.hrl").
+
+%% API
+-export([
+    lookup_jwt/1,
+    lookup_jwt/2
+]).
+
+-type jwt() :: binary().
+
+-spec lookup_jwt(resource_id()) -> {ok, jwt()} | {error, not_found}.
+lookup_jwt(ResourceId) ->
+    ?MODULE:lookup_jwt(?JWT_TABLE, ResourceId).
+
+-spec lookup_jwt(ets:table(), resource_id()) -> {ok, jwt()} | {error, not_found}.
+lookup_jwt(TId, ResourceId) ->
+    try
+        case ets:lookup(TId, {ResourceId, jwt}) of
+            [{{ResourceId, jwt}, JWT}] ->
+                {ok, JWT};
+            [] ->
+                {error, not_found}
+        end
+    catch
+        error:badarg ->
+            {error, not_found}
+    end.

apps/emqx_connector/src/emqx_connector_jwt_sup.erl

@@ -0,0 +1,99 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 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_connector_jwt_sup).
+
+-behaviour(supervisor).
+
+-include("emqx_connector_tables.hrl").
+
+-export([
+    start_link/0,
+    ensure_worker_present/2,
+    ensure_worker_deleted/1
+]).
+
+-export([init/1]).
+
+-type worker_id() :: term().
+
+start_link() ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+init([]) ->
+    ensure_jwt_table(),
+    SupFlags = #{
+        strategy => one_for_one,
+        intensity => 10,
+        period => 5,
+        auto_shutdown => never
+    },
+    ChildSpecs = [],
+    {ok, {SupFlags, ChildSpecs}}.
+
+%% @doc Starts a new JWT worker.  The caller should use
+%% `emqx_connector_jwt_sup:ensure_jwt/1' to ensure that a JWT has
+%% been stored, if synchronization is needed.
+-spec ensure_worker_present(worker_id(), map()) ->
+    {ok, supervisor:child()} | {error, term()}.
+ensure_worker_present(Id, Config) ->
+    ChildSpec = jwt_worker_child_spec(Id, Config),
+    case supervisor:start_child(?MODULE, ChildSpec) of
+        {ok, Pid} ->
+            {ok, Pid};
+        {error, {already_started, Pid}} ->
+            {ok, Pid};
+        {error, already_present} ->
+            supervisor:restart_child(?MODULE, Id)
+    end.
+
+%% @doc Stops a given JWT worker by its id.
+-spec ensure_worker_deleted(worker_id()) -> ok.
+ensure_worker_deleted(Id) ->
+    case supervisor:terminate_child(?MODULE, Id) of
+        ok ->
+            _ = supervisor:delete_child(?MODULE, Id),
+            ok;
+        {error, not_found} ->
+            ok
+    end.
+
+jwt_worker_child_spec(Id, Config) ->
+    #{
+        id => Id,
+        start => {emqx_connector_jwt_worker, start_link, [Config]},
+        restart => transient,
+        type => worker,
+        significant => false,
+        shutdown => brutal_kill,
+        modules => [emqx_connector_jwt_worker]
+    }.
+
+-spec ensure_jwt_table() -> ok.
+ensure_jwt_table() ->
+    case ets:whereis(?JWT_TABLE) of
+        undefined ->
+            Opts = [
+                named_table,
+                public,
+                {read_concurrency, true},
+                ordered_set
+            ],
+            _ = ets:new(?JWT_TABLE, Opts),
+            ok;
+        _ ->
+            ok
+    end.

apps/emqx_connector/src/emqx_connector_jwt_worker.erl

@@ -0,0 +1,237 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 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_connector_jwt_worker).
+
+-behaviour(gen_server).
+
+%% API
+-export([
+    start_link/1,
+    ensure_jwt/1
+]).
+
+%% gen_server API
+-export([
+    init/1,
+    handle_continue/2,
+    handle_call/3,
+    handle_cast/2,
+    handle_info/2,
+    format_status/1,
+    format_status/2
+]).
+
+-include_lib("emqx_resource/include/emqx_resource.hrl").
+-include_lib("jose/include/jose_jwk.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+
+-type config() :: #{
+    private_key := binary(),
+    resource_id := resource_id(),
+    expiration := timer:time(),
+    table := ets:table(),
+    iss := binary(),
+    sub := binary(),
+    aud := binary(),
+    kid := binary(),
+    alg := binary()
+}.
+-type jwt() :: binary().
+-type state() :: #{
+    refresh_timer := undefined | timer:tref(),
+    resource_id := resource_id(),
+    expiration := timer:time(),
+    table := ets:table(),
+    jwt := undefined | jwt(),
+    %% only undefined during startup
+    jwk := undefined | jose_jwk:key(),
+    iss := binary(),
+    sub := binary(),
+    aud := binary(),
+    kid := binary(),
+    alg := binary()
+}.
+
+-define(refresh_jwt, refresh_jwt).
+
+%%-----------------------------------------------------------------------------------------
+%% API
+%%-----------------------------------------------------------------------------------------
+
+-spec start_link(config()) -> gen_server:start_ret().
+start_link(
+    #{
+        private_key := _,
+        expiration := _,
+        resource_id := _,
+        table := _,
+        iss := _,
+        sub := _,
+        aud := _,
+        kid := _,
+        alg := _
+    } = Config
+) ->
+    gen_server:start_link(?MODULE, Config, []).
+
+-spec ensure_jwt(pid()) -> reference().
+ensure_jwt(Worker) ->
+    Ref = alias([reply]),
+    gen_server:cast(Worker, {ensure_jwt, Ref}),
+    Ref.
+
+%%-----------------------------------------------------------------------------------------
+%% gen_server API
+%%-----------------------------------------------------------------------------------------
+
+-spec init(config()) ->
+    {ok, state(), {continue, {make_key, binary()}}}
+    | {stop, {error, term()}}.
+init(#{private_key := PrivateKeyPEM} = Config) ->
+    State0 = maps:without([private_key], Config),
+    State = State0#{
+        jwk => undefined,
+        jwt => undefined,
+        refresh_timer => undefined
+    },
+    {ok, State, {continue, {make_key, PrivateKeyPEM}}}.
+
+handle_continue({make_key, PrivateKeyPEM}, State0) ->
+    ?tp(connector_jwt_worker_make_key, #{state => State0}),
+    case jose_jwk:from_pem(PrivateKeyPEM) of
+        JWK = #jose_jwk{} ->
+            State = State0#{jwk := JWK},
+            {noreply, State, {continue, create_token}};
+        [] ->
+            ?tp(connector_jwt_worker_startup_error, #{error => empty_key}),
+            {stop, {shutdown, {error, empty_key}}, State0};
+        {error, Reason} ->
+            Error = {invalid_private_key, Reason},
+            ?tp(connector_jwt_worker_startup_error, #{error => Error}),
+            {stop, {shutdown, {error, Error}}, State0};
+        Error0 ->
+            Error = {invalid_private_key, Error0},
+            ?tp(connector_jwt_worker_startup_error, #{error => Error}),
+            {stop, {shutdown, {error, Error}}, State0}
+    end;
+handle_continue(create_token, State0) ->
+    State = generate_and_store_jwt(State0),
+    {noreply, State}.
+
+handle_call(_Req, _From, State) ->
+    {reply, {error, bad_call}, State}.
+
+handle_cast({ensure_jwt, From}, State0 = #{jwt := JWT}) ->
+    State =
+        case JWT of
+            undefined ->
+                generate_and_store_jwt(State0);
+            _ ->
+                State0
+        end,
+    From ! {From, token_created},
+    {noreply, State};
+handle_cast(_Req, State) ->
+    {noreply, State}.
+
+handle_info({timeout, TRef, ?refresh_jwt}, State0 = #{refresh_timer := TRef}) ->
+    State = generate_and_store_jwt(State0),
+    {noreply, State};
+handle_info(_Msg, State) ->
+    {noreply, State}.
+
+format_status(State) ->
+    censor_secrets(State).
+
+format_status(_Opt, [_PDict, State0]) ->
+    State = censor_secrets(State0),
+    [{data, [{"State", State}]}].
+
+%%-----------------------------------------------------------------------------------------
+%% Helper fns
+%%-----------------------------------------------------------------------------------------
+
+-spec do_generate_jwt(state()) -> jwt().
+do_generate_jwt(
+    #{
+        expiration := ExpirationMS,
+        iss := Iss,
+        sub := Sub,
+        aud := Aud,
+        kid := KId,
+        alg := Alg,
+        jwk := JWK
+    } = _State
+) ->
+    Headers = #{
+        <<"alg">> => Alg,
+        <<"kid">> => KId
+    },
+    Now = erlang:system_time(seconds),
+    ExpirationS = erlang:convert_time_unit(ExpirationMS, millisecond, second),
+    Claims = #{
+        <<"iss">> => Iss,
+        <<"sub">> => Sub,
+        <<"aud">> => Aud,
+        <<"iat">> => Now,
+        <<"exp">> => Now + ExpirationS
+    },
+    JWT0 = jose_jwt:sign(JWK, Headers, Claims),
+    {_, JWT} = jose_jws:compact(JWT0),
+    JWT.
+
+-spec generate_and_store_jwt(state()) -> state().
+generate_and_store_jwt(State0) ->
+    JWT = do_generate_jwt(State0),
+    store_jwt(State0, JWT),
+    ?tp(connector_jwt_worker_refresh, #{jwt => JWT}),
+    State1 = State0#{jwt := JWT},
+    ensure_timer(State1).
+
+-spec store_jwt(state(), jwt()) -> ok.
+store_jwt(#{resource_id := ResourceId, table := TId}, JWT) ->
+    true = ets:insert(TId, {{ResourceId, jwt}, JWT}),
+    ?tp(connector_jwt_worker_token_stored, #{resource_id => ResourceId}),
+    ok.
+
+-spec ensure_timer(state()) -> state().
+ensure_timer(
+    State = #{
+        refresh_timer := undefined,
+        expiration := ExpirationMS0
+    }
+) ->
+    ExpirationMS = max(5_000, ExpirationMS0 - 5_000),
+    TRef = erlang:start_timer(ExpirationMS, self(), ?refresh_jwt),
+    State#{refresh_timer => TRef};
+ensure_timer(State) ->
+    State.
+
+-spec censor_secrets(state()) -> map().
+censor_secrets(State) ->
+    maps:map(
+        fun
+            (Key, _Value) when
+                Key =:= jwt;
+                Key =:= jwk
+            ->
+                "******";
+            (_Key, Value) ->
+                Value
+        end,
+        State
+    ).

apps/emqx_connector/src/emqx_connector_sup.erl

@@ -33,7 +33,8 @@ init([]) ->
         period => 20
     },
     ChildSpecs = [
-        child_spec(emqx_connector_mqtt)
+        child_spec(emqx_connector_mqtt),
+        child_spec(emqx_connector_jwt_sup)
     ],
     {ok, {SupFlags, ChildSpecs}}.
 
@@ -46,5 +47,3 @@ child_spec(Mod) ->
         type => supervisor,
         modules => [Mod]
     }.
-
-%% internal functions

apps/emqx_connector/test/emqx_connector_jwt_SUITE.erl

@@ -0,0 +1,69 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 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_connector_jwt_SUITE).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include("emqx_connector_tables.hrl").
+
+-compile([export_all, nowarn_export_all]).
+
+%%-----------------------------------------------------------------------------
+%% CT boilerplate
+%%-----------------------------------------------------------------------------
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+init_per_suite(Config) ->
+    emqx_common_test_helpers:start_apps([emqx_connector]),
+    Config.
+
+end_per_suite(_Config) ->
+    emqx_common_test_helpers:stop_apps([emqx_connector]),
+    ok.
+
+init_per_testcase(_TestCase, Config) ->
+    Config.
+
+end_per_testcase(_TestCase, _Config) ->
+    ets:delete_all_objects(?JWT_TABLE),
+    ok.
+
+%%-----------------------------------------------------------------------------
+%% Helper fns
+%%-----------------------------------------------------------------------------
+
+insert_jwt(TId, ResourceId, JWT) ->
+    ets:insert(TId, {{ResourceId, jwt}, JWT}).
+
+%%-----------------------------------------------------------------------------
+%% Test cases
+%%-----------------------------------------------------------------------------
+
+t_lookup_jwt_ok(_Config) ->
+    TId = ?JWT_TABLE,
+    JWT = <<"some jwt">>,
+    ResourceId = <<"resource id">>,
+    true = insert_jwt(TId, ResourceId, JWT),
+    ?assertEqual({ok, JWT}, emqx_connector_jwt:lookup_jwt(ResourceId)),
+    ok.
+
+t_lookup_jwt_missing(_Config) ->
+    ResourceId = <<"resource id">>,
+    ?assertEqual({error, not_found}, emqx_connector_jwt:lookup_jwt(ResourceId)),
+    ok.

apps/emqx_connector/test/emqx_connector_jwt_worker_SUITE.erl

@@ -0,0 +1,340 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 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_connector_jwt_worker_SUITE).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+-include_lib("jose/include/jose_jwt.hrl").
+-include_lib("jose/include/jose_jws.hrl").
+
+-compile([export_all, nowarn_export_all]).
+
+%%-----------------------------------------------------------------------------
+%% CT boilerplate
+%%-----------------------------------------------------------------------------
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+init_per_suite(Config) ->
+    Config.
+
+end_per_suite(_Config) ->
+    ok.
+
+%%-----------------------------------------------------------------------------
+%% Helper fns
+%%-----------------------------------------------------------------------------
+
+generate_private_key_pem() ->
+    PublicExponent = 65537,
+    Size = 2048,
+    Key = public_key:generate_key({rsa, Size, PublicExponent}),
+    DERKey = public_key:der_encode('PrivateKeyInfo', Key),
+    public_key:pem_encode([{'PrivateKeyInfo', DERKey, not_encrypted}]).
+
+generate_config() ->
+    PrivateKeyPEM = generate_private_key_pem(),
+    ResourceID = emqx_guid:gen(),
+    #{
+        private_key => PrivateKeyPEM,
+        expiration => timer:hours(1),
+        resource_id => ResourceID,
+        table => ets:new(test_jwt_table, [ordered_set, public]),
+        iss => <<"issuer">>,
+        sub => <<"subject">>,
+        aud => <<"audience">>,
+        kid => <<"key id">>,
+        alg => <<"RS256">>
+    }.
+
+is_expired(JWT) ->
+    #jose_jwt{fields = #{<<"exp">> := Exp}} = jose_jwt:peek(JWT),
+    Now = erlang:system_time(seconds),
+    Now >= Exp.
+
+%%-----------------------------------------------------------------------------
+%% Test cases
+%%-----------------------------------------------------------------------------
+
+t_create_success(_Config) ->
+    Config = generate_config(),
+    Res = emqx_connector_jwt_worker:start_link(Config),
+    ?assertMatch({ok, _}, Res),
+    {ok, Worker} = Res,
+    Ref = emqx_connector_jwt_worker:ensure_jwt(Worker),
+    receive
+        {Ref, token_created} ->
+            ok
+    after 1_000 ->
+        ct:fail(
+            "should have confirmed token creation; msgs: ~0p",
+            [process_info(self(), messages)]
+        )
+    end,
+    ok.
+
+t_empty_key(_Config) ->
+    Config0 = generate_config(),
+    Config = Config0#{private_key := <<>>},
+    process_flag(trap_exit, true),
+    ?check_trace(
+        ?wait_async_action(
+            ?assertMatch({ok, _}, emqx_connector_jwt_worker:start_link(Config)),
+            #{?snk_kind := connector_jwt_worker_startup_error},
+            1_000
+        ),
+        fun(Trace) ->
+            ?assertMatch(
+                [#{error := empty_key}],
+                ?of_kind(connector_jwt_worker_startup_error, Trace)
+            ),
+            ok
+        end
+    ),
+    ok.
+
+t_unknown_error(_Config) ->
+    Config0 = generate_config(),
+    Config = Config0#{private_key := <<>>},
+    process_flag(trap_exit, true),
+    ?check_trace(
+        {_, {ok, _}} = ?wait_async_action(
+            emqx_common_test_helpers:with_mock(
+                jose_jwk,
+                from_pem,
+                fun(_PrivateKeyPEM) -> {error, some_strange_error} end,
+                fun() ->
+                    ?assertMatch({ok, _}, emqx_connector_jwt_worker:start_link(Config))
+                end
+            ),
+            #{?snk_kind := connector_jwt_worker_startup_error},
+            1_000
+        ),
+        fun(Trace) ->
+            ?assertMatch(
+                [#{error := {invalid_private_key, some_strange_error}}],
+                ?of_kind(connector_jwt_worker_startup_error, Trace)
+            ),
+            ok
+        end
+    ),
+    ok.
+
+t_invalid_pem(_Config) ->
+    Config0 = generate_config(),
+    InvalidPEM = public_key:pem_encode([
+        {'PrivateKeyInfo', <<"xxxxxx">>, not_encrypted},
+        {'PrivateKeyInfo', <<"xxxxxx">>, not_encrypted}
+    ]),
+    Config = Config0#{private_key := InvalidPEM},
+    process_flag(trap_exit, true),
+    ?check_trace(
+        ?wait_async_action(
+            ?assertMatch({ok, _}, emqx_connector_jwt_worker:start_link(Config)),
+            #{?snk_kind := connector_jwt_worker_startup_error},
+            1_000
+        ),
+        fun(Trace) ->
+            ?assertMatch(
+                [#{error := {invalid_private_key, _}}],
+                ?of_kind(connector_jwt_worker_startup_error, Trace)
+            ),
+            ok
+        end
+    ),
+    ok.
+
+t_refresh(_Config) ->
+    Config0 =
+        #{
+            table := Table,
+            resource_id := ResourceId
+        } = generate_config(),
+    Config = Config0#{expiration => 5_000},
+    ?check_trace(
+        begin
+            {{ok, _Pid}, {ok, _Event}} =
+                ?wait_async_action(
+                    emqx_connector_jwt_worker:start_link(Config),
+                    #{?snk_kind := connector_jwt_worker_token_stored},
+                    5_000
+                ),
+            {ok, FirstJWT} = emqx_connector_jwt:lookup_jwt(Table, ResourceId),
+            ?block_until(
+                #{
+                    ?snk_kind := connector_jwt_worker_refresh,
+                    jwt := JWT0
+                } when JWT0 =/= FirstJWT,
+                15_000
+            ),
+            {ok, SecondJWT} = emqx_connector_jwt:lookup_jwt(Table, ResourceId),
+            ?assertNot(is_expired(SecondJWT)),
+            ?assert(is_expired(FirstJWT)),
+            {FirstJWT, SecondJWT}
+        end,
+        fun({FirstJWT, SecondJWT}, Trace) ->
+            ?assertMatch(
+                [_, _ | _],
+                ?of_kind(connector_jwt_worker_token_stored, Trace)
+            ),
+            ?assertNotEqual(FirstJWT, SecondJWT),
+            ok
+        end
+    ),
+    ok.
+
+t_format_status(_Config) ->
+    Config = generate_config(),
+    {ok, Pid} = emqx_connector_jwt_worker:start_link(Config),
+    {status, _, _, Props} = sys:get_status(Pid),
+    [State] = [
+        State
+     || Info = [_ | _] <- Props,
+        {data, Data = [_ | _]} <- Info,
+        {"State", State} <- Data
+    ],
+    ?assertMatch(
+        #{
+            jwt := "******",
+            jwk := "******"
+        },
+        State
+    ),
+    ok.
+
+t_lookup_ok(_Config) ->
+    Config =
+        #{
+            table := Table,
+            resource_id := ResourceId,
+            private_key := PrivateKeyPEM,
+            aud := Aud,
+            iss := Iss,
+            sub := Sub,
+            kid := KId
+        } = generate_config(),
+    {ok, Worker} = emqx_connector_jwt_worker:start_link(Config),
+    Ref = emqx_connector_jwt_worker:ensure_jwt(Worker),
+    receive
+        {Ref, token_created} ->
+            ok
+    after 500 ->
+        error(timeout)
+    end,
+    Res = emqx_connector_jwt:lookup_jwt(Table, ResourceId),
+    ?assertMatch({ok, _}, Res),
+    {ok, JWT} = Res,
+    ?assert(is_binary(JWT)),
+    JWK = jose_jwk:from_pem(PrivateKeyPEM),
+    {IsValid, ParsedJWT, JWS} = jose_jwt:verify_strict(JWK, [<<"RS256">>], JWT),
+    ?assertMatch(
+        #jose_jwt{
+            fields = #{
+                <<"aud">> := Aud,
+                <<"iss">> := Iss,
+                <<"sub">> := Sub,
+                <<"exp">> := _,
+                <<"iat">> := _
+            }
+        },
+        ParsedJWT
+    ),
+    ?assertNot(is_expired(JWT)),
+    ?assertMatch(
+        #jose_jws{
+            alg = {_, 'RS256'},
+            fields = #{
+                <<"kid">> := KId,
+                <<"typ">> := <<"JWT">>
+            }
+        },
+        JWS
+    ),
+    ?assert(IsValid),
+    ok.
+
+t_lookup_not_found(_Config) ->
+    Table = ets:new(test_jwt_table, [ordered_set, public]),
+    InexistentResource = <<"xxx">>,
+    ?assertEqual(
+        {error, not_found},
+        emqx_connector_jwt:lookup_jwt(Table, InexistentResource)
+    ),
+    ok.
+
+t_lookup_badarg(_Config) ->
+    InexistentTable = i_dont_exist,
+    InexistentResource = <<"xxx">>,
+    ?assertEqual(
+        {error, not_found},
+        emqx_connector_jwt:lookup_jwt(InexistentTable, InexistentResource)
+    ),
+    ok.
+
+t_start_supervised_worker(_Config) ->
+    {ok, _} = emqx_connector_jwt_sup:start_link(),
+    Config = #{resource_id := ResourceId} = generate_config(),
+    {ok, Pid} = emqx_connector_jwt_sup:ensure_worker_present(ResourceId, Config),
+    Ref = emqx_connector_jwt_worker:ensure_jwt(Pid),
+    receive
+        {Ref, token_created} ->
+            ok
+    after 5_000 ->
+        ct:fail("timeout")
+    end,
+    MRef = monitor(process, Pid),
+    ?assert(is_process_alive(Pid)),
+    ok = emqx_connector_jwt_sup:ensure_worker_deleted(ResourceId),
+    receive
+        {'DOWN', MRef, process, Pid, _} ->
+            ok
+    after 1_000 ->
+        ct:fail("timeout")
+    end,
+    ok.
+
+t_start_supervised_worker_already_started(_Config) ->
+    {ok, _} = emqx_connector_jwt_sup:start_link(),
+    Config = #{resource_id := ResourceId} = generate_config(),
+    {ok, Pid0} = emqx_connector_jwt_sup:ensure_worker_present(ResourceId, Config),
+    {ok, Pid1} = emqx_connector_jwt_sup:ensure_worker_present(ResourceId, Config),
+    ?assertEqual(Pid0, Pid1),
+    ok.
+
+t_start_supervised_worker_already_present(_Config) ->
+    {ok, _} = emqx_connector_jwt_sup:start_link(),
+    Config = #{resource_id := ResourceId} = generate_config(),
+    {ok, Pid0} = emqx_connector_jwt_sup:ensure_worker_present(ResourceId, Config),
+    Ref = monitor(process, Pid0),
+    exit(Pid0, {shutdown, normal}),
+    receive
+        {'DOWN', Ref, process, Pid0, {shutdown, normal}} -> ok
+    after 1_000 -> error(worker_didnt_stop)
+    end,
+    {ok, Pid1} = emqx_connector_jwt_sup:ensure_worker_present(ResourceId, Config),
+    ?assertNotEqual(Pid0, Pid1),
+    ok.
+
+t_unknown_requests(_Config) ->
+    Config = generate_config(),
+    {ok, Worker} = emqx_connector_jwt_worker:start_link(Config),
+    Worker ! unknown_info,
+    gen_server:cast(Worker, unknown_cast),
+    ?assertEqual({error, bad_call}, gen_server:call(Worker, unknown_call)),
+    ok.

apps/emqx_connector/test/emqx_connector_web_hook_server.erl

@@ -0,0 +1,105 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 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_connector_web_hook_server).
+
+-compile([nowarn_export_all, export_all]).
+
+-behaviour(supervisor).
+-behaviour(cowboy_handler).
+
+%%------------------------------------------------------------------------------
+%% API
+%%------------------------------------------------------------------------------
+
+start_link(Port, Path) ->
+    start_link(Port, Path, false).
+
+start_link(Port, Path, SSLOpts) ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, [Port, Path, SSLOpts]).
+
+stop() ->
+    try
+        gen_server:stop(?MODULE)
+    catch
+        exit:noproc ->
+            ok
+    end.
+
+set_handler(F) when is_function(F, 2) ->
+    true = ets:insert(?MODULE, {handler, F}),
+    ok.
+
+%%------------------------------------------------------------------------------
+%% supervisor API
+%%------------------------------------------------------------------------------
+
+init([Port, Path, SSLOpts]) ->
+    Dispatch = cowboy_router:compile(
+        [
+            {'_', [{Path, ?MODULE, []}]}
+        ]
+    ),
+
+    ProtoOpts = #{env => #{dispatch => Dispatch}},
+
+    Tab = ets:new(?MODULE, [set, named_table, public]),
+    ets:insert(Tab, {handler, fun default_handler/2}),
+
+    {Transport, TransOpts, CowboyModule} = transport_settings(Port, SSLOpts),
+
+    ChildSpec = ranch:child_spec(?MODULE, Transport, TransOpts, CowboyModule, ProtoOpts),
+
+    {ok, {#{}, [ChildSpec]}}.
+
+%%------------------------------------------------------------------------------
+%% cowboy_server API
+%%------------------------------------------------------------------------------
+
+init(Req, State) ->
+    [{handler, Handler}] = ets:lookup(?MODULE, handler),
+    Handler(Req, State).
+
+%%------------------------------------------------------------------------------
+%% Internal functions
+%%------------------------------------------------------------------------------
+
+transport_settings(Port, _SSLOpts = false) ->
+    TransOpts = #{
+        socket_opts => [{port, Port}],
+        connection_type => supervisor
+    },
+    {ranch_tcp, TransOpts, cowboy_clear};
+transport_settings(Port, SSLOpts) ->
+    TransOpts = #{
+        socket_opts => [
+            {port, Port},
+            {next_protocols_advertised, [<<"h2">>, <<"http/1.1">>]},
+            {alpn_preferred_protocols, [<<"h2">>, <<"http/1.1">>]}
+            | SSLOpts
+        ],
+        connection_type => supervisor
+    },
+    {ranch_ssl, TransOpts, cowboy_tls}.
+
+default_handler(Req0, State) ->
+    Req = cowboy_req:reply(
+        400,
+        #{<<"content-type">> => <<"text/plain">>},
+        <<"">>,
+        Req0
+    ),
+    {ok, Req, State}.

apps/emqx_dashboard/src/emqx_dashboard_swagger.erl

@@ -702,6 +702,8 @@ typename_to_spec("wordsize()", _Mod) ->
     #{type => string, example => <<"1024KB">>};
 typename_to_spec("map()", _Mod) ->
     #{type => object, example => #{}};
+typename_to_spec("service_account_json()", _Mod) ->
+    #{type => object, example => #{}};
 typename_to_spec("#{" ++ _, Mod) ->
     typename_to_spec("map()", Mod);
 typename_to_spec("qos()", _Mod) ->

apps/emqx_resource/include/emqx_resource.hrl

@@ -76,6 +76,8 @@
 -type query_result() ::
     ok
     | {ok, term()}
+    | {ok, term(), term()}
+    | {ok, term(), term(), term()}
     | {error, {recoverable_error, term()}}
     | {error, term()}.
 

lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_gcp_pubsub.conf

@@ -0,0 +1,122 @@
+emqx_ee_bridge_gcp_pubsub {
+    desc_config {
+        desc {
+            en: """Configuration for a GCP PubSub bridge."""
+            zh: """GCP PubSub 桥接配置"""
+        }
+        label {
+            en: "GCP PubSub Bridge Configuration"
+            zh: "GCP PubSub 桥接配置"
+        }
+    }
+
+    desc_type {
+        desc {
+            en: """The Bridge Type"""
+            zh: """桥接类型"""
+        }
+        label {
+            en: "Bridge Type"
+            zh: "桥接类型"
+        }
+    }
+    desc_name {
+        desc {
+            en: """Bridge name, used as a human-readable description of the bridge."""
+            zh: """桥接名字，可读描述"""
+        }
+        label {
+            en: "Bridge Name"
+            zh: "桥接名字"
+        }
+    }
+
+    connect_timeout {
+        desc {
+          en: "The timeout when connecting to the HTTP server."
+          zh: "连接HTTP服务器的超时时间。"
+        }
+        label: {
+              en: "Connect Timeout"
+              zh: "连接超时"
+            }
+    }
+
+    max_retries {
+        desc {
+          en: "Max retry times if an error occurs when sending a request."
+          zh: "请求出错时的最大重试次数。"
+        }
+        label: {
+              en: "Max Retries"
+              zh: "最大重试次数"
+            }
+    }
+
+    pool_size {
+        desc {
+          en: "The pool size."
+          zh: "连接池大小。"
+        }
+        label: {
+              en: "Pool Size"
+              zh: "连接池大小"
+            }
+    }
+
+    pipelining {
+        desc {
+          en: "A positive integer. Whether to send HTTP requests continuously, when set to 1, it means that after each HTTP request is sent, you need to wait for the server to return and then continue to send the next request."
+          zh: "正整数，设置最大可发送的异步 HTTP 请求数量。当设置为 1 时，表示每次发送完成 HTTP 请求后都需要等待服务器返回，再继续发送下一个请求。"
+        }
+        label: {
+              en: "HTTP Pipelineing"
+              zh: "HTTP 管道"
+            }
+    }
+
+    request_timeout {
+        desc {
+          en: "HTTP request timeout."
+          zh: "HTTP 请求超时。"
+        }
+        label: {
+              en: "Request Timeout"
+              zh: "HTTP 请求超时"
+            }
+    }
+
+    payload_template {
+        desc {
+          en: "The template for formatting the outgoing messages.  If undefined, will send all the available context in JSON format."
+          zh: "用于格式化外发信息的模板。 如果未定义，将以JSON格式发送所有可用的上下文。"
+        }
+        label: {
+              en: "Payload template"
+              zh: "有效载荷模板"
+            }
+    }
+
+    pubsub_topic {
+        desc {
+          en: "The GCP PubSub topic to publish messages to."
+          zh: "要发布消息的GCP PubSub主题。"
+        }
+        label: {
+              en: "GCP PubSub Topic"
+              zh: "GCP公共子课题"
+            }
+    }
+
+    service_account_json {
+        desc {
+          en: "JSON containing the GCP Service Account credentials to be used with PubSub."
+          zh: "包含将与PubSub一起使用的GCP服务账户凭证的JSON。"
+        }
+        label: {
+              en: "GCP Service Account JSON"
+              zh: "GCP服务账户JSON"
+            }
+    }
+
+}

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl

@@ -14,13 +14,13 @@
 
 api_schemas(Method) ->
     [
+        ref(emqx_ee_bridge_gcp_pubsub, Method),
         ref(emqx_ee_bridge_kafka, Method),
         ref(emqx_ee_bridge_mysql, Method),
         ref(emqx_ee_bridge_mongodb, Method ++ "_rs"),
         ref(emqx_ee_bridge_mongodb, Method ++ "_sharded"),
         ref(emqx_ee_bridge_mongodb, Method ++ "_single"),
         ref(emqx_ee_bridge_hstreamdb, Method),
-        %% ref(emqx_ee_bridge_influxdb, Method ++ "_udp"),
         ref(emqx_ee_bridge_influxdb, Method ++ "_api_v1"),
         ref(emqx_ee_bridge_influxdb, Method ++ "_api_v2")
     ].
@@ -29,6 +29,7 @@ schema_modules() ->
     [
         emqx_ee_bridge_kafka,
         emqx_ee_bridge_hstreamdb,
+        emqx_ee_bridge_gcp_pubsub,
         emqx_ee_bridge_influxdb,
         emqx_ee_bridge_mongodb,
         emqx_ee_bridge_mysql
@@ -49,11 +50,11 @@ examples(Method) ->
 resource_type(Type) when is_binary(Type) -> resource_type(binary_to_atom(Type, utf8));
 resource_type(kafka) -> emqx_bridge_impl_kafka;
 resource_type(hstreamdb) -> emqx_ee_connector_hstreamdb;
+resource_type(gcp_pubsub) -> emqx_ee_bridge_gcp_pubsub_resource;
 resource_type(mongodb_rs) -> emqx_connector_mongo;
 resource_type(mongodb_sharded) -> emqx_connector_mongo;
 resource_type(mongodb_single) -> emqx_connector_mongo;
 resource_type(mysql) -> emqx_connector_mysql;
-resource_type(influxdb_udp) -> emqx_ee_connector_influxdb;
 resource_type(influxdb_api_v1) -> emqx_ee_connector_influxdb;
 resource_type(influxdb_api_v2) -> emqx_ee_connector_influxdb.
 
@@ -75,6 +76,14 @@ fields(bridges) ->
                     required => false
                 }
             )},
+        {gcp_pubsub,
+            mk(
+                hoconsc:map(name, ref(emqx_ee_bridge_gcp_pubsub, "config")),
+                #{
+                    desc => <<"EMQX Enterprise Config">>,
+                    required => false
+                }
+            )},
         {mysql,
             mk(
                 hoconsc:map(name, ref(emqx_ee_bridge_mysql, "config")),
@@ -109,7 +118,6 @@ influxdb_structs() ->
                 }
             )}
      || Protocol <- [
-            %% influxdb_udp,
             influxdb_api_v1,
             influxdb_api_v2
         ]

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_gcp_pubsub.erl

@@ -0,0 +1,224 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_ee_bridge_gcp_pubsub).
+
+-include_lib("emqx_bridge/include/emqx_bridge.hrl").
+-include_lib("typerefl/include/types.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+
+-import(hoconsc, [mk/2, enum/1, ref/2]).
+
+%% hocon_schema API
+-export([
+    namespace/0,
+    roots/0,
+    fields/1,
+    desc/1
+]).
+-export([
+    service_account_json_validator/1,
+    service_account_json_converter/1
+]).
+
+%% emqx_ee_bridge "unofficial" API
+-export([conn_bridge_examples/1]).
+
+-type service_account_json() :: map().
+-reflect_type([service_account_json/0]).
+
+-define(DEFAULT_PIPELINE_SIZE, 100).
+
+%%-------------------------------------------------------------------------------------------------
+%% `hocon_schema' API
+%%-------------------------------------------------------------------------------------------------
+
+namespace() ->
+    "bridge_gcp_pubsub".
+
+roots() ->
+    [].
+
+fields("config") ->
+    emqx_bridge_schema:common_bridge_fields() ++
+        emqx_resource_schema:fields("resource_opts") ++
+        fields(bridge_config);
+fields(bridge_config) ->
+    [
+        {connect_timeout,
+            sc(
+                emqx_schema:duration_ms(),
+                #{
+                    default => "15s",
+                    desc => ?DESC("connect_timeout")
+                }
+            )},
+        {pool_size,
+            sc(
+                pos_integer(),
+                #{
+                    default => 8,
+                    desc => ?DESC("pool_size")
+                }
+            )},
+        {pipelining,
+            sc(
+                pos_integer(),
+                #{
+                    default => ?DEFAULT_PIPELINE_SIZE,
+                    desc => ?DESC("pipelining")
+                }
+            )},
+        {max_retries,
+            sc(
+                non_neg_integer(),
+                #{
+                    required => false,
+                    default => 2,
+                    desc => ?DESC("max_retries")
+                }
+            )},
+        {request_timeout,
+            sc(
+                emqx_schema:duration_ms(),
+                #{
+                    required => false,
+                    default => "15s",
+                    desc => ?DESC("request_timeout")
+                }
+            )},
+        {payload_template,
+            sc(
+                binary(),
+                #{
+                    default => <<>>,
+                    desc => ?DESC("payload_template")
+                }
+            )},
+        {pubsub_topic,
+            sc(
+                binary(),
+                #{
+                    required => true,
+                    desc => ?DESC("pubsub_topic")
+                }
+            )},
+        {service_account_json,
+            sc(
+                service_account_json(),
+                #{
+                    required => true,
+                    validator => fun ?MODULE:service_account_json_validator/1,
+                    converter => fun ?MODULE:service_account_json_converter/1,
+                    sensitive => true,
+                    desc => ?DESC("service_account_json")
+                }
+            )}
+    ];
+fields("get") ->
+    emqx_bridge_schema:metrics_status_fields() ++ fields("post");
+fields("post") ->
+    [type_field(), name_field() | fields("config")];
+fields("put") ->
+    fields("config").
+
+desc("config") ->
+    ?DESC("desc_config");
+desc(_) ->
+    undefined.
+
+conn_bridge_examples(Method) ->
+    [
+        #{
+            <<"gcp_pubsub">> => #{
+                summary => <<"GCP PubSub Bridge">>,
+                value => values(Method)
+            }
+        }
+    ].
+
+values(get) ->
+    maps:merge(values(post), ?METRICS_EXAMPLE);
+values(post) ->
+    #{
+        <<"pubsub_topic">> => <<"mytopic">>,
+        <<"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">>
+            }
+    };
+values(put) ->
+    values(post).
+
+%%-------------------------------------------------------------------------------------------------
+%% Helper fns
+%%-------------------------------------------------------------------------------------------------
+
+sc(Type, Meta) -> hoconsc:mk(Type, Meta).
+
+type_field() ->
+    {type, mk(enum([gcp_pubsub]), #{required => true, desc => ?DESC("desc_type")})}.
+
+name_field() ->
+    {name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}.
+
+-spec service_account_json_validator(map()) ->
+    ok
+    | {error, {wrong_type, term()}}
+    | {error, {missing_keys, [binary()]}}.
+service_account_json_validator(Map) ->
+    ExpectedKeys = [
+        <<"type">>,
+        <<"project_id">>,
+        <<"private_key_id">>,
+        <<"private_key">>,
+        <<"client_email">>
+    ],
+    MissingKeys = lists:sort([
+        K
+     || K <- ExpectedKeys,
+        not maps:is_key(K, Map)
+    ]),
+    Type = maps:get(<<"type">>, Map, null),
+    case {MissingKeys, Type} of
+        {[], <<"service_account">>} ->
+            ok;
+        {[], Type} ->
+            {error, {wrong_type, Type}};
+        {_, _} ->
+            {error, {missing_keys, MissingKeys}}
+    end.
+
+service_account_json_converter(Map) when is_map(Map) ->
+    ExpectedKeys = [
+        <<"type">>,
+        <<"project_id">>,
+        <<"private_key_id">>,
+        <<"private_key">>,
+        <<"client_email">>
+    ],
+    maps:with(ExpectedKeys, Map);
+service_account_json_converter(Val) ->
+    Val.

lib-ee/emqx_ee_bridge/src/gcp_pubsub/emqx_ee_bridge_gcp_pubsub_resource.erl

@@ -0,0 +1,599 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_ee_bridge_gcp_pubsub_resource).
+
+-behaviour(emqx_resource).
+
+-include_lib("emqx_connector/include/emqx_connector_tables.hrl").
+-include_lib("emqx_resource/include/emqx_resource.hrl").
+-include_lib("typerefl/include/types.hrl").
+-include_lib("emqx/include/logger.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+
+%% `emqx_resource' API
+-export([
+    callback_mode/0,
+    on_start/2,
+    on_stop/2,
+    on_query/3,
+    on_query_async/4,
+    on_batch_query/3,
+    on_batch_query_async/4,
+    on_get_status/2,
+    is_buffer_supported/0
+]).
+-export([reply_delegator/3]).
+
+-type bridge_id() :: binary().
+-type jwt_worker() :: binary().
+-type service_account_json() :: emqx_ee_bridge_gcp_pubsub:service_account_json().
+-type config() :: #{
+    connect_timeout := emqx_schema:duration_ms(),
+    max_retries := non_neg_integer(),
+    pubsub_topic := binary(),
+    request_timeout := emqx_schema:duration_ms(),
+    service_account_json := service_account_json(),
+    any() => term()
+}.
+-type state() :: #{
+    connect_timeout := timer:time(),
+    instance_id := manager_id(),
+    jwt_worker_id := jwt_worker(),
+    max_retries := non_neg_integer(),
+    payload_template := emqx_plugin_libs_rule:tmpl_token(),
+    pool_name := atom(),
+    project_id := binary(),
+    pubsub_topic := binary(),
+    request_timeout := timer:time()
+}.
+-type headers() :: [{binary(), iodata()}].
+-type body() :: iodata().
+-type status_code() :: 100..599.
+
+-define(DEFAULT_PIPELINE_SIZE, 100).
+
+%%-------------------------------------------------------------------------------------------------
+%% emqx_resource API
+%%-------------------------------------------------------------------------------------------------
+
+%% TODO: check
+is_buffer_supported() -> false.
+
+callback_mode() -> async_if_possible.
+
+-spec on_start(manager_id(), config()) -> {ok, state()} | {error, term()}.
+on_start(
+    InstanceId,
+    #{
+        connect_timeout := ConnectTimeout,
+        max_retries := MaxRetries,
+        payload_template := PayloadTemplate,
+        pool_size := PoolSize,
+        pubsub_topic := PubSubTopic,
+        request_timeout := RequestTimeout
+    } = Config
+) ->
+    ?SLOG(info, #{
+        msg => "starting_gcp_pubsub_bridge",
+        connector => InstanceId,
+        config => Config
+    }),
+    %% emulating the emulator behavior
+    %% https://cloud.google.com/pubsub/docs/emulator
+    HostPort = os:getenv("PUBSUB_EMULATOR_HOST", "pubsub.googleapis.com:443"),
+    {Host, Port} = emqx_connector_schema_lib:parse_server(
+        HostPort, #{host_type => hostname, default_port => 443}
+    ),
+    PoolType = random,
+    Transport = tls,
+    TransportOpts = emqx_tls_lib:to_client_opts(#{enable => true, verify => verify_none}),
+    NTransportOpts = emqx_misc:ipv6_probe(TransportOpts),
+    PoolOpts = [
+        {host, Host},
+        {port, Port},
+        {connect_timeout, ConnectTimeout},
+        {keepalive, 30_000},
+        {pool_type, PoolType},
+        {pool_size, PoolSize},
+        {transport, Transport},
+        {transport_opts, NTransportOpts},
+        {enable_pipelining, maps:get(enable_pipelining, Config, ?DEFAULT_PIPELINE_SIZE)}
+    ],
+    #{
+        jwt_worker_id := JWTWorkerId,
+        project_id := ProjectId
+    } = ensure_jwt_worker(InstanceId, Config),
+    PoolName = emqx_plugin_libs_pool:pool_name(InstanceId),
+    State = #{
+        connect_timeout => ConnectTimeout,
+        instance_id => InstanceId,
+        jwt_worker_id => JWTWorkerId,
+        max_retries => MaxRetries,
+        payload_template => emqx_plugin_libs_rule:preproc_tmpl(PayloadTemplate),
+        pool_name => PoolName,
+        project_id => ProjectId,
+        pubsub_topic => PubSubTopic,
+        request_timeout => RequestTimeout
+    },
+    ?tp(
+        gcp_pubsub_on_start_before_starting_pool,
+        #{
+            instance_id => InstanceId,
+            pool_name => PoolName,
+            pool_opts => PoolOpts
+        }
+    ),
+    ?tp(gcp_pubsub_starting_ehttpc_pool, #{pool_name => PoolName}),
+    case ehttpc_sup:start_pool(PoolName, PoolOpts) of
+        {ok, _} ->
+            {ok, State};
+        {error, {already_started, _}} ->
+            ?tp(gcp_pubsub_ehttpc_pool_already_started, #{pool_name => PoolName}),
+            {ok, State};
+        {error, Reason} ->
+            ?tp(gcp_pubsub_ehttpc_pool_start_failure, #{
+                pool_name => PoolName,
+                reason => Reason
+            }),
+            {error, Reason}
+    end.
+
+-spec on_stop(manager_id(), state()) -> ok | {error, term()}.
+on_stop(
+    InstanceId,
+    _State = #{
+        jwt_worker_id := JWTWorkerId,
+        pool_name := PoolName
+    }
+) ->
+    ?tp(gcp_pubsub_stop, #{instance_id => InstanceId, jwt_worker_id => JWTWorkerId}),
+    ?SLOG(info, #{
+        msg => "stopping_gcp_pubsub_bridge",
+        connector => InstanceId
+    }),
+    emqx_connector_jwt_sup:ensure_worker_deleted(JWTWorkerId),
+    ehttpc_sup:stop_pool(PoolName).
+
+-spec on_query(
+    bridge_id(),
+    {send_message, map()},
+    state()
+) ->
+    {ok, status_code(), headers()}
+    | {ok, status_code(), headers(), body()}
+    | {error, {recoverable_error, term()}}
+    | {error, term()}.
+on_query(BridgeId, {send_message, Selected}, State) ->
+    Requests = [{send_message, Selected}],
+    ?TRACE(
+        "QUERY_SYNC",
+        "gcp_pubsub_received",
+        #{requests => Requests, connector => BridgeId, state => State}
+    ),
+    do_send_requests_sync(State, Requests, BridgeId).
+
+-spec on_query_async(
+    bridge_id(),
+    {send_message, map()},
+    {ReplyFun :: function(), Args :: list()},
+    state()
+) -> ok.
+on_query_async(BridgeId, {send_message, Selected}, ReplyFunAndArgs, State) ->
+    Requests = [{send_message, Selected}],
+    ?TRACE(
+        "QUERY_ASYNC",
+        "gcp_pubsub_received",
+        #{requests => Requests, connector => BridgeId, state => State}
+    ),
+    do_send_requests_async(State, Requests, ReplyFunAndArgs, BridgeId).
+
+-spec on_batch_query(
+    bridge_id(),
+    [{send_message, map()}],
+    state()
+) ->
+    {ok, status_code(), headers()}
+    | {ok, status_code(), headers(), body()}
+    | {error, {recoverable_error, term()}}
+    | {error, term()}.
+on_batch_query(BridgeId, Requests, State) ->
+    ?TRACE(
+        "QUERY_SYNC",
+        "gcp_pubsub_received",
+        #{requests => Requests, connector => BridgeId, state => State}
+    ),
+    do_send_requests_sync(State, Requests, BridgeId).
+
+-spec on_batch_query_async(
+    bridge_id(),
+    [{send_message, map()}],
+    {ReplyFun :: function(), Args :: list()},
+    state()
+) -> ok.
+on_batch_query_async(BridgeId, Requests, ReplyFunAndArgs, State) ->
+    ?TRACE(
+        "QUERY_ASYNC",
+        "gcp_pubsub_received",
+        #{requests => Requests, connector => BridgeId, state => State}
+    ),
+    do_send_requests_async(State, Requests, ReplyFunAndArgs, BridgeId).
+
+-spec on_get_status(manager_id(), state()) -> connected | disconnected.
+on_get_status(InstanceId, State) ->
+    #{
+        connect_timeout := Timeout,
+        pool_name := PoolName
+    } = State,
+    case do_get_status(InstanceId, PoolName, Timeout) of
+        true ->
+            connected;
+        false ->
+            ?SLOG(error, #{
+                msg => "gcp_pubsub_bridge_get_status_failed",
+                state => State
+            }),
+            disconnected
+    end.
+
+%%-------------------------------------------------------------------------------------------------
+%% Helper fns
+%%-------------------------------------------------------------------------------------------------
+
+-spec ensure_jwt_worker(manager_id(), config()) ->
+    #{
+        jwt_worker_id := jwt_worker(),
+        project_id := binary()
+    }.
+ensure_jwt_worker(InstanceId, #{
+    service_account_json := ServiceAccountJSON,
+    pubsub_topic := PubSubTopic
+}) ->
+    #{
+        project_id := ProjectId,
+        private_key_id := KId,
+        private_key := PrivateKeyPEM,
+        client_email := ServiceAccountEmail
+    } = ServiceAccountJSON,
+    %% fixed for pubsub; trailing slash is important.
+    Aud = <<"https://pubsub.googleapis.com/">>,
+    ExpirationMS = timer:hours(1),
+    Alg = <<"RS256">>,
+    Config = #{
+        private_key => PrivateKeyPEM,
+        resource_id => InstanceId,
+        expiration => ExpirationMS,
+        table => ?JWT_TABLE,
+        iss => ServiceAccountEmail,
+        sub => ServiceAccountEmail,
+        aud => Aud,
+        kid => KId,
+        alg => Alg
+    },
+
+    JWTWorkerId = <<"gcp_pubsub_jwt_worker:", InstanceId/binary>>,
+    Worker =
+        case emqx_connector_jwt_sup:ensure_worker_present(JWTWorkerId, Config) of
+            {ok, Worker0} ->
+                Worker0;
+            Error ->
+                ?tp(
+                    gcp_pubsub_bridge_jwt_worker_failed_to_start,
+                    #{instance_id => InstanceId, reason => Error}
+                ),
+                ?SLOG(error, #{
+                    msg => "failed_to_start_gcp_pubsub_jwt_worker",
+                    instance_id => InstanceId,
+                    pubsub_topic => PubSubTopic,
+                    reason => Error
+                }),
+                _ = emqx_connector_jwt_sup:ensure_worker_deleted(JWTWorkerId),
+                throw(failed_to_start_jwt_worker)
+        end,
+    MRef = monitor(process, Worker),
+    Ref = emqx_connector_jwt_worker:ensure_jwt(Worker),
+
+    %% to ensure that this resource and its actions will be ready to
+    %% serve when started, we must ensure that the first JWT has been
+    %% produced by the worker.
+    receive
+        {Ref, token_created} ->
+            ?tp(gcp_pubsub_bridge_jwt_created, #{resource_id => InstanceId}),
+            demonitor(MRef, [flush]),
+            ok;
+        {'DOWN', MRef, process, Worker, Reason} ->
+            ?tp(
+                gcp_pubsub_bridge_jwt_worker_failed_to_start,
+                #{
+                    resource_id => InstanceId,
+                    reason => Reason
+                }
+            ),
+            ?SLOG(error, #{
+                msg => "gcp_pubsub_bridge_jwt_worker_failed_to_start",
+                connector => InstanceId,
+                reason => Reason
+            }),
+            _ = emqx_connector_jwt_sup:ensure_worker_deleted(JWTWorkerId),
+            throw(failed_to_start_jwt_worker)
+    after 10_000 ->
+        ?tp(gcp_pubsub_bridge_jwt_timeout, #{resource_id => InstanceId}),
+        ?SLOG(warning, #{
+            msg => "gcp_pubsub_bridge_jwt_timeout",
+            connector => InstanceId
+        }),
+        demonitor(MRef, [flush]),
+        _ = emqx_connector_jwt_sup:ensure_worker_deleted(JWTWorkerId),
+        throw(timeout_creating_jwt)
+    end,
+    #{
+        jwt_worker_id => JWTWorkerId,
+        project_id => ProjectId
+    }.
+
+-spec encode_payload(state(), Selected :: map()) -> #{data := binary()}.
+encode_payload(_State = #{payload_template := PayloadTemplate}, Selected) ->
+    Interpolated =
+        case PayloadTemplate of
+            [] -> emqx_json:encode(Selected);
+            _ -> emqx_plugin_libs_rule:proc_tmpl(PayloadTemplate, Selected)
+        end,
+    #{data => base64:encode(Interpolated)}.
+
+-spec to_pubsub_request([#{data := binary()}]) -> binary().
+to_pubsub_request(Payloads) ->
+    emqx_json:encode(#{messages => Payloads}).
+
+-spec publish_path(state()) -> binary().
+publish_path(
+    _State = #{
+        project_id := ProjectId,
+        pubsub_topic := PubSubTopic
+    }
+) ->
+    <<"/v1/projects/", ProjectId/binary, "/topics/", PubSubTopic/binary, ":publish">>.
+
+-spec get_jwt_authorization_header(resource_id()) -> [{binary(), binary()}].
+get_jwt_authorization_header(InstanceId) ->
+    case emqx_connector_jwt:lookup_jwt(?JWT_TABLE, InstanceId) of
+        %% Since we synchronize the JWT creation during resource start
+        %% (see `on_start/2'), this will be always be populated.
+        {ok, JWT} ->
+            [{<<"Authorization">>, <<"Bearer ", JWT/binary>>}]
+    end.
+
+-spec do_send_requests_sync(
+    state(),
+    [{send_message, map()}],
+    resource_id()
+) ->
+    {ok, status_code(), headers()}
+    | {ok, status_code(), headers(), body()}
+    | {error, {recoverable_error, term()}}
+    | {error, term()}.
+do_send_requests_sync(State, Requests, ResourceId) ->
+    #{
+        pool_name := PoolName,
+        instance_id := InstanceId,
+        max_retries := MaxRetries,
+        request_timeout := RequestTimeout
+    } = State,
+    ?tp(
+        gcp_pubsub_bridge_do_send_requests,
+        #{
+            query_mode => sync,
+            instance_id => InstanceId,
+            resource_id => ResourceId,
+            requests => Requests
+        }
+    ),
+    Headers = get_jwt_authorization_header(InstanceId),
+    Payloads =
+        lists:map(
+            fun({send_message, Selected}) ->
+                encode_payload(State, Selected)
+            end,
+            Requests
+        ),
+    Body = to_pubsub_request(Payloads),
+    Path = publish_path(State),
+    Method = post,
+    Request = {Path, Headers, Body},
+    case
+        ehttpc:request(
+            PoolName,
+            Method,
+            Request,
+            RequestTimeout,
+            MaxRetries
+        )
+    of
+        {error, Reason} when
+            Reason =:= econnrefused;
+            %% this comes directly from `gun'...
+            Reason =:= {closed, "The connection was lost."};
+            Reason =:= timeout
+        ->
+            ?tp(
+                warning,
+                gcp_pubsub_request_failed,
+                #{
+                    reason => Reason,
+                    query_mode => sync,
+                    recoverable_error => true,
+                    connector => ResourceId
+                }
+            ),
+            {error, {recoverable_error, Reason}};
+        {error, Reason} = Result ->
+            ?tp(
+                error,
+                gcp_pubsub_request_failed,
+                #{
+                    reason => Reason,
+                    query_mode => sync,
+                    recoverable_error => false,
+                    connector => ResourceId
+                }
+            ),
+            Result;
+        {ok, StatusCode, _} = Result when StatusCode >= 200 andalso StatusCode < 300 ->
+            ?tp(
+                gcp_pubsub_response,
+                #{
+                    response => Result,
+                    query_mode => sync,
+                    connector => ResourceId
+                }
+            ),
+            Result;
+        {ok, StatusCode, _, _} = Result when StatusCode >= 200 andalso StatusCode < 300 ->
+            ?tp(
+                gcp_pubsub_response,
+                #{
+                    response => Result,
+                    query_mode => sync,
+                    connector => ResourceId
+                }
+            ),
+            Result;
+        {ok, StatusCode, RespHeaders} = _Result ->
+            ?tp(
+                gcp_pubsub_response,
+                #{
+                    response => _Result,
+                    query_mode => sync,
+                    connector => ResourceId
+                }
+            ),
+            ?SLOG(error, #{
+                msg => "gcp_pubsub_error_response",
+                request => Request,
+                connector => ResourceId,
+                status_code => StatusCode
+            }),
+            {error, #{status_code => StatusCode, headers => RespHeaders}};
+        {ok, StatusCode, RespHeaders, RespBody} = _Result ->
+            ?tp(
+                gcp_pubsub_response,
+                #{
+                    response => _Result,
+                    query_mode => sync,
+                    connector => ResourceId
+                }
+            ),
+            ?SLOG(error, #{
+                msg => "gcp_pubsub_error_response",
+                request => Request,
+                connector => ResourceId,
+                status_code => StatusCode
+            }),
+            {error, #{status_code => StatusCode, headers => RespHeaders, body => RespBody}}
+    end.
+
+-spec do_send_requests_async(
+    state(),
+    [{send_message, map()}],
+    {ReplyFun :: function(), Args :: list()},
+    resource_id()
+) -> ok.
+do_send_requests_async(State, Requests, ReplyFunAndArgs, ResourceId) ->
+    #{
+        pool_name := PoolName,
+        instance_id := InstanceId,
+        request_timeout := RequestTimeout
+    } = State,
+    ?tp(
+        gcp_pubsub_bridge_do_send_requests,
+        #{
+            query_mode => async,
+            instance_id => InstanceId,
+            resource_id => ResourceId,
+            requests => Requests
+        }
+    ),
+    Headers = get_jwt_authorization_header(InstanceId),
+    Payloads =
+        lists:map(
+            fun({send_message, Selected}) ->
+                encode_payload(State, Selected)
+            end,
+            Requests
+        ),
+    Body = to_pubsub_request(Payloads),
+    Path = publish_path(State),
+    Method = post,
+    Request = {Path, Headers, Body},
+    Worker = ehttpc_pool:pick_worker(PoolName),
+    ok = ehttpc:request_async(
+        Worker,
+        Method,
+        Request,
+        RequestTimeout,
+        {fun ?MODULE:reply_delegator/3, [ResourceId, ReplyFunAndArgs]}
+    ).
+
+-spec reply_delegator(
+    resource_id(),
+    {ReplyFun :: function(), Args :: list()},
+    term() | {error, econnrefused | timeout | term()}
+) -> ok.
+reply_delegator(_ResourceId, ReplyFunAndArgs, Result) ->
+    case Result of
+        {error, Reason} when
+            Reason =:= econnrefused;
+            %% this comes directly from `gun'...
+            Reason =:= {closed, "The connection was lost."};
+            Reason =:= timeout
+        ->
+            ?tp(
+                gcp_pubsub_request_failed,
+                #{
+                    reason => Reason,
+                    query_mode => async,
+                    recoverable_error => true,
+                    connector => _ResourceId
+                }
+            ),
+            Result1 = {error, {recoverable_error, Reason}},
+            emqx_resource:apply_reply_fun(ReplyFunAndArgs, Result1);
+        _ ->
+            ?tp(
+                gcp_pubsub_response,
+                #{
+                    response => Result,
+                    query_mode => async,
+                    connector => _ResourceId
+                }
+            ),
+            emqx_resource:apply_reply_fun(ReplyFunAndArgs, Result)
+    end.
+
+-spec do_get_status(manager_id(), atom(), timer:time()) -> boolean().
+do_get_status(InstanceId, PoolName, Timeout) ->
+    Workers = [Worker || {_WorkerName, Worker} <- ehttpc:workers(PoolName)],
+    DoPerWorker =
+        fun(Worker) ->
+            case ehttpc:health_check(Worker, Timeout) of
+                ok ->
+                    true;
+                {error, Reason} ->
+                    ?SLOG(error, #{
+                        msg => "ehttpc_health_check_failed",
+                        instance_id => InstanceId,
+                        reason => Reason,
+                        worker => Worker
+                    }),
+                    false
+            end
+        end,
+    try emqx_misc:pmap(DoPerWorker, Workers, Timeout) of
+        [_ | _] = Status ->
+            lists:all(fun(St) -> St =:= true end, Status);
+        [] ->
+            false
+    catch
+        exit:timeout ->
+            false
+    end.