157 lines
4.4 KiB
Erlang
157 lines
4.4 KiB
Erlang
-module(emqx_retainer_schema).
|
|
|
|
-include_lib("typerefl/include/types.hrl").
|
|
-include_lib("hocon/include/hoconsc.hrl").
|
|
|
|
-export([roots/0, fields/1, desc/1, namespace/0]).
|
|
|
|
-define(DEFAULT_INDICES, [
|
|
[1, 2, 3],
|
|
[1, 3],
|
|
[2, 3],
|
|
[3]
|
|
]).
|
|
|
|
-define(INVALID_SPEC(_REASON_), throw({_REASON_, #{default => ?DEFAULT_INDICES}})).
|
|
|
|
namespace() -> "retainer".
|
|
|
|
roots() -> ["retainer"].
|
|
|
|
fields("retainer") ->
|
|
[
|
|
{enable, sc(boolean(), enable, true)},
|
|
{msg_expiry_interval,
|
|
sc(
|
|
emqx_schema:duration_ms(),
|
|
msg_expiry_interval,
|
|
"0s"
|
|
)},
|
|
{msg_clear_interval,
|
|
sc(
|
|
emqx_schema:duration_ms(),
|
|
msg_clear_interval,
|
|
"0s"
|
|
)},
|
|
{flow_control,
|
|
sc(
|
|
?R_REF(flow_control),
|
|
flow_control,
|
|
#{}
|
|
)},
|
|
{max_payload_size,
|
|
sc(
|
|
emqx_schema:bytesize(),
|
|
max_payload_size,
|
|
"1MB"
|
|
)},
|
|
{stop_publish_clear_msg,
|
|
sc(
|
|
boolean(),
|
|
stop_publish_clear_msg,
|
|
false
|
|
)},
|
|
{backend, backend_config()}
|
|
];
|
|
fields(mnesia_config) ->
|
|
[
|
|
{type, sc(built_in_database, mnesia_config_type, built_in_database)},
|
|
{storage_type,
|
|
sc(
|
|
hoconsc:enum([ram, disc]),
|
|
mnesia_config_storage_type,
|
|
ram
|
|
)},
|
|
{max_retained_messages,
|
|
sc(
|
|
non_neg_integer(),
|
|
max_retained_messages,
|
|
0
|
|
)},
|
|
{index_specs, fun retainer_indices/1}
|
|
];
|
|
fields(flow_control) ->
|
|
[
|
|
{batch_read_number,
|
|
sc(
|
|
non_neg_integer(),
|
|
batch_read_number,
|
|
0
|
|
)},
|
|
{batch_deliver_number,
|
|
sc(
|
|
range(0, 1000),
|
|
batch_deliver_number,
|
|
0
|
|
)},
|
|
{batch_deliver_limiter,
|
|
sc(
|
|
?R_REF(emqx_limiter_schema, internal),
|
|
batch_deliver_limiter,
|
|
undefined
|
|
)}
|
|
].
|
|
|
|
desc("retainer") ->
|
|
"Configuration related to handling `PUBLISH` packets with a `retain` flag set to 1.";
|
|
desc(mnesia_config) ->
|
|
"Configuration of the internal database storing retained messages.";
|
|
desc(flow_control) ->
|
|
"Retainer batching and rate limiting.";
|
|
desc(_) ->
|
|
undefined.
|
|
|
|
%%--------------------------------------------------------------------
|
|
%% Internal functions
|
|
%%--------------------------------------------------------------------
|
|
|
|
%%sc(Type, DescId) ->
|
|
%% hoconsc:mk(Type, #{desc => ?DESC(DescId)}).
|
|
|
|
sc(Type, DescId, Default) ->
|
|
hoconsc:mk(Type, #{default => Default, desc => ?DESC(DescId)}).
|
|
|
|
backend_config() ->
|
|
hoconsc:mk(hoconsc:ref(?MODULE, mnesia_config), #{desc => ?DESC(backend)}).
|
|
|
|
retainer_indices(type) ->
|
|
list(list(integer()));
|
|
retainer_indices(desc) ->
|
|
"Retainer index specifications: list of arrays of positive ascending integers. "
|
|
"Each array specifies an index. Numbers in an index specification are 1-based "
|
|
"word positions in topics. Words from specified positions will be used for indexing.<br/>"
|
|
"For example, it is good to have <code>[2, 4]</code> index to optimize "
|
|
"<code>+/X/+/Y/...</code> topic wildcard subscriptions.";
|
|
retainer_indices(example) ->
|
|
[[2, 4], [1, 3]];
|
|
retainer_indices(default) ->
|
|
?DEFAULT_INDICES;
|
|
retainer_indices(validator) ->
|
|
fun check_index_specs/1;
|
|
retainer_indices(_) ->
|
|
undefined.
|
|
|
|
check_index_specs([]) ->
|
|
ok;
|
|
check_index_specs(IndexSpecs) when is_list(IndexSpecs) ->
|
|
lists:foreach(fun check_index_spec/1, IndexSpecs),
|
|
check_duplicate(IndexSpecs);
|
|
check_index_specs(_IndexSpecs) ->
|
|
?INVALID_SPEC(list_index_spec_limited).
|
|
|
|
check_index_spec([]) ->
|
|
?INVALID_SPEC(non_empty_index_spec_limited);
|
|
check_index_spec(IndexSpec) when is_list(IndexSpec) ->
|
|
case lists:all(fun(Idx) -> is_integer(Idx) andalso Idx > 0 end, IndexSpec) of
|
|
false -> ?INVALID_SPEC(pos_integer_index_limited);
|
|
true -> check_duplicate(IndexSpec)
|
|
end;
|
|
check_index_spec(_IndexSpec) ->
|
|
?INVALID_SPEC(list_index_spec_limited).
|
|
|
|
check_duplicate(List) ->
|
|
case length(List) =:= length(lists:usort(List)) of
|
|
false -> ?INVALID_SPEC(unique_index_spec_limited);
|
|
true -> ok
|
|
end.
|