fix: add query_mode_sync_only for mysql pgsql redis mongodb bridge

This commit is contained in:
Zhongwen Deng 2023-02-02 16:52:52 +08:00
parent dcc6bd9c21
commit 22c3f50020
6 changed files with 46 additions and 47 deletions

View File

@ -89,6 +89,17 @@ For bridges only have ingress direction data flow, it can be set to 0 otherwise
}
}
query_mode_sync_only {
desc {
en: """Query mode. Currently only support 'sync'."""
zh: """请求模式。目前只支持同步模式。"""
}
label {
en: """Query mode"""
zh: """请求模式"""
}
}
request_timeout {
desc {
en: """Timeout for requests. If <code>query_mode</code> is <code>sync</code>, calls to the resource will be blocked for this amount of time before timing out."""

View File

@ -30,16 +30,25 @@ namespace() -> "resource_schema".
roots() -> [].
fields("resource_opts_sync_only") ->
[
{resource_opts,
mk(
ref(?MODULE, "creation_opts_sync_only"),
resource_opts_meta()
)}
];
fields("creation_opts_sync_only") ->
Fields0 = fields("creation_opts"),
Fields1 = lists:keydelete(async_inflight_window, 1, Fields0),
QueryMod = {query_mode, fun query_mode_sync_only/1},
lists:keyreplace(query_mode, 1, Fields1, QueryMod);
fields("resource_opts") ->
[
{resource_opts,
mk(
ref(?MODULE, "creation_opts"),
#{
required => false,
default => #{},
desc => ?DESC(<<"resource_opts">>)
}
resource_opts_meta()
)}
];
fields("creation_opts") ->
@ -59,6 +68,13 @@ fields("creation_opts") ->
{max_queue_bytes, fun max_queue_bytes/1}
].
resource_opts_meta() ->
#{
required => false,
default => #{},
desc => ?DESC(<<"resource_opts">>)
}.
worker_pool_size(type) -> non_neg_integer();
worker_pool_size(desc) -> ?DESC("worker_pool_size");
worker_pool_size(default) -> ?WORKER_POOL_SIZE;
@ -95,6 +111,12 @@ query_mode(default) -> async;
query_mode(required) -> false;
query_mode(_) -> undefined.
query_mode_sync_only(type) -> enum([sync]);
query_mode_sync_only(desc) -> ?DESC("query_mode_sync_only");
query_mode_sync_only(default) -> async;
query_mode_sync_only(required) -> false;
query_mode_sync_only(_) -> undefined.
request_timeout(type) -> hoconsc:union([infinity, emqx_schema:duration_ms()]);
request_timeout(desc) -> ?DESC("request_timeout");
request_timeout(default) -> <<"15s">>;
@ -139,4 +161,6 @@ max_queue_bytes(required) -> false;
max_queue_bytes(_) -> undefined.
desc("creation_opts") ->
?DESC("creation_opts");
desc("creation_opts_sync_only") ->
?DESC("creation_opts").

View File

@ -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");
] ++ emqx_resource_schema:fields("resource_opts_sync_only");
fields(mongodb_rs) ->
emqx_connector_mongo:fields(rs) ++ fields("config");
fields(mongodb_sharded) ->
@ -69,32 +69,7 @@ fields("get_sharded") ->
fields("get_single") ->
emqx_bridge_schema:status_fields() ++
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 => async
}
)};
(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")}
)}
].
type_and_name_fields(mongodb_single).
conn_bridge_examples(Method) ->
[

View File

@ -98,8 +98,7 @@ fields("config") ->
(emqx_connector_mysql:fields(config) --
emqx_connector_schema_lib:prepare_statement_fields());
fields("creation_opts") ->
Opts = emqx_resource_schema:fields("creation_opts"),
[O || {Field, _} = O <- Opts, not is_hidden_opts(Field)];
emqx_resource_schema:fields("creation_opts_sync_only");
fields("post") ->
[type_field(), name_field() | fields("config")];
fields("put") ->
@ -118,10 +117,6 @@ desc(_) ->
%% -------------------------------------------------------------------------------------------------
%% internal
is_hidden_opts(Field) ->
lists:member(Field, [
async_inflight_window
]).
type_field() ->
{type, mk(enum([mysql]), #{required => true, desc => ?DESC("desc_type")})}.

View File

@ -100,8 +100,7 @@ fields("config") ->
(emqx_connector_pgsql:fields(config) --
emqx_connector_schema_lib:prepare_statement_fields());
fields("creation_opts") ->
Opts = emqx_resource_schema:fields("creation_opts"),
[O || {Field, _} = O <- Opts, not is_hidden_opts(Field)];
emqx_resource_schema:fields("creation_opts_sync_only");
fields("post") ->
fields("post", pgsql);
fields("put") ->
@ -122,11 +121,6 @@ desc(_) ->
undefined.
%% -------------------------------------------------------------------------------------------------
%% internal
is_hidden_opts(Field) ->
lists:member(Field, [
async_inflight_window
]).
type_field(Type) ->
{type, mk(enum([Type]), #{required => true, desc => ?DESC("desc_type")})}.

View File

@ -181,10 +181,10 @@ resource_fields(Type) ->
resource_creation_fields("redis_cluster") ->
% TODO
% Cluster bridge is currently incompatible with batching.
Fields = emqx_resource_schema:fields("creation_opts"),
Fields = emqx_resource_schema:fields("creation_opts_sync_only"),
lists:foldl(fun proplists:delete/2, Fields, [batch_size, batch_time]);
resource_creation_fields(_) ->
emqx_resource_schema:fields("creation_opts").
emqx_resource_schema:fields("creation_opts_sync_only").
desc("config") ->
?DESC("desc_config");