feat: refactor kinesis bridge to connector and action

apps/emqx_bridge/src/emqx_action_info.erl

@@ -89,6 +89,7 @@ hard_coded_action_info_modules_ee() ->
         emqx_bridge_confluent_producer_action_info,
         emqx_bridge_gcp_pubsub_producer_action_info,
         emqx_bridge_kafka_action_info,
+        emqx_bridge_kinesis_action_info,
         emqx_bridge_matrix_action_info,
         emqx_bridge_mongodb_action_info,
         emqx_bridge_influxdb_action_info,

apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis.app.src

@@ -1,13 +1,13 @@
 {application, emqx_bridge_kinesis, [
     {description, "EMQX Enterprise Amazon Kinesis Bridge"},
-    {vsn, "0.1.3"},
+    {vsn, "0.1.4"},
     {registered, []},
     {applications, [
         kernel,
         stdlib,
         erlcloud
     ]},
-    {env, []},
+    {env, [{emqx_action_info_modules, [emqx_bridge_kinesis_action_info]}]},
     {modules, []},
     {links, []}
 ]}.

apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis.erl

@@ -15,7 +15,9 @@
 ]).
 
 -export([
-    conn_bridge_examples/1
+    bridge_v2_examples/1,
+    conn_bridge_examples/1,
+    connector_examples/1
 ]).
 
 %%-------------------------------------------------------------------------------------------------
@@ -28,6 +30,37 @@ namespace() ->
 roots() ->
     [].
 
+fields(Field) when
+    Field == "get_connector";
+    Field == "put_connector";
+    Field == "post_connector"
+->
+    emqx_connector_schema:api_fields(
+        Field,
+        kinesis,
+        connector_config_fields()
+    );
+fields(action) ->
+    {kinesis,
+        hoconsc:mk(
+            hoconsc:map(name, hoconsc:ref(?MODULE, kinesis_action)),
+            #{
+                desc => <<"Kinesis Action Config">>,
+                required => false
+            }
+        )};
+fields(action_parameters) ->
+    fields(producer);
+fields(kinesis_action) ->
+    emqx_bridge_v2_schema:make_producer_action_schema(
+        hoconsc:mk(
+            hoconsc:ref(?MODULE, action_parameters),
+            #{
+                required => true,
+                desc => ?DESC("action_parameters")
+            }
+        )
+    );
 fields("config_producer") ->
     emqx_bridge_schema:common_bridge_fields() ++
         fields("resource_opts") ++
@@ -134,12 +167,38 @@ fields("get_producer") ->
 fields("post_producer") ->
     [type_field_producer(), name_field() | fields("config_producer")];
 fields("put_producer") ->
-    fields("config_producer").
+    fields("config_producer");
+fields("config_connector") ->
+    emqx_connector_schema:common_fields() ++
+        connector_config_fields() ++
+        emqx_connector_schema:resource_opts_ref(?MODULE, connector_resource_opts);
+fields(connector_resource_opts) ->
+    emqx_connector_schema:resource_opts_fields();
+fields("put_bridge_v2") ->
+    fields(kinesis_action);
+fields("get_bridge_v2") ->
+    fields(kinesis_action);
+fields("post_bridge_v2") ->
+    fields("post", kinesis, kinesis_action).
+
+fields("post", Type, StructName) ->
+    [type_field(Type), name_field() | fields(StructName)].
+
+type_field(Type) ->
+    {type, hoconsc:mk(hoconsc:enum([Type]), #{required => true, desc => ?DESC("desc_type")})}.
 
 desc("config_producer") ->
     ?DESC("desc_config");
 desc("creation_opts") ->
     ?DESC(emqx_resource_schema, "creation_opts");
+desc("config_connector") ->
+    ?DESC("config_connector");
+desc(kinesis_action) ->
+    ?DESC("kinesis_action");
+desc(action_parameters) ->
+    ?DESC("action_parameters");
+desc(connector_resource_opts) ->
+    ?DESC(emqx_resource_schema, "resource_opts");
 desc(_) ->
     undefined.
 
@@ -153,6 +212,103 @@ conn_bridge_examples(Method) ->
         }
     ].
 
+connector_examples(Method) ->
+    [
+        #{
+            <<"kinesis">> => #{
+                summary => <<"Kinesis Connector">>,
+                value => values({Method, connector})
+            }
+        }
+    ].
+
+bridge_v2_examples(Method) ->
+    [
+        #{
+            <<"kinesis">> => #{
+                summary => <<"Kinesis Action">>,
+                value => values({Method, bridge_v2_producer})
+            }
+        }
+    ].
+
+values({get, connector}) ->
+    maps:merge(
+        #{
+            status => <<"connected">>,
+            node_status => [
+                #{
+                    node => <<"emqx@localhost">>,
+                    status => <<"connected">>
+                }
+            ],
+            actions => [<<"my_action">>]
+        },
+        values({post, connector})
+    );
+values({get, Type}) ->
+    maps:merge(
+        #{
+            status => <<"connected">>,
+            node_status => [
+                #{
+                    node => <<"emqx@localhost">>,
+                    status => <<"connected">>
+                }
+            ]
+        },
+        values({post, Type})
+    );
+values({post, connector}) ->
+    maps:merge(
+        #{
+            name => <<"my_kinesis_connector">>,
+            type => <<"kinesis">>
+        },
+        values(common_config)
+    );
+values({post, Type}) ->
+    maps:merge(
+        #{
+            name => <<"my_kinesis_action">>,
+            type => <<"kinesis">>
+        },
+        values({put, Type})
+    );
+values({put, bridge_v2_producer}) ->
+    values(bridge_v2_producer);
+values({put, connector}) ->
+    values(common_config);
+values({put, Type}) ->
+    maps:merge(values(common_config), values(Type));
+values(bridge_v2_producer) ->
+    #{
+        enable => true,
+        connector => <<"my_kinesis_connector">>,
+        parameters => values(producer_values),
+        resource_opts => #{
+            <<"batch_size">> => 100,
+            <<"inflight_window">> => 100,
+            <<"max_buffer_bytes">> => <<"256MB">>,
+            <<"request_ttl">> => <<"45s">>
+        }
+    };
+values(common_config) ->
+    #{
+        <<"enable">> => true,
+        <<"aws_access_key_id">> => <<"your_access_key">>,
+        <<"aws_secret_access_key">> => <<"aws_secret_key">>,
+        <<"endpoint">> => <<"http://localhost:4566">>,
+        <<"max_retries">> => 2,
+        <<"pool_size">> => 8
+    };
+values(producer_values) ->
+    #{
+        <<"partition_key">> => <<"any_key">>,
+        <<"payload_template">> => <<"${.}">>,
+        <<"stream_name">> => <<"my_stream">>
+    }.
+
 values(producer, _Method) ->
     #{
         aws_access_key_id => <<"aws_access_key_id">>,
@@ -174,6 +330,9 @@ values(producer, _Method) ->
 %% Helper fns
 %%-------------------------------------------------------------------------------------------------
 
+connector_config_fields() ->
+    fields(connector_config).
+
 sc(Type, Meta) -> hoconsc:mk(Type, Meta).
 
 mk(Type, Meta) -> hoconsc:mk(Type, Meta).

apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis_action_info.erl

@@ -0,0 +1,22 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_bridge_kinesis_action_info).
+
+-behaviour(emqx_action_info).
+
+-export([
+    bridge_v1_type_name/0,
+    action_type_name/0,
+    connector_type_name/0,
+    schema_module/0
+]).
+
+bridge_v1_type_name() -> kinesis_producer.
+
+action_type_name() -> kinesis.
+
+connector_type_name() -> kinesis.
+
+schema_module() -> emqx_bridge_kinesis.

apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis_connector_client.erl

@@ -11,9 +11,7 @@
 -behaviour(gen_server).
 
 -type state() :: #{
-    instance_id := resource_id(),
-    partition_key := binary(),
-    stream_name := binary()
+    instance_id := resource_id()
 }.
 -type record() :: {Data :: binary(), PartitionKey :: binary()}.
 
@@ -23,7 +21,8 @@
 -export([
     start_link/1,
     connection_status/1,
-    query/2
+    connection_status/2,
+    query/3
 ]).
 
 %% gen_server callbacks
@@ -56,8 +55,16 @@ connection_status(Pid) ->
             {error, timeout}
     end.
 
-query(Pid, Records) ->
-    gen_server:call(Pid, {query, Records}, infinity).
+connection_status(Pid, StreamName) ->
+    try
+        gen_server:call(Pid, {connection_status, StreamName}, ?HEALTH_CHECK_TIMEOUT)
+    catch
+        _:_ ->
+            {error, timeout}
+    end.
+
+query(Pid, Records, StreamName) ->
+    gen_server:call(Pid, {query, Records, StreamName}, infinity).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -72,13 +79,12 @@ start_link(Options) ->
 %%%===================================================================
 
 %% Initialize kinesis connector
--spec init(emqx_bridge_kinesis_impl_producer:config()) -> {ok, state()}.
+-spec init(emqx_bridge_kinesis_impl_producer:config_connector()) ->
+    {ok, state()} | {stop, Reason :: term()}.
 init(#{
     aws_access_key_id := AwsAccessKey,
     aws_secret_access_key := AwsSecretAccessKey,
     endpoint := Endpoint,
-    partition_key := PartitionKey,
-    stream_name := StreamName,
     max_retries := MaxRetries,
     instance_id := InstanceId
 }) ->
@@ -93,9 +99,7 @@ init(#{
             }
         ),
     State = #{
-        instance_id => InstanceId,
-        partition_key => PartitionKey,
-        stream_name => StreamName
+        instance_id => InstanceId
     },
     %% TODO: teach `erlcloud` to to accept 0-arity closures as passwords.
     ok = erlcloud_config:configure(
@@ -124,18 +128,19 @@ init(#{
             {stop, Reason}
     end.
 
-handle_call(connection_status, _From, #{stream_name := StreamName} = State) ->
+handle_call({connection_status, StreamName}, _From, State) ->
+    Status = get_status(StreamName),
+    {reply, Status, State};
+handle_call(connection_status, _From, State) ->
     Status =
-        case erlcloud_kinesis:describe_stream(StreamName) of
-            {ok, _} ->
+        case erlcloud_kinesis:list_streams() of
+            {ok, _ListStreamsResult} ->
                 {ok, connected};
-            {error, {<<"ResourceNotFoundException">>, _}} ->
-                {error, unhealthy_target};
             Error ->
                 {error, Error}
         end,
     {reply, Status, State};
-handle_call({query, Records}, _From, #{stream_name := StreamName} = State) ->
+handle_call({query, Records, StreamName}, _From, State) ->
     Result = do_query(StreamName, Records),
     {reply, Result, State};
 handle_call(_Request, _From, State) ->
@@ -158,6 +163,16 @@ code_change(_OldVsn, State, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
+get_status(StreamName) ->
+    case erlcloud_kinesis:describe_stream(StreamName) of
+        {ok, _} ->
+            {ok, connected};
+        {error, {<<"ResourceNotFoundException">>, _}} ->
+            {error, unhealthy_target};
+        Error ->
+            {error, Error}
+    end.
+
 -spec do_query(binary(), [record()]) ->
     {ok, jsx:json_term() | binary()}
     | {error, {unrecoverable_error, term()}}