test(bridges): add test suite for RocketMQ

.ci/docker-compose-file/docker-compose-rocketmq.yaml

@@ -0,0 +1,34 @@
+version: '3.9'
+
+services:
+  mqnamesrv:
+    image: apache/rocketmq:4.9.4
+    container_name: rocketmq_namesrv
+#    ports:
+#      - 9876:9876
+    volumes:
+      - ./rocketmq/logs:/opt/logs
+      - ./rocketmq/store:/opt/store
+    command: ./mqnamesrv 
+    networks:
+      - emqx_bridge
+
+  mqbroker:
+    image: apache/rocketmq:4.9.4 
+    container_name: rocketmq_broker
+#    ports:
+#      - 10909:10909
+#      - 10911:10911
+    volumes:
+      - ./rocketmq/logs:/opt/logs
+      - ./rocketmq/store:/opt/store
+      - ./rocketmq/conf/broker.conf:/etc/rocketmq/broker.conf
+    environment:
+        NAMESRV_ADDR: "rocketmq_namesrv:9876"
+        JAVA_OPTS: " -Duser.home=/opt"
+        JAVA_OPT_EXT: "-server -Xms1024m -Xmx1024m -Xmn1024m"
+    command: ./mqbroker -c /etc/rocketmq/broker.conf
+    depends_on:
+      - mqnamesrv
+    networks:
+      - emqx_bridge

.ci/docker-compose-file/docker-compose-toxiproxy.yaml

@@ -22,6 +22,7 @@ services:
       - 15433:5433
       - 16041:6041
       - 18000:8000
+      - 19876:9876
     command:
       - "-host=0.0.0.0"
       - "-config=/config/toxiproxy.json"

.ci/docker-compose-file/rocketmq/conf/broker.conf

@@ -0,0 +1,22 @@
+brokerClusterName=DefaultCluster
+brokerName=broker-a
+brokerId=0
+
+brokerIP1=rocketmq_broker
+
+defaultTopicQueueNums=4
+autoCreateTopicEnable=true
+autoCreateSubscriptionGroup=true
+
+listenPort=10911
+deleteWhen=04
+
+fileReservedTime=120
+mapedFileSizeCommitLog=1073741824
+mapedFileSizeConsumeQueue=300000
+diskMaxUsedSpaceRatio=100
+maxMessageSize=65536
+
+brokerRole=ASYNC_MASTER
+
+flushDiskType=ASYNC_FLUSH

.ci/docker-compose-file/toxiproxy.json

@@ -77,5 +77,11 @@
     "listen": "0.0.0.0:9295",
     "upstream": "kafka-1.emqx.net:9295",
     "enabled": true
+  },
+  {
+    "name": "rocketmq",
+    "listen": "0.0.0.0:9876",
+    "upstream": "rocketmq_namesrv:9876",
+    "enabled": true
   }
 ]

lib-ee/emqx_ee_bridge/docker-ct

@@ -10,3 +10,4 @@ pgsql
 tdengine
 clickhouse
 dynamo
+rocketmq

lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_rocketmq_SUITE.erl

@@ -0,0 +1,267 @@
+%%--------------------------------------------------------------------
+% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_ee_bridge_rocketmq_SUITE).
+
+-compile(nowarn_export_all).
+-compile(export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+
+% Bridge defaults
+-define(TOPIC, "TopicTest").
+-define(BATCH_SIZE, 10).
+-define(PAYLOAD, <<"HELLO">>).
+
+-define(GET_CONFIG(KEY__, CFG__), proplists:get_value(KEY__, CFG__)).
+
+%%------------------------------------------------------------------------------
+%% CT boilerplate
+%%------------------------------------------------------------------------------
+
+all() ->
+    [
+        {group, with_batch},
+        {group, without_batch}
+    ].
+
+groups() ->
+    TCs = emqx_common_test_helpers:all(?MODULE),
+    [
+        {with_batch, TCs},
+        {without_batch, TCs}
+    ].
+
+init_per_group(with_batch, Config0) ->
+    Config = [{batch_size, ?BATCH_SIZE} | Config0],
+    common_init(Config);
+init_per_group(without_batch, Config0) ->
+    Config = [{batch_size, 1} | Config0],
+    common_init(Config);
+init_per_group(_Group, Config) ->
+    Config.
+
+end_per_group(Group, Config) when Group =:= with_batch; Group =:= without_batch ->
+    ProxyHost = ?config(proxy_host, Config),
+    ProxyPort = ?config(proxy_port, Config),
+    emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
+    ok;
+end_per_group(_Group, _Config) ->
+    ok.
+
+init_per_suite(Config) ->
+    Config.
+
+end_per_suite(_Config) ->
+    emqx_mgmt_api_test_util:end_suite(),
+    ok = emqx_common_test_helpers:stop_apps([emqx_bridge, emqx_conf]),
+    ok.
+
+init_per_testcase(_Testcase, Config) ->
+    delete_bridge(Config),
+    Config.
+
+end_per_testcase(_Testcase, Config) ->
+    ProxyHost = ?config(proxy_host, Config),
+    ProxyPort = ?config(proxy_port, Config),
+    emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
+    ok = snabbkaffe:stop(),
+    delete_bridge(Config),
+    ok.
+
+%%------------------------------------------------------------------------------
+%% Helper fns
+%%------------------------------------------------------------------------------
+
+common_init(ConfigT) ->
+    BridgeType = <<"rocketmq">>,
+    Host = os:getenv("ROCKETMQ_HOST", "toxiproxy"),
+    Port = list_to_integer(os:getenv("ROCKETMQ_PORT", "9876")),
+
+    Config0 = [
+        {host, Host},
+        {port, Port},
+        {query_mode, sync},
+        {proxy_name, "rocketmq"}
+        | ConfigT
+    ],
+
+    case emqx_common_test_helpers:is_tcp_server_available(Host, Port) of
+        true ->
+            % Setup toxiproxy
+            ProxyHost = os:getenv("PROXY_HOST", "toxiproxy"),
+            ProxyPort = list_to_integer(os:getenv("PROXY_PORT", "8474")),
+            emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
+            % Ensure EE bridge module is loaded
+            _ = application:load(emqx_ee_bridge),
+            _ = emqx_ee_bridge:module_info(),
+            ok = emqx_common_test_helpers:start_apps([emqx_conf, emqx_bridge]),
+            emqx_mgmt_api_test_util:init_suite(),
+            {Name, RocketMQConf} = rocketmq_config(BridgeType, Config0),
+            Config =
+                [
+                    {rocketmq_config, RocketMQConf},
+                    {rocketmq_bridge_type, BridgeType},
+                    {rocketmq_name, Name},
+                    {proxy_host, ProxyHost},
+                    {proxy_port, ProxyPort}
+                    | Config0
+                ],
+            Config;
+        false ->
+            case os:getenv("IS_CI") of
+                false ->
+                    {skip, no_rocketmq};
+                _ ->
+                    throw(no_rocketmq)
+            end
+    end.
+
+rocketmq_config(BridgeType, Config) ->
+    Port = integer_to_list(?GET_CONFIG(port, Config)),
+    Server = ?GET_CONFIG(host, Config) ++ ":" ++ Port,
+    Name = atom_to_binary(?MODULE),
+    BatchSize = ?config(batch_size, Config),
+    QueryMode = ?config(query_mode, Config),
+    ConfigString =
+        io_lib:format(
+            "bridges.~s.~s {\n"
+            "  enable = true\n"
+            "  server = ~p\n"
+            "  topic = ~p\n"
+            "  resource_opts = {\n"
+            "    request_timeout = 1500ms\n"
+            "    batch_size = ~b\n"
+            "    query_mode = ~s\n"
+            "  }\n"
+            "}",
+            [
+                BridgeType,
+                Name,
+                Server,
+                ?TOPIC,
+                BatchSize,
+                QueryMode
+            ]
+        ),
+    {Name, parse_and_check(ConfigString, BridgeType, Name)}.
+
+parse_and_check(ConfigString, BridgeType, Name) ->
+    {ok, RawConf} = hocon:binary(ConfigString, #{format => map}),
+    hocon_tconf:check_plain(emqx_bridge_schema, RawConf, #{required => false, atom_key => false}),
+    #{<<"bridges">> := #{BridgeType := #{Name := Config}}} = RawConf,
+    Config.
+
+create_bridge(Config) ->
+    BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config),
+    Name = ?GET_CONFIG(rocketmq_name, Config),
+    RocketMQConf = ?GET_CONFIG(rocketmq_config, Config),
+    emqx_bridge:create(BridgeType, Name, RocketMQConf).
+
+delete_bridge(Config) ->
+    BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config),
+    Name = ?GET_CONFIG(rocketmq_name, Config),
+    emqx_bridge:remove(BridgeType, Name).
+
+create_bridge_http(Params) ->
+    Path = emqx_mgmt_api_test_util:api_path(["bridges"]),
+    AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
+    case emqx_mgmt_api_test_util:request_api(post, Path, "", AuthHeader, Params) of
+        {ok, Res} -> {ok, emqx_json:decode(Res, [return_maps])};
+        Error -> Error
+    end.
+
+send_message(Config, Payload) ->
+    Name = ?GET_CONFIG(rocketmq_name, Config),
+    BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config),
+    BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name),
+    emqx_bridge:send_message(BridgeID, Payload).
+
+query_resource(Config, Request) ->
+    Name = ?GET_CONFIG(rocketmq_name, Config),
+    BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config),
+    ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
+    emqx_resource:query(ResourceID, Request, #{timeout => 500}).
+
+%%------------------------------------------------------------------------------
+%% Testcases
+%%------------------------------------------------------------------------------
+
+t_setup_via_config_and_publish(Config) ->
+    ?assertMatch(
+        {ok, _},
+        create_bridge(Config)
+    ),
+    SentData = #{payload => ?PAYLOAD},
+    ?check_trace(
+        begin
+            ?wait_async_action(
+                ?assertEqual(ok, send_message(Config, SentData)),
+                #{?snk_kind := rocketmq_connector_query_return},
+                10_000
+            ),
+            ok
+        end,
+        fun(Trace0) ->
+            Trace = ?of_kind(rocketmq_connector_query_return, Trace0),
+            ?assertMatch([#{result := ok}], Trace),
+            ok
+        end
+    ),
+    ok.
+
+t_setup_via_http_api_and_publish(Config) ->
+    BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config),
+    Name = ?GET_CONFIG(rocketmq_name, Config),
+    RocketMQConf = ?GET_CONFIG(rocketmq_config, Config),
+    RocketMQConf2 = RocketMQConf#{
+        <<"name">> => Name,
+        <<"type">> => BridgeType
+    },
+    ?assertMatch(
+        {ok, _},
+        create_bridge_http(RocketMQConf2)
+    ),
+    SentData = #{payload => ?PAYLOAD},
+    ?check_trace(
+        begin
+            ?wait_async_action(
+                ?assertEqual(ok, send_message(Config, SentData)),
+                #{?snk_kind := rocketmq_connector_query_return},
+                10_000
+            ),
+            ok
+        end,
+        fun(Trace0) ->
+            Trace = ?of_kind(rocketmq_connector_query_return, Trace0),
+            ?assertMatch([#{result := ok}], Trace),
+            ok
+        end
+    ),
+    ok.
+
+t_get_status(Config) ->
+    ?assertMatch(
+        {ok, _},
+        create_bridge(Config)
+    ),
+
+    Name = ?GET_CONFIG(rocketmq_name, Config),
+    BridgeType = ?GET_CONFIG(rocketmq_bridge_type, Config),
+    ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
+
+    ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceID)),
+    ok.
+
+t_simple_query(Config) ->
+    ?assertMatch(
+        {ok, _},
+        create_bridge(Config)
+    ),
+    Request = {send_message, #{message => <<"Hello">>}},
+    Result = query_resource(Config, Request),
+    ?assertEqual(ok, Result),
+    ok.

lib-ee/emqx_ee_connector/src/emqx_ee_connector_rocketmq.erl

@@ -31,10 +31,6 @@
     default_port => 9876
 }).
 
--ifdef(TEST).
--export([execute/2]).
--endif.
-
 %%=====================================================================
 %% Hocon schema
 roots() ->

scripts/ct/run.sh

@@ -170,6 +170,9 @@ for dep in ${CT_DEPS}; do
         dynamo)
             FILES+=( '.ci/docker-compose-file/docker-compose-dynamo.yaml' )
             ;;
+        rocketmq)
+            FILES+=( '.ci/docker-compose-file/docker-compose-rocketmq.yaml' )
+            ;; 
         *)
             echo "unknown_ct_dependency $dep"
             exit 1