feat(mongo): expose buffer worker opts to the bridge frontend (5.0)

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mongodb.erl

@@ -39,7 +39,7 @@ fields("config") ->
         {enable, mk(boolean(), #{desc => ?DESC("enable"), default => true})},
         {collection, mk(binary(), #{desc => ?DESC("collection"), default => <<"mqtt">>})},
         {payload_template, mk(binary(), #{required => false, desc => ?DESC("payload_template")})}
-    ];
+    ] ++ fields("resource_opts");
 fields(mongodb_rs) ->
     emqx_connector_mongo:fields(rs) ++ fields("config");
 fields(mongodb_sharded) ->
@@ -69,7 +69,32 @@ fields("get_sharded") ->
 fields("get_single") ->
     emqx_bridge_schema:metrics_status_fields() ++
         fields(mongodb_single) ++
-        type_and_name_fields(mongodb_single).
+        type_and_name_fields(mongodb_single);
+fields("creation_opts") ->
+    lists:map(
+        fun
+            ({query_mode, _FieldSchema}) ->
+                {query_mode,
+                    mk(
+                        enum([sync, async]),
+                        #{
+                            desc => ?DESC(emqx_resource_schema, "query_mode"),
+                            default => sync
+                        }
+                    )};
+            (Field) ->
+                Field
+        end,
+        emqx_resource_schema:fields("creation_opts")
+    );
+fields("resource_opts") ->
+    [
+        {resource_opts,
+            mk(
+                ref(?MODULE, "creation_opts"),
+                #{default => #{}, desc => ?DESC(emqx_resource_schema, "resource_opts")}
+            )}
+    ].
 
 conn_bridge_examples(Method) ->
     [
@@ -95,6 +120,8 @@ conn_bridge_examples(Method) ->
 
 desc("config") ->
     ?DESC("desc_config");
+desc("creation_opts") ->
+    ?DESC(emqx_resource_schema, "creation_opts");
 desc(mongodb_rs) ->
     ?DESC(mongodb_rs_conf);
 desc(mongodb_sharded) ->

lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mongodb_SUITE.erl

@@ -152,6 +152,9 @@ mongo_config(MongoHost, MongoPort0, rs = Type) ->
             "  servers = [~p]\n"
             "  w_mode = safe\n"
             "  database = mqtt\n"
+            "  resource_opts = {\n"
+            "    worker_pool_size = 1\n"
+            "  }\n"
             "}",
             [Name, Servers]
         ),
@@ -168,6 +171,9 @@ mongo_config(MongoHost, MongoPort0, sharded = Type) ->
             "  servers = [~p]\n"
             "  w_mode = safe\n"
             "  database = mqtt\n"
+            "  resource_opts = {\n"
+            "    worker_pool_size = 1\n"
+            "  }\n"
             "}",
             [Name, Servers]
         ),
@@ -184,6 +190,9 @@ mongo_config(MongoHost, MongoPort0, single = Type) ->
             "  server = ~p\n"
             "  w_mode = safe\n"
             "  database = mqtt\n"
+            "  resource_opts = {\n"
+            "    worker_pool_size = 1\n"
+            "  }\n"
             "}",
             [Name, Server]
         ),