yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(bridge): add is_template flag to bridge config fields

This commit is contained in:
zmstone 2024-04-16 19:00:32 +02:00
parent df458b98d7
commit 51c8173174
32 changed files with 133 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
apps/emqx/src/emqx_schema.erl
View File

@ -61,6 +61,7 @@
}.
-type url() :: binary().
-type json_binary() :: binary().
-type template() :: binary().
-typerefl_from_string({duration/0, emqx_schema, to_duration}).
-typerefl_from_string({duration_s/0, emqx_schema, to_duration_s}).
@ -78,6 +79,7 @@
-typerefl_from_string({comma_separated_atoms/0, emqx_schema, to_comma_separated_atoms}).
-typerefl_from_string({url/0, emqx_schema, to_url}).
-typerefl_from_string({json_binary/0, emqx_schema, to_json_binary}).
-typerefl_from_string({template/0, emqx_schema, to_template}).
-type parsed_server() :: #{
hostname := string(),
@ -120,7 +122,8 @@
to_erl_cipher_suite/1,
to_comma_separated_atoms/1,
to_url/1,
to_json_binary/1
to_json_binary/1,
to_template/1
]).
-export([
@ -160,7 +163,8 @@
comma_separated_atoms/0,
url/0,
json_binary/0,
port_number/0
port_number/0,
template/0
]).
-export([namespace/0, roots/0, roots/1, fields/1, desc/1, tags/0]).
@ -2594,6 +2598,9 @@ to_json_binary(Str) ->
Error
end.
to_template(Str) ->
{ok, iolist_to_binary(Str)}.
%% @doc support the following format:
%% - 127.0.0.1:1883
%% - ::1:1883

2
apps/emqx_bridge_cassandra/src/emqx_bridge_cassandra.erl
View File

@ -181,7 +181,7 @@ fields("post", Type) ->
cql_field() ->
{cql,
mk(
binary(),
emqx_schema:template(),
#{desc => ?DESC("cql_template"), default => ?DEFAULT_CQL, format => <<"sql">>}
)}.

8
apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse.erl
View File

@ -184,8 +184,12 @@ fields("post", Type) ->
sql_field() ->
{sql,
mk(
binary(),
#{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
emqx_schema:template(),
#{
desc => ?DESC("sql_template"),
default => ?DEFAULT_SQL,
format => <<"sql">>
}
)}.
batch_value_separator_field() ->

23
apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo.erl
View File

@ -160,13 +160,7 @@ fields(dynamo_action) ->
);
fields(action_parameters) ->
Parameters =
[
{template,
mk(
binary(),
#{desc => ?DESC("template"), default => ?DEFAULT_TEMPLATE}
)}
] ++ emqx_bridge_dynamo_connector:fields(config),
[{template, template_field_schema()}] ++ emqx_bridge_dynamo_connector:fields(config),
lists:foldl(
fun(Key, Acc) ->
proplists:delete(Key, Acc)
@ -199,11 +193,7 @@ fields(connector_resource_opts) ->
fields("config") ->
[
{enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
{template,
mk(
binary(),
#{desc => ?DESC("template"), default => ?DEFAULT_TEMPLATE}
)},
{template, template_field_schema()},
{local_topic,
mk(
binary(),
@ -230,6 +220,15 @@ fields("put") ->
fields("get") ->
emqx_bridge_schema:status_fields() ++ fields("post").
template_field_schema() ->
mk(
emqx_schema:template(),
#{
desc => ?DESC("template"),
default => ?DEFAULT_TEMPLATE
}
).
desc("config") ->
?DESC("desc_config");
desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->

8
apps/emqx_bridge_es/src/emqx_bridge_es.erl
View File

@ -135,7 +135,7 @@ overwrite() ->
index() ->
{index,
?HOCON(
binary(),
emqx_schema:template(),
#{
required => true,
example => <<"${payload.index}">>,
@ -146,7 +146,7 @@ index() ->
id(Required) ->
{id,
?HOCON(
binary(),
emqx_schema:template(),
#{
required => Required,
example => <<"${payload.id}">>,
@ -157,7 +157,7 @@ id(Required) ->
doc() ->
{doc,
?HOCON(
binary(),
emqx_schema:template(),
#{
required => false,
example => <<"${payload.doc}">>,
@ -187,7 +187,7 @@ doc_as_upsert() ->
routing() ->
{routing,
?HOCON(
binary(),
emqx_schema:template(),
#{
required => false,
example => <<"${payload.routing}">>,

19
apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub.erl
View File

@ -122,7 +122,7 @@ fields(producer) ->
)},
{ordering_key_template,
sc(
binary(),
emqx_schema:template(),
#{
default => <<>>,
desc => ?DESC("ordering_key_template")
@ -130,7 +130,7 @@ fields(producer) ->
)},
{payload_template,
sc(
binary(),
emqx_schema:template(),
#{
default => <<>>,
desc => ?DESC("payload_template")
@ -201,8 +201,11 @@ fields(consumer_topic_mapping) ->
{qos, mk(emqx_schema:qos(), #{default => 0, desc => ?DESC(consumer_mqtt_qos)})},
{payload_template,
mk(
string(),
#{default => <<"${.}">>, desc => ?DESC(consumer_mqtt_payload)}
emqx_schema:template(),
#{
default => <<"${.}">>,
desc => ?DESC(consumer_mqtt_payload)
}
)}
];
fields("consumer_resource_opts") ->
@ -221,14 +224,18 @@ fields("consumer_resource_opts") ->
fields(key_value_pair) ->
[
{key,
mk(binary(), #{
mk(emqx_schema:template(), #{
required => true,
validator => [
emqx_resource_validator:not_empty("Key templates must not be empty")
],
desc => ?DESC(kv_pair_key)
})},
{value, mk(binary(), #{required => true, desc => ?DESC(kv_pair_value)})}
{value,
mk(emqx_schema:template(), #{
required => true,
desc => ?DESC(kv_pair_value)
})}
];
fields("get_producer") ->
emqx_bridge_schema:status_fields() ++ fields("post_producer");

17
apps/emqx_bridge_hstreamdb/src/emqx_bridge_hstreamdb.erl
View File

@ -167,13 +167,13 @@ fields(action_parameters) ->
})},
{partition_key,
mk(binary(), #{
required => false, desc => ?DESC(emqx_bridge_hstreamdb_connector, "partition_key")
mk(emqx_schema:template(), #{
required => false,
desc => ?DESC(emqx_bridge_hstreamdb_connector, "partition_key")
})},
{grpc_flush_timeout, fun grpc_flush_timeout/1},
{record_template,
mk(binary(), #{default => <<"${payload}">>, desc => ?DESC("record_template")})},
{record_template, record_template_schema()},
{aggregation_pool_size,
mk(pos_integer(), #{
default => ?DEFAULT_AGG_POOL_SIZE, desc => ?DESC("aggregation_pool_size")
@ -222,6 +222,12 @@ fields("put") ->
hstream_bridge_common_fields() ++
connector_fields().
record_template_schema() ->
mk(emqx_schema:template(), #{
default => <<"${payload}">>,
desc => ?DESC("record_template")
}).
grpc_timeout(type) -> emqx_schema:timeout_duration_ms();
grpc_timeout(desc) -> ?DESC(emqx_bridge_hstreamdb_connector, "grpc_timeout");
grpc_timeout(default) -> ?DEFAULT_GRPC_TIMEOUT_RAW;
@ -239,8 +245,7 @@ hstream_bridge_common_fields() ->
[
{direction, mk(egress, #{desc => ?DESC("config_direction"), default => egress})},
{local_topic, mk(binary(), #{desc => ?DESC("local_topic")})},
{record_template,
mk(binary(), #{default => <<"${payload}">>, desc => ?DESC("record_template")})}
{record_template, record_template_schema()}
] ++
emqx_resource_schema:fields("resource_opts").

2
apps/emqx_bridge_http/src/emqx_bridge_http_schema.erl
View File

@ -287,7 +287,7 @@ method_field() ->
body_field() ->
{body,
mk(
binary(),
emqx_schema:template(),
#{
default => undefined,
desc => ?DESC("config_body")

2
apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.erl
View File

@ -42,7 +42,7 @@
%% api
write_syntax_type() ->
typerefl:alias("string", write_syntax()).
typerefl:alias("template", write_syntax()).
%% Examples
conn_bridge_examples(Method) ->

10
apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.erl
View File

@ -84,7 +84,7 @@ fields(action_parameters) ->
)},
{device_id,
mk(
binary(),
emqx_schema:template(),
#{
desc => ?DESC("config_device_id")
}
@ -114,7 +114,7 @@ fields(action_parameters_data) ->
)},
{measurement,
mk(
binary(),
emqx_schema:template(),
#{
required => true,
desc => ?DESC("config_parameters_measurement")
@ -122,7 +122,9 @@ fields(action_parameters_data) ->
)},
{data_type,
mk(
hoconsc:union([enum([text, boolean, int32, int64, float, double]), binary()]),
hoconsc:union([
enum([text, boolean, int32, int64, float, double]), emqx_schema:template()
]),
#{
required => true,
desc => ?DESC("config_parameters_data_type")
@ -130,7 +132,7 @@ fields(action_parameters_data) ->
)},
{value,
mk(
binary(),
emqx_schema:template(),
#{
required => true,
desc => ?DESC("config_parameters_value")

24
apps/emqx_bridge_kafka/src/emqx_bridge_kafka.erl
View File

@ -477,11 +477,20 @@ fields(producer_kafka_ext_headers) ->
];
fields(kafka_message) ->
[
{key, mk(string(), #{default => <<"${.clientid}">>, desc => ?DESC(kafka_message_key)})},
{value, mk(string(), #{default => <<"${.}">>, desc => ?DESC(kafka_message_value)})},
{key,
mk(emqx_schema:template(), #{
default => <<"${.clientid}">>,
desc => ?DESC(kafka_message_key)
})},
{value,
mk(emqx_schema:template(), #{
default => <<"${.}">>,
desc => ?DESC(kafka_message_value)
})},
{timestamp,
mk(string(), #{
default => <<"${.timestamp}">>, desc => ?DESC(kafka_message_timestamp)
mk(emqx_schema:template(), #{
default => <<"${.timestamp}">>,
desc => ?DESC(kafka_message_timestamp)
})}
];
fields(producer_buffer) ->
@ -536,8 +545,11 @@ fields(consumer_topic_mapping) ->
{qos, mk(emqx_schema:qos(), #{default => 0, desc => ?DESC(consumer_mqtt_qos)})},
{payload_template,
mk(
string(),
#{default => <<"${.}">>, desc => ?DESC(consumer_mqtt_payload)}
emqx_schema:template(),
#{
default => <<"${.}">>,
desc => ?DESC(consumer_mqtt_payload)
}
)}
];
fields(consumer_kafka_opts) ->

2
apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis.erl
View File

@ -150,7 +150,7 @@ fields(producer) ->
[
{payload_template,
sc(
binary(),
emqx_schema:template(),
#{
default => <<"${.}">>,
desc => ?DESC("payload_template")

6
apps/emqx_bridge_mongodb/src/emqx_bridge_mongodb.erl
View File

@ -44,8 +44,10 @@ roots() -> [].
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")})},
{collection,
mk(emqx_schema:template(), #{desc => ?DESC("collection"), default => <<"mqtt">>})},
{payload_template,
mk(emqx_schema:template(), #{required => false, desc => ?DESC("payload_template")})},
{resource_opts,
mk(
ref(?MODULE, "creation_opts"),

14
apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt_connector_schema.erl
View File

@ -200,7 +200,7 @@ fields("ingress_local") ->
[
{topic,
mk(
binary(),
emqx_schema:template(),
#{
validator => fun emqx_schema:non_empty_string/1,
desc => ?DESC("ingress_local_topic"),
@ -217,7 +217,7 @@ fields("ingress_local") ->
)},
{retain,
mk(
hoconsc:union([boolean(), binary()]),
hoconsc:union([boolean(), emqx_schema:template()]),
#{
default => <<"${retain}">>,
desc => ?DESC("retain")
@ -225,7 +225,7 @@ fields("ingress_local") ->
)},
{payload,
mk(
binary(),
emqx_schema:template(),
#{
default => undefined,
desc => ?DESC("payload")
@ -268,7 +268,7 @@ fields("egress_remote") ->
[
{topic,
mk(
binary(),
emqx_schema:template(),
#{
required => true,
validator => fun emqx_schema:non_empty_string/1,
@ -286,7 +286,7 @@ fields("egress_remote") ->
)},
{retain,
mk(
hoconsc:union([boolean(), binary()]),
hoconsc:union([boolean(), emqx_schema:template()]),
#{
required => false,
default => false,
@ -295,7 +295,7 @@ fields("egress_remote") ->
)},
{payload,
mk(
binary(),
emqx_schema:template(),
#{
default => undefined,
desc => ?DESC("payload")
@ -344,7 +344,7 @@ desc(_) ->
undefined.
qos() ->
hoconsc:union([emqx_schema:qos(), binary()]).
hoconsc:union([emqx_schema:qos(), emqx_schema:template()]).
parse_server(Str) ->
#{hostname := Host, port := Port} = emqx_schema:parse_server(Str, ?MQTT_HOST_OPTS),

2
apps/emqx_bridge_mysql/src/emqx_bridge_mysql.erl
View File

@ -117,7 +117,7 @@ fields("config") ->
{enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
{sql,
mk(
binary(),
emqx_schema:template(),
#{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
)},
{local_topic,

8
apps/emqx_bridge_opents/src/emqx_bridge_opents.erl
View File

@ -146,7 +146,7 @@ fields(action_parameters_data) ->
[
{timestamp,
mk(
binary(),
emqx_schema:template(),
#{
desc => ?DESC("config_parameters_timestamp"),
required => false
@ -154,7 +154,7 @@ fields(action_parameters_data) ->
)},
{metric,
mk(
binary(),
emqx_schema:template(),
#{
required => true,
desc => ?DESC("config_parameters_metric")
@ -162,7 +162,7 @@ fields(action_parameters_data) ->
)},
{tags,
mk(
hoconsc:union([map(), binary()]),
hoconsc:union([map(), emqx_schema:template()]),
#{
required => true,
desc => ?DESC("config_parameters_tags"),
@ -188,7 +188,7 @@ fields(action_parameters_data) ->
)},
{value,
mk(
hoconsc:union([integer(), float(), binary()]),
hoconsc:union([integer(), float(), emqx_schema:template()]),
#{
required => true,
desc => ?DESC("config_parameters_value")

4
apps/emqx_bridge_oracle/src/emqx_bridge_oracle.erl
View File

@ -158,7 +158,7 @@ fields(action_parameters) ->
[
{sql,
hoconsc:mk(
binary(),
emqx_schema:template(),
#{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
)}
];
@ -177,7 +177,7 @@ fields("config") ->
)},
{sql,
hoconsc:mk(
binary(),
emqx_schema:template(),
#{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
)},
{local_topic,

2
apps/emqx_bridge_pgsql/src/emqx_bridge_pgsql.erl
View File

@ -61,7 +61,7 @@ fields(action_parameters) ->
[
{sql,
hoconsc:mk(
binary(),
emqx_schema:template(),
#{desc => ?DESC("sql_template"), default => default_sql(), format => <<"sql">>}
)}
];

4
apps/emqx_bridge_pulsar/src/emqx_bridge_pulsar_pubsub_schema.erl
View File

@ -51,12 +51,12 @@ fields(action_parameters) ->
fields(producer_pulsar_message) ->
[
{key,
?HOCON(string(), #{
?HOCON(emqx_schema:template(), #{
default => <<"${.clientid}">>,
desc => ?DESC("producer_key_template")
})},
{value,
?HOCON(string(), #{
?HOCON(emqx_schema:template(), #{
default => <<"${.}">>,
desc => ?DESC("producer_value_template")
})}

2
apps/emqx_bridge_rabbitmq/src/emqx_bridge_rabbitmq_pubsub_schema.erl
View File

@ -99,7 +99,7 @@ fields(action_parameters) ->
)},
{payload_template,
hoconsc:mk(
binary(),
emqx_schema:template(),
#{
default => <<"">>,
desc => ?DESC(?CONNECTOR_SCHEMA, "payload_template")

2
apps/emqx_bridge_redis/src/emqx_bridge_redis.erl
View File

@ -211,7 +211,7 @@ desc(_) ->
undefined.
command_template(type) ->
list(binary());
hoconsc:array(emqx_schema:template());
command_template(required) ->
true;
command_template(validator) ->

12
apps/emqx_bridge_rocketmq/src/emqx_bridge_rocketmq.erl
View File

@ -162,8 +162,11 @@ fields(action_parameters) ->
[
{template,
mk(
binary(),
#{desc => ?DESC("template"), default => ?DEFAULT_TEMPLATE}
emqx_schema:template(),
#{
desc => ?DESC("template"),
default => ?DEFAULT_TEMPLATE
}
)}
] ++ emqx_bridge_rocketmq_connector:fields(config),
lists:foldl(
@ -205,7 +208,7 @@ fields("config") ->
{enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
{template,
mk(
binary(),
emqx_schema:template(),
#{desc => ?DESC("template"), default => ?DEFAULT_TEMPLATE}
)},
{local_topic,
@ -214,8 +217,7 @@ fields("config") ->
#{desc => ?DESC("local_topic"), required => false}
)}
] ++ emqx_resource_schema:fields("resource_opts") ++
(emqx_bridge_rocketmq_connector:fields(config) --
emqx_connector_schema_lib:prepare_statement_fields());
emqx_bridge_rocketmq_connector:fields(config);
fields("post") ->
[type_field(), name_field() | fields("config")];
fields("put") ->

2
apps/emqx_bridge_rocketmq/src/emqx_bridge_rocketmq_connector.erl
View File

@ -47,7 +47,7 @@ fields(config) ->
{servers, servers()},
{topic,
mk(
binary(),
emqx_schema:template(),
#{default => <<"TopicTest">>, desc => ?DESC(topic)}
)},
{access_key,

2
apps/emqx_bridge_s3/src/emqx_bridge_s3.erl
View File

@ -77,7 +77,7 @@ fields(s3_upload_parameters) ->
[
{content,
hoconsc:mk(
string(),
emqx_schema:template(),
#{
required => false,
default => <<"${.}">>,

2
apps/emqx_bridge_sqlserver/src/emqx_bridge_sqlserver.erl
View File

@ -192,7 +192,7 @@ fields(action_parameters) ->
[
{sql,
mk(
binary(),
emqx_schema:template(),
#{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
)}
];

4
apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper.erl
View File

@ -112,7 +112,7 @@ fields("parameters") ->
[
{target_topic,
mk(
binary(),
emqx_schema:template(),
#{desc => ?DESC("target_topic"), default => <<"${topic}">>}
)},
{target_qos,
@ -122,7 +122,7 @@ fields("parameters") ->
)},
{template,
mk(
binary(),
emqx_schema:template(),
#{desc => ?DESC("template"), default => <<"${payload}">>}
)}
];

4
apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine.erl
View File

@ -83,7 +83,7 @@ fields("config") ->
{enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
{sql,
mk(
binary(),
emqx_schema:template(),
#{
desc => ?DESC("sql_template"),
default => ?DEFAULT_SQL,
@ -125,7 +125,7 @@ fields(action_parameters) ->
{database, fun emqx_connector_schema_lib:database/1},
{sql,
mk(
binary(),
emqx_schema:template(),
#{
desc => ?DESC("sql_template"),
default => ?DEFAULT_SQL,

6
apps/emqx_conf/src/emqx_conf_schema_types.erl
View File

@ -65,6 +65,12 @@ readable("boolean()") ->
dashboard => #{type => boolean},
docgen => #{type => "Boolean"}
};
readable("template()") ->
#{
swagger => #{type => string},
dashboard => #{type => string, is_template => true},
docgen => #{type => "String", desc => ?DESC(template)}
};
readable("binary()") ->
#{
swagger => #{type => string},

2
apps/emqx_s3/src/emqx_s3.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_s3, [
{description, "EMQX S3"},
{vsn, "5.0.14"},
{vsn, "5.1.0"},
{modules, []},
{registered, [emqx_s3_sup]},
{applications, [

3
apps/emqx_s3/src/emqx_s3_client.erl
View File

@ -103,7 +103,7 @@ put_object(Client, Key, Value) ->
-spec put_object(client(), key(), upload_options(), iodata()) -> ok_or_error(term()).
put_object(
#{bucket := Bucket, headers := BaseHeaders, aws_config := AwsConfig = #aws_config{}},
#{bucket := Bucket0, headers := BaseHeaders, aws_config := AwsConfig = #aws_config{}},
Key,
UploadOpts,
Content
@ -111,6 +111,7 @@ put_object(
ECKey = erlcloud_key(Key),
ECOpts = erlcloud_upload_options(UploadOpts),
Headers = join_headers(BaseHeaders, maps:get(headers, UploadOpts, undefined)),
Bucket = to_list_string(Bucket0),
try erlcloud_s3:put_object(Bucket, ECKey, Content, ECOpts, Headers, AwsConfig) of
Props when is_list(Props) ->
ok

4
apps/emqx_s3/src/emqx_s3_schema.erl
View File

@ -74,7 +74,7 @@ fields(s3_upload) ->
[
{bucket,
mk(
string(),
emqx_schema:template(),
#{
desc => ?DESC("bucket"),
required => true
@ -82,7 +82,7 @@ fields(s3_upload) ->
)},
{key,
mk(
string(),
emqx_schema:template(),
#{
desc => ?DESC("key"),
required => true

3
rel/i18n/emqx_conf_schema_types.hocon
View File

@ -9,4 +9,7 @@ emqx_conf_schema_types {
secret.desc:
"""A string holding some sensitive information, such as a password. When secret starts with <code>file://</code>, the rest of the string is interpreted as a path to a file containing the secret itself: whole content of the file except any trailing whitespace characters is considered a secret value. Note: when clustered, all EMQX nodes should have the same file present before using <code>file://</code> secrets."""
template.desc: """~
A string for `${.path.to.var}` style value interpolation,
where the leading dot is optional, and `${.}` represents all values as an object."""
}