Merge pull request #8559 from DDDHuang/ee5_hstream

feat: hstream bridge & connector

apps/emqx_bridge/src/emqx_bridge.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_bridge, [
     {description, "An OTP application"},
-    {vsn, "0.1.0"},
+    {vsn, "0.1.1"},
     {registered, []},
     {mod, {emqx_bridge_app, []}},
     {applications, [

apps/emqx_bridge/src/emqx_bridge_api.erl

@@ -163,7 +163,7 @@ bridge_info_examples(Method) ->
     }).
 
 conn_bridge_examples(Method) ->
-    lists:foldl(
+    Fun =
         fun(Type, Acc) ->
             SType = atom_to_list(Type),
             KeyIngress = bin(SType ++ "_ingress"),
@@ -179,9 +179,17 @@ conn_bridge_examples(Method) ->
                 }
             })
         end,
-        #{},
-        ?CONN_TYPES
-    ).
+    Broker = lists:foldl(Fun, #{}, ?CONN_TYPES),
+    EE = ee_conn_bridge_examples(Method),
+    maps:merge(Broker, EE).
+
+-if(?EMQX_RELEASE_EDITION == ee).
+ee_conn_bridge_examples(Method) ->
+    emqx_ee_bridge:conn_bridge_examples(Method).
+-else.
+ee_conn_bridge_examples(_Method) ->
+    #{}.
+-endif.
 
 info_example(Type, Direction, Method) ->
     maps:merge(

apps/emqx_bridge/src/emqx_bridge_app.erl

@@ -29,6 +29,7 @@
 
 start(_StartType, _StartArgs) ->
     {ok, Sup} = emqx_bridge_sup:start_link(),
+    ok = start_ee_apps(),
     ok = emqx_bridge:load(),
     ok = emqx_bridge:load_hook(),
     ok = emqx_config_handler:add_handler(?LEAF_NODE_HDLR_PATH, ?MODULE),
@@ -41,6 +42,16 @@ stop(_State) ->
     ok = emqx_bridge:unload_hook(),
     ok.
 
+-if(?EMQX_RELEASE_EDITION == ee).
+start_ee_apps() ->
+    {ok, _} = application:ensure_all_started(emqx_ee_bridge),
+    {ok, _} = application:ensure_all_started(emqx_ee_connector),
+    ok.
+-else.
+start_ee_apps() ->
+    ok.
+-endif.
+
 %% NOTE: We depends on the `emqx_bridge:pre_config_update/3` to restart/stop the
 %%       underlying resources.
 pre_config_update(_, {_Oper, _, _}, undefined) ->

apps/emqx_bridge/src/emqx_bridge_resource.erl

@@ -42,10 +42,18 @@
     reset_metrics/1
 ]).
 
+-if(?EMQX_RELEASE_EDITION == ee).
+bridge_to_resource_type(<<"mqtt">>) -> emqx_connector_mqtt;
+bridge_to_resource_type(mqtt) -> emqx_connector_mqtt;
+bridge_to_resource_type(<<"webhook">>) -> emqx_connector_http;
+bridge_to_resource_type(webhook) -> emqx_connector_http;
+bridge_to_resource_type(BridgeType) -> emqx_ee_bridge:resource_type(BridgeType).
+-else.
 bridge_to_resource_type(<<"mqtt">>) -> emqx_connector_mqtt;
 bridge_to_resource_type(mqtt) -> emqx_connector_mqtt;
 bridge_to_resource_type(<<"webhook">>) -> emqx_connector_http;
 bridge_to_resource_type(webhook) -> emqx_connector_http.
+-endif.
 
 resource_id(BridgeId) when is_binary(BridgeId) ->
     <<"bridge:", BridgeId/binary>>.
@@ -231,7 +239,7 @@ is_tmp_path(TmpPath, File) ->
     string:str(str(File), str(TmpPath)) > 0.
 
 parse_confs(
-    webhook,
+    Type,
     _Name,
     #{
         url := Url,
@@ -240,7 +248,7 @@ parse_confs(
         headers := Headers,
         request_timeout := ReqTimeout
     } = Conf
-) ->
+) when Type == webhook orelse Type == <<"webhook">> ->
     {BaseUrl, Path} = parse_url(Url),
     {ok, BaseUrl2} = emqx_http_lib:uri_parse(BaseUrl),
     Conf#{
@@ -279,7 +287,9 @@ parse_confs(Type, Name, #{connector := ConnectorConfs, direction := Direction} =
         maps:without([connector, direction], Conf),
         Type,
         Name
-    ).
+    );
+parse_confs(_Type, _Name, Conf) ->
+    Conf.
 
 make_resource_confs(ingress, ConnectorConfs, BridgeConf, Type, Name) ->
     BName = bridge_id(Type, Name),

apps/emqx_bridge/src/emqx_bridge_schema.erl

@@ -45,10 +45,16 @@ http_schema(Method) ->
         end,
         ?CONN_TYPES
     ),
-    hoconsc:union([
-        ref(emqx_bridge_webhook_schema, Method)
-        | Schemas
-    ]).
+    ExtSchemas = [ref(Module, Method) || Module <- schema_modules()],
+    hoconsc:union(Schemas ++ ExtSchemas).
+
+-if(?EMQX_RELEASE_EDITION == ee).
+schema_modules() ->
+    [emqx_bridge_webhook_schema] ++ emqx_ee_bridge:schema_modules().
+-else.
+schema_modules() ->
+    [emqx_bridge_webhook_schema].
+-endif.
 
 common_bridge_fields(ConnectorRef) ->
     [
@@ -127,7 +133,7 @@ fields(bridges) ->
                     #{desc => ?DESC("bridges_name")}
                 )}
          || T <- ?CONN_TYPES
-        ];
+        ] ++ ee_fields_bridges();
 fields("metrics") ->
     [
         {"matched", mk(integer(), #{desc => ?DESC("metric_matched")})},
@@ -152,6 +158,14 @@ fields("node_status") ->
         {"status", mk(status(), #{})}
     ].
 
+-if(?EMQX_RELEASE_EDITION == ee).
+ee_fields_bridges() ->
+    emqx_ee_bridge:fields(bridges).
+-else.
+ee_fields_bridges() ->
+    [].
+-endif.
+
 desc(bridges) ->
     ?DESC("desc_bridges");
 desc("metrics") ->

apps/emqx_resource/src/emqx_resource_manager.erl

@@ -214,7 +214,7 @@ get_metrics(ResId) ->
 reset_metrics(ResId) ->
     emqx_metrics_worker:reset_metrics(resource_metrics, ResId).
 
-%% @doc Returns the data for all resorces
+%% @doc Returns the data for all resources
 -spec list_all() -> [resource_data()] | [].
 list_all() ->
     try

lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_hstream.conf

@@ -0,0 +1,84 @@
+emqx_ee_bridge_hstream {
+    local_topic {
+        desc {
+            en: """
+The MQTT topic filter to be forwarded to the HStreamDB. All MQTT 'PUBLISH' messages with the topic
+matching the local_topic will be forwarded.</br>
+NOTE: if this bridge is used as the action of a rule (EMQX rule engine), and also local_topic is
+configured, then both the data got from the rule and the MQTT messages that match local_topic
+will be forwarded.
+"""
+            zh: """
+发送到 'local_topic' 的消息都会转发到 HStreamDB。 </br>
+注意：如果这个 Bridge 被用作规则（EMQX 规则引擎）的输出，同时也配置了 'local_topic' ，那么这两部分的消息都会被转发到 HStreamDB。
+"""
+        }
+    label {
+            en: "Local Topic"
+            zh: "本地 Topic"
+        }
+    }
+    payload {
+        desc {
+            en: """The payload to be forwarded to the HStreamDB. Placeholders supported."""
+            zh: """要转发到 HStreamDB 的数据内容，支持占位符"""
+            }
+        label {
+            en: "Payload"
+            zh: "消息内容"
+        }
+    }
+    config_enable {
+        desc {
+            en: """Enable or disable this bridge"""
+            zh: """启用/禁用 Bridge"""
+        }
+        label {
+            en: "Enable Or Disable Bridge"
+            zh: "启用/禁用 Bridge"
+        }
+        }
+    config_direction {
+        desc {
+            en: """The direction of this bridge, MUST be 'egress'"""
+            zh: """Bridge 的方向， 必须是 egress"""
+        }
+        label {
+            en: "Bridge Direction"
+            zh: "Bridge 方向"
+        }
+    }
+
+    desc_config {
+        desc {
+            en: """Configuration for an HStreamDB bridge."""
+            zh: """HStreamDB Bridge 配置"""
+        }
+        label: {
+            en: "HStreamDB Bridge Configuration"
+            zh: "HStreamDB Bridge 配置"
+        }
+    }
+
+    desc_type {
+        desc {
+            en: """The Bridge Type"""
+            zh: """Bridge 类型"""
+        }
+        label {
+            en: "Bridge Type"
+            zh: "Bridge 类型"
+        }
+    }
+
+    desc_name {
+        desc {
+            en: """Bridge name, used as a human-readable description of the bridge."""
+            zh: """Bridge 名字，Bridge 的可读描述"""
+        }
+        label {
+            en: "Bridge Name"
+            zh: "Bridge 名字"
+        }
+    }
+}

lib-ee/emqx_ee_bridge/include/emqx_ee_bridge.hrl

@@ -0,0 +1,18 @@
+-define(METRICS(MATCH, SUCC, FAILED, RATE, RATE_5, RATE_MAX), #{
+    matched => MATCH,
+    success => SUCC,
+    failed => FAILED,
+    rate => RATE,
+    rate_last5m => RATE_5,
+    rate_max => RATE_MAX
+}).
+
+-define(METRICS_EXAMPLE, #{
+    metrics => ?METRICS(0, 0, 0, 0, 0, 0),
+    node_metrics => [
+        #{
+            node => node(),
+            metrics => ?METRICS(0, 0, 0, 0, 0, 0)
+        }
+    ]
+}).

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.app.src

@@ -1,8 +1,6 @@
 {application, emqx_ee_bridge, [
-    {description, "An OTP application"},
     {vsn, "0.1.0"},
     {registered, []},
-    {mod, {emqx_ee_bridge_app, []}},
     {applications, [
         kernel,
         stdlib

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl

@@ -0,0 +1,36 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_ee_bridge).
+
+-import(hoconsc, [mk/2, enum/1, ref/2]).
+
+-export([
+    schema_modules/0,
+    conn_bridge_examples/1,
+    resource_type/1,
+    fields/1
+]).
+
+schema_modules() ->
+    [emqx_ee_bridge_hstream].
+
+conn_bridge_examples(Method) ->
+    Fun =
+        fun(Module, Examples) ->
+            Example = erlang:apply(Module, conn_bridge_example, [Method]),
+            maps:merge(Examples, Example)
+        end,
+    lists:foldl(Fun, #{}, schema_modules()).
+
+resource_type(hstream) -> emqx_ee_connector_hstream;
+resource_type(<<"hstream">>) -> emqx_ee_connector_hstream.
+
+fields(bridges) ->
+    [
+        {hstream,
+            mk(
+                hoconsc:map(name, ref(emqx_ee_bridge_hstream, "config")),
+                #{desc => <<"EMQX Enterprise Config">>}
+            )}
+    ].

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_app.erl

@@ -1,17 +0,0 @@
-%%--------------------------------------------------------------------
-%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
-%%--------------------------------------------------------------------
-
--module(emqx_ee_bridge_app).
-
--behaviour(application).
-
--export([start/2, stop/1]).
-
-start(_StartType, _StartArgs) ->
-    emqx_ee_bridge_sup:start_link().
-
-stop(_State) ->
-    ok.
-
-%% internal functions

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_hstream.erl

@@ -0,0 +1,91 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_ee_bridge_hstream).
+
+-include_lib("typerefl/include/types.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+-include("emqx_ee_bridge.hrl").
+
+-import(hoconsc, [mk/2, enum/1]).
+
+-export([
+    conn_bridge_example/1
+]).
+
+-export([
+    namespace/0,
+    roots/0,
+    fields/1,
+    desc/1
+]).
+
+%% -------------------------------------------------------------------------------------------------
+%% api
+
+conn_bridge_example(Method) ->
+    #{
+        <<"hstream">> => #{
+            summary => <<"HStreamDB Bridge">>,
+            value => values(Method)
+        }
+    }.
+
+values(get) ->
+    maps:merge(values(post), ?METRICS_EXAMPLE);
+values(post) ->
+    #{
+        type => hstream,
+        name => <<"demo">>,
+        url => <<"http://127.0.0.1:6570">>,
+        stream => <<"stream1">>,
+        ordering_key => <<"some_key">>,
+        pool_size => 8,
+        enable => true,
+        direction => egress,
+        local_topic => <<"local/topic/#">>,
+        payload => <<"${payload}">>
+    };
+values(put) ->
+    values(post).
+
+%% -------------------------------------------------------------------------------------------------
+%% Hocon Schema Definitions
+namespace() -> "bridge".
+
+roots() -> [].
+
+fields("config") ->
+    ExtConfig = [
+        {local_topic, mk(binary(), #{desc => ?DESC("local_topic")})},
+        {payload, mk(binary(), #{default => <<"${payload}">>, desc => ?DESC("payload")})}
+    ],
+    basic_config() ++ ExtConfig;
+fields("post") ->
+    [type_field(), name_field() | fields("config")];
+fields("put") ->
+    fields("config");
+fields("get") ->
+    emqx_bridge_schema:metrics_status_fields() ++ fields("post").
+
+desc("config") ->
+    ?DESC("desc_config");
+desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
+    ["Configuration for HStream using `", string:to_upper(Method), "` method."];
+desc(_) ->
+    undefined.
+
+basic_config() ->
+    Basic = [
+        {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
+        {direction, mk(egress, #{desc => ?DESC("config_direction"), default => egress})}
+    ],
+    emqx_ee_connector_hstream:fields(config) ++ Basic.
+
+%% -------------------------------------------------------------------------------------------------
+%% internal
+type_field() ->
+    {type, mk(enum([hstream]), #{required => true, desc => ?DESC("desc_type")})}.
+
+name_field() ->
+    {name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}.

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_sup.erl

@@ -1,27 +0,0 @@
-%%--------------------------------------------------------------------
-%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
-%%--------------------------------------------------------------------
-
--module(emqx_ee_bridge_sup).
-
--behaviour(supervisor).
-
--export([start_link/0]).
-
--export([init/1]).
-
--define(SERVER, ?MODULE).
-
-start_link() ->
-    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
-
-init([]) ->
-    SupFlags = #{
-        strategy => one_for_all,
-        intensity => 0,
-        period => 1
-    },
-    ChildSpecs = [],
-    {ok, {SupFlags, ChildSpecs}}.
-
-%% internal functions

lib-ee/emqx_ee_bridge/test/ee_bridge_hstream_SUITE.erl

@@ -0,0 +1,16 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(ee_bridge_hstream_SUITE).
+
+-compile(nowarn_export_all).
+-compile(export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+%% TODO:

lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_hstream.conf

@@ -0,0 +1,43 @@
+
+emqx_ee_connector_hstream {
+    url {
+        desc {
+                en: """HStream Server URL"""
+                zh: """HStream 服务器 URL"""
+            }
+        label {
+                en: """HStream Server URL"""
+                zh: """HStream 服务器 URL"""
+                }
+        }
+    stream_name {
+        desc {
+                en: """HStream Stream Name"""
+                zh: """HStream 流名称"""
+            }
+        label {
+                en: """HStream Stream Name"""
+                zh: """HStream 流名称"""
+                }
+    }
+    ordering_key {
+        desc {
+                en: """HStream Ordering Key"""
+                zh: """HStream 分区键"""
+            }
+        label {
+                en: """HStream Ordering Key"""
+                zh: """HStream 分区键"""
+                }
+    }
+    pool_size {
+        desc {
+                en: """HStream Pool Size"""
+                zh: """HStream 连接池大小"""
+            }
+        label {
+                en: """HStream Pool Size"""
+                zh: """HStream 连接池大小"""
+            }
+    }
+}

lib-ee/emqx_ee_connector/rebar.config

@@ -1,5 +1,7 @@
 {erl_opts, [debug_info]}.
-{deps, []}.
+{deps, [
+  {hstreamdb_erl, {git, "https://github.com/hstreamdb/hstreamdb_erl.git", {tag, "0.2.5"}}}
+]}.
 
 {shell, [
     {apps, [emqx_ee_connector]}

lib-ee/emqx_ee_connector/src/emqx_ee_connector.app.src

@@ -1,11 +1,10 @@
 {application, emqx_ee_connector, [
-    {description, "An OTP application"},
     {vsn, "0.1.0"},
     {registered, []},
-    {mod, {emqx_ee_connector_app, []}},
     {applications, [
         kernel,
-        stdlib
+        stdlib,
+        hstreamdb_erl
     ]},
     {env, []},
     {modules, []},

lib-ee/emqx_ee_connector/src/emqx_ee_connector_app.erl

@@ -1,17 +0,0 @@
-%%--------------------------------------------------------------------
-%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
-%%--------------------------------------------------------------------
-
--module(emqx_ee_connector_app).
-
--behaviour(application).
-
--export([start/2, stop/1]).
-
-start(_StartType, _StartArgs) ->
-    emqx_ee_connector_sup:start_link().
-
-stop(_State) ->
-    ok.
-
-%% internal functions

lib-ee/emqx_ee_connector/src/emqx_ee_connector_hstream.erl

@@ -0,0 +1,265 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_ee_connector_hstream).
+
+-include_lib("hocon/include/hoconsc.hrl").
+-include_lib("typerefl/include/types.hrl").
+-include_lib("emqx/include/logger.hrl").
+
+-import(hoconsc, [mk/2, enum/1]).
+
+-behaviour(emqx_resource).
+
+%% callbacks of behaviour emqx_resource
+-export([
+    on_start/2,
+    on_stop/2,
+    on_query/4,
+    on_get_status/2
+]).
+
+-export([
+    on_flush_result/1
+]).
+
+-export([
+    roots/0,
+    fields/1
+]).
+
+%% -------------------------------------------------------------------------------------------------
+%% resource callback
+
+on_start(InstId, Config) ->
+    start_client(InstId, Config).
+
+on_stop(InstId, #{client := Client, producer := Producer}) ->
+    StopClientRes = hstreamdb:stop_client(Client),
+    StopProducerRes = hstreamdb:stop_producer(Producer),
+    ?SLOG(info, #{
+        msg => "stop hstream connector",
+        connector => InstId,
+        client => Client,
+        producer => Producer,
+        stop_client => StopClientRes,
+        stop_producer => StopProducerRes
+    }).
+
+on_query(
+    _InstId,
+    {send_message, Data},
+    AfterQuery,
+    #{producer := Producer, ordering_key := OrderingKey, payload := Payload}
+) ->
+    Record = to_record(OrderingKey, Payload, Data),
+    do_append(AfterQuery, Producer, Record).
+
+on_get_status(_InstId, #{client := Client}) ->
+    case is_alive(Client) of
+        true ->
+            connected;
+        false ->
+            disconnected
+    end.
+
+%% -------------------------------------------------------------------------------------------------
+%% hstream batch callback
+%% TODO: maybe remove it after disk cache is ready
+
+on_flush_result({{flush, _Stream, _Records}, {ok, _Resp}}) ->
+    ok;
+on_flush_result({{flush, _Stream, _Records}, {error, _Reason}}) ->
+    ok.
+
+%% -------------------------------------------------------------------------------------------------
+%% schema
+
+roots() ->
+    fields(config).
+
+fields(config) ->
+    [
+        {url, mk(binary(), #{required => true, desc => ?DESC("url")})},
+        {stream, mk(binary(), #{required => true, desc => ?DESC("stream_name")})},
+        {ordering_key, mk(binary(), #{required => true, desc => ?DESC("ordering_key")})},
+        {pool_size, mk(pos_integer(), #{required => true, desc => ?DESC("pool_size")})}
+    ].
+
+%% -------------------------------------------------------------------------------------------------
+%% internal functions
+start_client(InstId, Config) ->
+    try
+        do_start_client(InstId, Config)
+    catch
+        E:R:S ->
+            ?SLOG(error, #{
+                msg => "start hstream connector error",
+                connector => InstId,
+                error => E,
+                reason => R,
+                stack => S
+            })
+    end.
+
+do_start_client(InstId, Config = #{url := Server, pool_size := PoolSize}) ->
+    ?SLOG(info, #{
+        msg => "starting hstream connector: client",
+        connector => InstId,
+        config => Config
+    }),
+    ClientName = client_name(InstId),
+    ClientOptions = [
+        {url, binary_to_list(Server)},
+        {rpc_options, #{pool_size => PoolSize}}
+    ],
+    case hstreamdb:start_client(ClientName, ClientOptions) of
+        {ok, Client} ->
+            case is_alive(Client) of
+                true ->
+                    ?SLOG(info, #{
+                        msg => "hstream connector: client started",
+                        connector => InstId,
+                        client => Client
+                    }),
+                    start_producer(InstId, Client, Config);
+                _ ->
+                    ?SLOG(error, #{
+                        msg => "hstream connector: client not alive",
+                        connector => InstId
+                    }),
+                    {error, connect_failed}
+            end;
+        {error, {already_started, Pid}} ->
+            ?SLOG(info, #{
+                msg => "starting hstream connector: client, find old client. restart client",
+                old_client_pid => Pid,
+                old_client_name => ClientName
+            }),
+            _ = hstreamdb:stop_client(ClientName),
+            start_client(InstId, Config);
+        {error, Error} ->
+            ?SLOG(error, #{
+                msg => "hstream connector: client failed",
+                connector => InstId,
+                reason => Error
+            }),
+            {error, Error}
+    end.
+
+is_alive(Client) ->
+    case hstreamdb:echo(Client) of
+        {ok, _Echo} ->
+            true;
+        _ErrorEcho ->
+            false
+    end.
+
+start_producer(InstId, Client, Options = #{stream := Stream, pool_size := PoolSize}) ->
+    %% TODO: change these batch options after we have better disk cache.
+    BatchSize = maps:get(batch_size, Options, 100),
+    Interval = maps:get(batch_interval, Options, 1000),
+    ProducerOptions = [
+        {stream, Stream},
+        {callback, {?MODULE, on_flush_result, []}},
+        {max_records, BatchSize},
+        {interval, Interval},
+        {pool_size, PoolSize}
+    ],
+    Name = produce_name(InstId),
+    ?SLOG(info, #{
+        msg => "starting hstream connector: producer",
+        connector => InstId
+    }),
+    case hstreamdb:start_producer(Client, Name, ProducerOptions) of
+        {ok, Producer} ->
+            ?SLOG(info, #{
+                msg => "hstream connector: producer started"
+            }),
+            EnableBatch = maps:get(enable_batch, Options, false),
+            PayloadBin = maps:get(payload, Options, <<"">>),
+            Payload = emqx_plugin_libs_rule:preproc_tmpl(PayloadBin),
+            OrderingKeyBin = maps:get(ordering_key, Options, <<"">>),
+            OrderingKey = emqx_plugin_libs_rule:preproc_tmpl(OrderingKeyBin),
+            State = #{
+                client => Client,
+                producer => Producer,
+                enable_batch => EnableBatch,
+                ordering_key => OrderingKey,
+                payload => Payload
+            },
+            {ok, State};
+        {error, {already_started, Pid}} ->
+            ?SLOG(info, #{
+                msg => "starting hstream connector: producer, find old producer. restart producer",
+                old_producer_pid => Pid,
+                old_producer_name => Name
+            }),
+            _ = hstreamdb:stop_producer(Name),
+            start_producer(InstId, Client, Options);
+        {error, Reason} ->
+            ?SLOG(error, #{
+                msg => "starting hstream connector: producer, failed",
+                reason => Reason
+            }),
+            {error, Reason}
+    end.
+
+to_record(OrderingKeyTmpl, PayloadTmpl, Data) ->
+    OrderingKey = emqx_plugin_libs_rule:proc_tmpl(OrderingKeyTmpl, Data),
+    Payload = emqx_plugin_libs_rule:proc_tmpl(PayloadTmpl, Data),
+    to_record(OrderingKey, Payload).
+
+to_record(OrderingKey, Payload) when is_binary(OrderingKey) ->
+    to_record(binary_to_list(OrderingKey), Payload);
+to_record(OrderingKey, Payload) ->
+    hstreamdb:to_record(OrderingKey, raw, Payload).
+
+do_append(AfterQuery, Producer, Record) ->
+    do_append(AfterQuery, false, Producer, Record).
+
+%% TODO: this append is async, remove or change it after we have better disk cache.
+% do_append(AfterQuery, true, Producer, Record) ->
+%     case hstreamdb:append(Producer, Record) of
+%         ok ->
+%             ?SLOG(debug, #{
+%                 msg => "hstream producer async append success",
+%                 record => Record
+%             }),
+%             emqx_resource:query_success(AfterQuery);
+%         {error, Reason} ->
+%             ?SLOG(error, #{
+%                 msg => "hstream producer async append failed",
+%                 reason => Reason,
+%                 record => Record
+%             }),
+%             emqx_resource:query_failed(AfterQuery)
+%     end;
+do_append(AfterQuery, false, Producer, Record) ->
+    %% TODO: this append is sync, but it does not support [Record], can only append one Record.
+    %% Change it after we have better dick cache.
+    case hstreamdb:append_flush(Producer, Record) of
+        {ok, _} ->
+            ?SLOG(debug, #{
+                msg => "hstream producer sync append success",
+                record => Record
+            }),
+            emqx_resource:query_success(AfterQuery);
+        {error, Reason} ->
+            ?SLOG(error, #{
+                msg => "hstream producer sync append failed",
+                reason => Reason,
+                record => Record
+            }),
+            emqx_resource:query_failed(AfterQuery)
+    end.
+
+client_name(InstId) ->
+    "client:" ++ to_string(InstId).
+
+produce_name(ActionId) ->
+    list_to_atom("producer:" ++ to_string(ActionId)).
+
+to_string(List) when is_list(List) -> List;
+to_string(Bin) when is_binary(Bin) -> binary_to_list(Bin);
+to_string(Atom) when is_atom(Atom) -> atom_to_list(Atom).

lib-ee/emqx_ee_connector/src/emqx_ee_connector_sup.erl

@@ -1,27 +0,0 @@
-%%--------------------------------------------------------------------
-%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
-%%--------------------------------------------------------------------
-
--module(emqx_ee_connector_sup).
-
--behaviour(supervisor).
-
--export([start_link/0]).
-
--export([init/1]).
-
--define(SERVER, ?MODULE).
-
-start_link() ->
-    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
-
-init([]) ->
-    SupFlags = #{
-        strategy => one_for_all,
-        intensity => 0,
-        period => 1
-    },
-    ChildSpecs = [],
-    {ok, {SupFlags, ChildSpecs}}.
-
-%% internal functions

lib-ee/emqx_ee_connector/test/ee_connector_hstream_SUITE.erl

@@ -0,0 +1,16 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(ee_connector_hstream_SUITE).
+
+-compile(nowarn_export_all).
+-compile(export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+%% TODO:

mix.exs

@@ -88,7 +88,8 @@ defmodule EMQXUmbrella.MixProject do
       {:ranch,
        github: "ninenines/ranch", ref: "a692f44567034dacf5efcaa24a24183788594eb7", override: true},
       # in conflict by grpc and eetcd
-      {:gpb, "4.11.2", override: true, runtime: false}
+      {:gpb, "4.11.2", override: true, runtime: false},
+      {:hstreamdb_erl, github: "hstreamdb/hstreamdb_erl", tag: "0.2.5"}
     ] ++ umbrella_apps() ++ bcrypt_dep() ++ jq_dep() ++ quicer_dep()
   end
 
@@ -215,7 +216,9 @@ defmodule EMQXUmbrella.MixProject do
       if(edition_type == :enterprise,
         do: [
           emqx_license: :permanent,
-          emqx_enterprise_conf: :load
+          emqx_enterprise_conf: :load,
+          emqx_ee_connector: :permanent,
+          emqx_ee_bridge: :permanent
         ],
         else: []
       )

scripts/merge-i18n.escript

@@ -4,7 +4,7 @@
 
 main(_) ->
     BaseConf = <<"">>,
-    Cfgs = get_all_cfgs("apps/"),
+    Cfgs = get_all_cfgs("apps/") ++ get_all_cfgs("lib-ee/"),
     Conf = [merge(BaseConf, Cfgs),
             io_lib:nl()
             ],