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

Merge pull request #11446 from lafirest/refactor/calendar 

refactor(calendar): refactor datetime-related code and remove redundant
This commit is contained in:
lafirest 2023-08-17 07:40:44 +08:00 committed by GitHub
parent 740680d54b 040d28c35e
commit 5e448f5a02
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
36 changed files with 255 additions and 537 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apps/emqx/src/emqx_alarm.erl
View File

@ -213,7 +213,7 @@ format(Node, #deactivated_alarm{
to_rfc3339(Timestamp) ->
%% rfc3339 accuracy to millisecond
list_to_binary(calendar:system_time_to_rfc3339(Timestamp div 1000, [{unit, millisecond}])).
emqx_utils_calendar:epoch_to_rfc3339(Timestamp div 1000).
%%--------------------------------------------------------------------
%% gen_server callbacks

2
apps/emqx/src/emqx_banned.erl
View File

@ -172,7 +172,7 @@ maybe_format_host({As, Who}) ->
{As, Who}.
to_rfc3339(Timestamp) ->
list_to_binary(calendar:system_time_to_rfc3339(Timestamp, [{unit, second}])).
emqx_utils_calendar:epoch_to_rfc3339(Timestamp, second).
-spec create(emqx_types:banned() | map()) ->
{ok, emqx_types:banned()} | {error, {already_exist, emqx_types:banned()}}.

156
apps/emqx/src/emqx_datetime.erl
View File

@ -1,156 +0,0 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2017-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_datetime).
-include_lib("typerefl/include/types.hrl").
%% API
-export([
to_epoch_millisecond/1,
to_epoch_second/1,
human_readable_duration_string/1
]).
-export([
epoch_to_rfc3339/1,
epoch_to_rfc3339/2
]).
-reflect_type([
epoch_millisecond/0,
epoch_second/0
]).
-type epoch_second() :: non_neg_integer().
-type epoch_millisecond() :: non_neg_integer().
-typerefl_from_string({epoch_second/0, ?MODULE, to_epoch_second}).
-typerefl_from_string({epoch_millisecond/0, ?MODULE, to_epoch_millisecond}).
%% the maximum value is the SECONDS_FROM_0_TO_10000 in the calendar.erl,
%% here minus SECONDS_PER_DAY to tolerate timezone time offset,
%% so the maximum date can reach 9999-12-31 which is ample.
-define(MAXIMUM_EPOCH, 253402214400).
-define(MAXIMUM_EPOCH_MILLI, 253402214400_000).
to_epoch_second(DateTime) ->
to_epoch(DateTime, second).
to_epoch_millisecond(DateTime) ->
to_epoch(DateTime, millisecond).
to_epoch(DateTime, Unit) ->
try
case string:to_integer(DateTime) of
{Epoch, []} -> validate_epoch(Epoch, Unit);
_ -> {ok, calendar:rfc3339_to_system_time(DateTime, [{unit, Unit}])}
end
catch
error:_ ->
{error, bad_rfc3339_timestamp}
end.
epoch_to_rfc3339(TimeStamp) ->
epoch_to_rfc3339(TimeStamp, millisecond).
epoch_to_rfc3339(TimeStamp, Unit) when is_integer(TimeStamp) ->
list_to_binary(calendar:system_time_to_rfc3339(TimeStamp, [{unit, Unit}])).
-spec human_readable_duration_string(integer()) -> string().
human_readable_duration_string(Milliseconds) ->
Seconds = Milliseconds div 1000,
{D, {H, M, S}} = calendar:seconds_to_daystime(Seconds),
L0 = [{D, " days"}, {H, " hours"}, {M, " minutes"}, {S, " seconds"}],
L1 = lists:dropwhile(fun({K, _}) -> K =:= 0 end, L0),
L2 = lists:map(fun({Time, Unit}) -> [integer_to_list(Time), Unit] end, L1),
lists:flatten(lists:join(", ", L2)).
validate_epoch(Epoch, _Unit) when Epoch < 0 ->
{error, bad_epoch};
validate_epoch(Epoch, second) when Epoch =< ?MAXIMUM_EPOCH ->
{ok, Epoch};
validate_epoch(Epoch, millisecond) when Epoch =< ?MAXIMUM_EPOCH_MILLI ->
{ok, Epoch};
validate_epoch(_Epoch, _Unit) ->
{error, bad_epoch}.
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-compile(nowarn_export_all).
-compile(export_all).
roots() -> [bar].
fields(bar) ->
[
{second, ?MODULE:epoch_second()},
{millisecond, ?MODULE:epoch_millisecond()}
].
-define(FORMAT(_Sec_, _Ms_),
lists:flatten(
io_lib:format("bar={second=~w,millisecond=~w}", [_Sec_, _Ms_])
)
).
epoch_ok_test() ->
BigStamp = 1 bsl 37,
Args = [
{0, 0, 0, 0},
{1, 1, 1, 1},
{BigStamp, BigStamp * 1000, BigStamp, BigStamp * 1000},
{"2022-01-01T08:00:00+08:00", "2022-01-01T08:00:00+08:00", 1640995200, 1640995200000}
],
lists:foreach(
fun({Sec, Ms, EpochSec, EpochMs}) ->
check_ok(?FORMAT(Sec, Ms), EpochSec, EpochMs)
end,
Args
),
ok.
check_ok(Input, Sec, Ms) ->
{ok, Data} = hocon:binary(Input, #{}),
?assertMatch(
#{bar := #{second := Sec, millisecond := Ms}},
hocon_tconf:check_plain(?MODULE, Data, #{atom_key => true}, [bar])
),
ok.
epoch_failed_test() ->
BigStamp = 1 bsl 38,
Args = [
{-1, -1},
{"1s", "1s"},
{BigStamp, 0},
{0, BigStamp * 1000},
{"2022-13-13T08:00:00+08:00", "2022-13-13T08:00:00+08:00"}
],
lists:foreach(
fun({Sec, Ms}) ->
check_failed(?FORMAT(Sec, Ms))
end,
Args
),
ok.
check_failed(Input) ->
{ok, Data} = hocon:binary(Input, #{}),
?assertException(
throw,
_,
hocon_tconf:check_plain(?MODULE, Data, #{atom_key => true}, [bar])
),
ok.
-endif.

2
apps/emqx/src/emqx_trace/emqx_trace_formatter.erl
View File

@ -28,7 +28,7 @@ format(
#{level := debug, meta := Meta = #{trace_tag := Tag}, msg := Msg},
#{payload_encode := PEncode}
) ->
Time = calendar:system_time_to_rfc3339(erlang:system_time(microsecond), [{unit, microsecond}]),
Time = emqx_utils_calendar:now_to_rfc3339(microsecond),
ClientId = to_iolist(maps:get(clientid, Meta, "")),
Peername = maps:get(peername, Meta, ""),
MetaBin = format_meta(Meta, PEncode),

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

@ -1,6 +1,6 @@
{application, emqx_ft, [
{description, "EMQX file transfer over MQTT"},
{vsn, "0.1.4"},
{vsn, "0.1.5"},
{registered, []},
{mod, {emqx_ft_app, []}},
{applications, [

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

@ -71,7 +71,7 @@
%% the resulting file is corrupted during transmission).
size => _Bytes :: non_neg_integer(),
checksum => checksum(),
expire_at := emqx_datetime:epoch_second(),
expire_at := emqx_utils_calendar:epoch_second(),
%% TTL of individual segments
%% Somewhat confusing that we won't know it on the nodes where the filemeta
%% is missing.

2
apps/emqx_ft/src/emqx_ft_api.erl
View File

@ -278,7 +278,7 @@ format_file_info(
end.
format_timestamp(Timestamp) ->
iolist_to_binary(calendar:system_time_to_rfc3339(Timestamp, [{unit, second}])).
emqx_utils_calendar:epoch_to_rfc3339(Timestamp, second).
format_name(NameBin) when is_binary(NameBin) ->
NameBin;

2
apps/emqx_ft/src/emqx_ft_storage.erl
View File

@ -68,7 +68,7 @@
transfer := emqx_ft:transfer(),
name := file:name(),
size := _Bytes :: non_neg_integer(),
timestamp := emqx_datetime:epoch_second(),
timestamp := emqx_utils_calendar:epoch_second(),
uri => uri_string:uri_string(),
meta => emqx_ft:filemeta()
}.

2
apps/emqx_ft/src/emqx_ft_storage_exporter_s3.erl
View File

@ -43,7 +43,7 @@
transfer := transfer(),
name := file:name(),
uri := uri_string:uri_string(),
timestamp := emqx_datetime:epoch_second(),
timestamp := emqx_utils_calendar:epoch_second(),
size := _Bytes :: non_neg_integer(),
filemeta => filemeta()
}.

2
apps/emqx_ft/src/emqx_ft_storage_fs.erl
View File

@ -76,7 +76,7 @@
% TODO naming
-type filefrag(T) :: #{
path := file:name(),
timestamp := emqx_datetime:epoch_second(),
timestamp := emqx_utils_calendar:epoch_second(),
size := _Bytes :: non_neg_integer(),
fragment := T
}.

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

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_gateway, [
{description, "The Gateway management application"},
{vsn, "0.1.22"},
{vsn, "0.1.23"},
{registered, []},
{mod, {emqx_gateway_app, []}},
{applications, [kernel, stdlib, emqx, emqx_authn, emqx_ctl]},

20
apps/emqx_gateway/src/emqx_gateway_api_clients.erl
View File

@ -397,13 +397,13 @@ format_channel_info(WhichNode, {_, Infos, Stats} = R) ->
{ip_address, {peername, ConnInfo, fun peer_to_binary_addr/1}},
{port, {peername, ConnInfo, fun peer_to_port/1}},
{is_bridge, ClientInfo, false},
{connected_at, {connected_at, ConnInfo, fun emqx_gateway_utils:unix_ts_to_rfc3339/1}},
{disconnected_at, {disconnected_at, ConnInfo, fun emqx_gateway_utils:unix_ts_to_rfc3339/1}},
{connected_at, {connected_at, ConnInfo, fun emqx_utils_calendar:epoch_to_rfc3339/1}},
{disconnected_at, {disconnected_at, ConnInfo, fun emqx_utils_calendar:epoch_to_rfc3339/1}},
{connected, {conn_state, Infos, fun conn_state_to_connected/1}},
{keepalive, ClientInfo, 0},
{clean_start, ConnInfo, true},
{expiry_interval, ConnInfo, 0},
{created_at, {created_at, SessInfo, fun emqx_gateway_utils:unix_ts_to_rfc3339/1}},
{created_at, {created_at, SessInfo, fun emqx_utils_calendar:epoch_to_rfc3339/1}},
{subscriptions_cnt, Stats, 0},
{subscriptions_max, Stats, infinity},
{inflight_cnt, Stats, 0},
@ -640,28 +640,28 @@ params_client_searching_in_qs() ->
)},
{gte_created_at,
mk(
emqx_datetime:epoch_millisecond(),
emqx_utils_calendar:epoch_millisecond(),
M#{
desc => ?DESC(param_gte_created_at)
}
)},
{lte_created_at,
mk(
emqx_datetime:epoch_millisecond(),
emqx_utils_calendar:epoch_millisecond(),
M#{
desc => ?DESC(param_lte_created_at)
}
)},
{gte_connected_at,
mk(
emqx_datetime:epoch_millisecond(),
emqx_utils_calendar:epoch_millisecond(),
M#{
desc => ?DESC(param_gte_connected_at)
}
)},
{lte_connected_at,
mk(
emqx_datetime:epoch_millisecond(),
emqx_utils_calendar:epoch_millisecond(),
M#{
desc => ?DESC(param_lte_connected_at)
}
@ -888,12 +888,12 @@ common_client_props() ->
)},
{connected_at,
mk(
emqx_datetime:epoch_millisecond(),
emqx_utils_calendar:epoch_millisecond(),
#{desc => ?DESC(connected_at)}
)},
{disconnected_at,
mk(
emqx_datetime:epoch_millisecond(),
emqx_utils_calendar:epoch_millisecond(),
#{
desc => ?DESC(disconnected_at)
}
@ -931,7 +931,7 @@ common_client_props() ->
)},
{created_at,
mk(
emqx_datetime:epoch_millisecond(),
emqx_utils_calendar:epoch_millisecond(),
#{desc => ?DESC(created_at)}
)},
{subscriptions_cnt,

4
apps/emqx_gateway/src/emqx_gateway_cli.erl
View File

@ -313,9 +313,9 @@ format_gateway(
[
Name,
Status,
emqx_gateway_utils:unix_ts_to_rfc3339(CreatedAt),
emqx_utils_calendar:epoch_to_rfc3339(CreatedAt),
StopOrStart,
emqx_gateway_utils:unix_ts_to_rfc3339(Timestamp),
emqx_utils_calendar:epoch_to_rfc3339(Timestamp),
Config
]
).

7
apps/emqx_gateway/src/emqx_gateway_utils.erl
View File

@ -38,7 +38,6 @@
-export([
apply/2,
parse_listenon/1,
unix_ts_to_rfc3339/1,
unix_ts_to_rfc3339/2,
listener_id/3,
parse_listener_id/1,
@ -364,14 +363,10 @@ unix_ts_to_rfc3339(Key, Map) ->
Map;
Ts ->
Map#{
Key =>
emqx_rule_funcs:unix_ts_to_rfc3339(Ts, <<"millisecond">>)
Key => emqx_utils_calendar:epoch_to_rfc3339(Ts)
}
end.
unix_ts_to_rfc3339(Ts) ->
emqx_rule_funcs:unix_ts_to_rfc3339(Ts, <<"millisecond">>).
-spec stringfy(term()) -> binary().
stringfy(T) when is_list(T); is_binary(T) ->
iolist_to_binary(T);

4
apps/emqx_gateway_mqttsn/test/emqx_sn_protocol_SUITE.erl
View File

@ -2312,9 +2312,7 @@ t_socket_passvice(_) ->
ok.
t_clients_api(_) ->
TsNow = emqx_gateway_utils:unix_ts_to_rfc3339(
erlang:system_time(millisecond)
),
TsNow = emqx_utils_calendar:now_to_rfc3339(millisecond),
ClientId = <<"client_id_test1">>,
{ok, Socket} = gen_udp:open(0, [binary]),
send_connect_msg(Socket, ClientId),

2
apps/emqx_management/src/emqx_mgmt.erl
View File

@ -244,7 +244,7 @@ broker_info() ->
Info#{node => node(), otp_release => otp_rel(), node_status => 'running'}.
convert_broker_info({uptime, Uptime}, M) ->
M#{uptime => emqx_datetime:human_readable_duration_string(Uptime)};
M#{uptime => emqx_utils_calendar:human_readable_duration_string(Uptime)};
convert_broker_info({K, V}, M) ->
M#{K => iolist_to_binary(V)}.

4
apps/emqx_management/src/emqx_mgmt_api_api_keys.erl
View File

@ -127,7 +127,7 @@ fields(app) ->
)},
{expired_at,
hoconsc:mk(
hoconsc:union([infinity, emqx_datetime:epoch_second()]),
hoconsc:union([infinity, emqx_utils_calendar:epoch_second()]),
#{
desc => "No longer valid datetime",
example => <<"2021-12-05T02:01:34.186Z">>,
@ -137,7 +137,7 @@ fields(app) ->
)},
{created_at,
hoconsc:mk(
emqx_datetime:epoch_second(),
emqx_utils_calendar:epoch_second(),
#{
desc => "ApiKey create datetime",
example => <<"2021-12-01T00:00:00.000Z">>

4
apps/emqx_management/src/emqx_mgmt_api_banned.erl
View File

@ -147,13 +147,13 @@ fields(ban) ->
example => <<"Too many requests">>
})},
{at,
hoconsc:mk(emqx_datetime:epoch_second(), #{
hoconsc:mk(emqx_utils_calendar:epoch_second(), #{
desc => ?DESC(at),
required => false,
example => <<"2021-10-25T21:48:47+08:00">>
})},
{until,
hoconsc:mk(emqx_datetime:epoch_second(), #{
hoconsc:mk(emqx_utils_calendar:epoch_second(), #{
desc => ?DESC(until),
required => false,
example => <<"2021-10-25T21:53:47+08:00">>

16
apps/emqx_management/src/emqx_mgmt_api_clients.erl
View File

@ -161,7 +161,7 @@ schema("/clients") ->
desc => <<"Fuzzy search `username` as substring">>
})},
{gte_created_at,
hoconsc:mk(emqx_datetime:epoch_millisecond(), #{
hoconsc:mk(emqx_utils_calendar:epoch_millisecond(), #{
in => query,
required => false,
desc =>
@ -169,7 +169,7 @@ schema("/clients") ->
" than or equal method, rfc3339 or timestamp(millisecond)">>
})},
{lte_created_at,
hoconsc:mk(emqx_datetime:epoch_millisecond(), #{
hoconsc:mk(emqx_utils_calendar:epoch_millisecond(), #{
in => query,
required => false,
desc =>
@ -177,7 +177,7 @@ schema("/clients") ->
" than or equal method, rfc3339 or timestamp(millisecond)">>
})},
{gte_connected_at,
hoconsc:mk(emqx_datetime:epoch_millisecond(), #{
hoconsc:mk(emqx_utils_calendar:epoch_millisecond(), #{
in => query,
required => false,
desc => <<
@ -186,7 +186,7 @@ schema("/clients") ->
>>
})},
{lte_connected_at,
hoconsc:mk(emqx_datetime:epoch_millisecond(), #{
hoconsc:mk(emqx_utils_calendar:epoch_millisecond(), #{
in => query,
required => false,
desc => <<
@ -399,16 +399,16 @@ fields(client) ->
{connected, hoconsc:mk(boolean(), #{desc => <<"Whether the client is connected">>})},
{connected_at,
hoconsc:mk(
emqx_datetime:epoch_millisecond(),
emqx_utils_calendar:epoch_millisecond(),
#{desc => <<"Client connection time, rfc3339 or timestamp(millisecond)">>}
)},
{created_at,
hoconsc:mk(
emqx_datetime:epoch_millisecond(),
emqx_utils_calendar:epoch_millisecond(),
#{desc => <<"Session creation time, rfc3339 or timestamp(millisecond)">>}
)},
{disconnected_at,
hoconsc:mk(emqx_datetime:epoch_millisecond(), #{
hoconsc:mk(emqx_utils_calendar:epoch_millisecond(), #{
desc =>
<<
"Client offline time."
@ -950,7 +950,7 @@ result_format_time_fun(Key, NClientInfoMap) ->
case NClientInfoMap of
#{Key := TimeStamp} ->
NClientInfoMap#{
Key => emqx_datetime:epoch_to_rfc3339(TimeStamp)
Key => emqx_utils_calendar:epoch_to_rfc3339(TimeStamp)
};
#{} ->
NClientInfoMap

12
apps/emqx_management/src/emqx_mgmt_api_trace.erl
View File

@ -281,7 +281,7 @@ fields(trace) ->
})},
{start_at,
hoconsc:mk(
emqx_datetime:epoch_second(),
emqx_utils_calendar:epoch_second(),
#{
description => ?DESC(time_format),
required => false,
@ -290,7 +290,7 @@ fields(trace) ->
)},
{end_at,
hoconsc:mk(
emqx_datetime:epoch_second(),
emqx_utils_calendar:epoch_second(),
#{
description => ?DESC(time_format),
required => false,
@ -410,8 +410,8 @@ trace(get, _Params) ->
Trace0#{
log_size => LogSize,
Type => iolist_to_binary(Filter),
start_at => list_to_binary(calendar:system_time_to_rfc3339(Start)),
end_at => list_to_binary(calendar:system_time_to_rfc3339(End)),
start_at => emqx_utils_calendar:epoch_to_rfc3339(Start, second),
end_at => emqx_utils_calendar:epoch_to_rfc3339(End, second),
status => status(Enable, Start, End, Now)
}
end,
@ -468,8 +468,8 @@ format_trace(Trace0) ->
Trace2#{
log_size => LogSize,
Type => iolist_to_binary(Filter),
start_at => list_to_binary(calendar:system_time_to_rfc3339(Start)),
end_at => list_to_binary(calendar:system_time_to_rfc3339(End)),
start_at => emqx_utils_calendar:epoch_to_rfc3339(Start, second),
end_at => emqx_utils_calendar:epoch_to_rfc3339(Start, second),
status => status(Enable, Start, End, Now)
}.

4
apps/emqx_management/src/emqx_mgmt_auth.erl
View File

@ -142,11 +142,11 @@ format(App = #{expired_at := ExpiredAt0, created_at := CreateAt}) ->
ExpiredAt =
case ExpiredAt0 of
infinity -> <<"infinity">>;
_ -> list_to_binary(calendar:system_time_to_rfc3339(ExpiredAt0))
_ -> emqx_utils_calendar:epoch_to_rfc3339(ExpiredAt0, second)
end,
App#{
expired_at => ExpiredAt,
created_at => list_to_binary(calendar:system_time_to_rfc3339(CreateAt))
created_at => emqx_utils_calendar:epoch_to_rfc3339(CreateAt, second)
}.
list() ->

2
apps/emqx_management/src/emqx_mgmt_cli.erl
View File

@ -87,7 +87,7 @@ broker([]) ->
Funs = [sysdescr, version, datetime],
[emqx_ctl:print("~-10s: ~ts~n", [Fun, emqx_sys:Fun()]) || Fun <- Funs],
emqx_ctl:print("~-10s: ~ts~n", [
uptime, emqx_datetime:human_readable_duration_string(emqx_sys:uptime())
uptime, emqx_utils_calendar:human_readable_duration_string(emqx_sys:uptime())
]);
broker(["stats"]) ->
[

2
apps/emqx_management/test/emqx_mgmt_api_clients_SUITE.erl
View File

@ -260,7 +260,7 @@ t_query_clients_with_time(_) ->
%% Do not uri_encode `=` to `%3D`
Rfc3339String = emqx_http_lib:uri_encode(
binary:bin_to_list(
emqx_datetime:epoch_to_rfc3339(NowTimeStampInt)
emqx_utils_calendar:epoch_to_rfc3339(NowTimeStampInt)
)
),
TimeStampString = emqx_http_lib:uri_encode(integer_to_list(NowTimeStampInt)),

7
apps/emqx_modules/src/emqx_delayed.erl
View File

@ -208,8 +208,8 @@ format_delayed(
},
WithPayload
) ->
PublishTime = to_rfc3339(PublishTimeStamp div 1000),
ExpectTime = to_rfc3339(ExpectTimeStamp div 1000),
PublishTime = emqx_utils_calendar:epoch_to_rfc3339(PublishTimeStamp),
ExpectTime = emqx_utils_calendar:epoch_to_rfc3339(ExpectTimeStamp),
RemainingTime = ExpectTimeStamp - ?NOW,
Result = #{
msgid => emqx_guid:to_hexstr(Id),
@ -230,9 +230,6 @@ format_delayed(
Result
end.
to_rfc3339(Timestamp) ->
list_to_binary(calendar:system_time_to_rfc3339(Timestamp, [{unit, second}])).
-spec get_delayed_message(binary()) -> with_id_return(map()).
get_delayed_message(Id) ->
case ets:select(?TAB, ?QUERY_MS(Id)) of

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

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_modules, [
{description, "EMQX Modules"},
{vsn, "5.0.19"},
{vsn, "5.0.20"},
{modules, []},
{applications, [kernel, stdlib, emqx, emqx_ctl]},
{mod, {emqx_modules_app, []}},

2
apps/emqx_modules/src/emqx_topic_metrics.erl
View File

@ -295,7 +295,7 @@ terminate(_Reason, _State) ->
reset_topic({Topic, Data}, Speeds) ->
CRef = maps:get(counter_ref, Data),
ok = reset_counter(CRef),
ResetTime = emqx_rule_funcs:now_rfc3339(),
ResetTime = emqx_utils_calendar:now_to_rfc3339(),
true = ets:insert(?TAB, {Topic, Data#{reset_time => ResetTime}}),
Fun =
fun(Metric, CurrentSpeeds) ->

4
apps/emqx_modules/src/emqx_topic_metrics_api.erl
View File

@ -183,7 +183,7 @@ fields(topic_metrics) ->
)},
{create_time,
mk(
emqx_datetime:epoch_second(),
emqx_utils_calendar:epoch_second(),
#{
desc => ?DESC(create_time),
required => true,
@ -192,7 +192,7 @@ fields(topic_metrics) ->
)},
{reset_time,
mk(
emqx_datetime:epoch_second(),
emqx_utils_calendar:epoch_second(),
#{
desc => ?DESC(reset_time),
required => false,

7
apps/emqx_retainer/src/emqx_retainer_api.erl
View File

@ -211,11 +211,8 @@ format_message(#message{
msgid => emqx_guid:to_hexstr(ID),
qos => Qos,
topic => Topic,
publish_at => list_to_binary(
calendar:system_time_to_rfc3339(
Timestamp, [{unit, millisecond}]
)
),
publish_at =>
emqx_utils_calendar:epoch_to_rfc3339(Timestamp),
from_clientid => to_bin_string(From),
from_username => maps:get(username, Headers, <<>>)
}.

270
apps/emqx_rule_engine/src/emqx_rule_date.erl
View File

@ -1,270 +0,0 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2020-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_rule_date).
-export([date/3, date/4, parse_date/4]).
-export([
is_int_char/1,
is_symbol_char/1,
is_m_char/1
]).
-record(result, {
%%year()
year = "1970" :: string(),
%%month()
month = "1" :: string(),
%%day()
day = "1" :: string(),
%%hour()
hour = "0" :: string(),
%%minute() %% epoch in millisecond precision
minute = "0" :: string(),
%%second() %% epoch in millisecond precision
second = "0" :: string(),
%%integer() %% zone maybe some value
zone = "+00:00" :: string()
}).
%% -type time_unit() :: 'microsecond'
%% | 'millisecond'
%% | 'nanosecond'
%% | 'second'.
%% -type offset() :: [byte()] | (Time :: integer()).
date(TimeUnit, Offset, FormatString) ->
date(TimeUnit, Offset, FormatString, erlang:system_time(TimeUnit)).
date(TimeUnit, Offset, FormatString, TimeEpoch) ->
[Head | Other] = string:split(FormatString, "%", all),
R = create_tag([{st, Head}], Other),
Res = lists:map(
fun(Expr) ->
eval_tag(rmap(make_time(TimeUnit, Offset, TimeEpoch)), Expr)
end,
R
),
lists:concat(Res).
parse_date(TimeUnit, Offset, FormatString, InputString) ->
[Head | Other] = string:split(FormatString, "%", all),
R = create_tag([{st, Head}], Other),
IsZ = fun(V) ->
case V of
{tag, $Z} -> true;
_ -> false
end
end,
R1 = lists:filter(IsZ, R),
IfFun = fun(Con, A, B) ->
case Con of
[] -> A;
_ -> B
end
end,
Res = parse_input(FormatString, InputString),
Str =
Res#result.year ++ "-" ++
Res#result.month ++ "-" ++
Res#result.day ++ "T" ++
Res#result.hour ++ ":" ++
Res#result.minute ++ ":" ++
Res#result.second ++
IfFun(R1, Offset, Res#result.zone),
calendar:rfc3339_to_system_time(Str, [{unit, TimeUnit}]).
mlist(R) ->
%% %H Shows hour in 24-hour format [15]
[
{$H, R#result.hour},
%% %M Displays minutes [00-59]
{$M, R#result.minute},
%% %S Displays seconds [00-59]
{$S, R#result.second},
%% %y Displays year YYYY [2021]
{$y, R#result.year},
%% %m Displays the number of the month [01-12]
{$m, R#result.month},
%% %d Displays the number of the month [01-12]
{$d, R#result.day},
%% %Z Displays Time zone
{$Z, R#result.zone}
].
rmap(Result) ->
maps:from_list(mlist(Result)).
support_char() -> "HMSymdZ".
create_tag(Head, []) ->
Head;
create_tag(Head, [Val1 | RVal]) ->
case Val1 of
[] ->
create_tag(Head ++ [{st, [$%]}], RVal);
[H | Other] ->
case lists:member(H, support_char()) of
true -> create_tag(Head ++ [{tag, H}, {st, Other}], RVal);
false -> create_tag(Head ++ [{st, [$% | Val1]}], RVal)
end
end.
eval_tag(_, {st, Str}) ->
Str;
eval_tag(Map, {tag, Char}) ->
maps:get(Char, Map, "undefined").
%% make_time(TimeUnit, Offset) ->
%% make_time(TimeUnit, Offset, erlang:system_time(TimeUnit)).
make_time(TimeUnit, Offset, TimeEpoch) ->
Res = calendar:system_time_to_rfc3339(
TimeEpoch,
[{unit, TimeUnit}, {offset, Offset}]
),
[
Y1,
Y2,
Y3,
Y4,
$-,
Mon1,
Mon2,
$-,
D1,
D2,
_T,
H1,
H2,
$:,
Min1,
Min2,
$:,
S1,
S2
| TimeStr
] = Res,
IsFractionChar = fun(C) -> C >= $0 andalso C =< $9 orelse C =:= $. end,
{FractionStr, UtcOffset} = lists:splitwith(IsFractionChar, TimeStr),
#result{
year = [Y1, Y2, Y3, Y4],
month = [Mon1, Mon2],
day = [D1, D2],
hour = [H1, H2],
minute = [Min1, Min2],
second = [S1, S2] ++ FractionStr,
zone = UtcOffset
}.
is_int_char(C) ->
C >= $0 andalso C =< $9.
is_symbol_char(C) ->
C =:= $- orelse C =:= $+.
is_m_char(C) ->
C =:= $:.
parse_char_with_fun(_, []) ->
error(null_input);
parse_char_with_fun(ValidFun, [C | Other]) ->
Res =
case erlang:is_function(ValidFun) of
true -> ValidFun(C);
false -> erlang:apply(emqx_rule_date, ValidFun, [C])
end,
case Res of
true -> {C, Other};
false -> error({unexpected, [C | Other]})
end.
parse_string([], Input) ->
{[], Input};
parse_string([C | Other], Input) ->
{C1, Input1} = parse_char_with_fun(fun(V) -> V =:= C end, Input),
{Res, Input2} = parse_string(Other, Input1),
{[C1 | Res], Input2}.
parse_times(0, _, Input) ->
{[], Input};
parse_times(Times, Fun, Input) ->
{C1, Input1} = parse_char_with_fun(Fun, Input),
{Res, Input2} = parse_times((Times - 1), Fun, Input1),
{[C1 | Res], Input2}.
parse_int_times(Times, Input) ->
parse_times(Times, is_int_char, Input).
parse_fraction(Input) ->
IsFractionChar = fun(C) -> C >= $0 andalso C =< $9 orelse C =:= $. end,
lists:splitwith(IsFractionChar, Input).
parse_second(Input) ->
{M, Input1} = parse_int_times(2, Input),
{M1, Input2} = parse_fraction(Input1),
{M ++ M1, Input2}.
parse_zone(Input) ->
{S, Input1} = parse_char_with_fun(is_symbol_char, Input),
{M, Input2} = parse_int_times(2, Input1),
{C, Input3} = parse_char_with_fun(is_m_char, Input2),
{V, Input4} = parse_int_times(2, Input3),
{[S | M ++ [C | V]], Input4}.
mlist1() ->
maps:from_list(
%% %H Shows hour in 24-hour format [15]
[
{$H, fun(Input) -> parse_int_times(2, Input) end},
%% %M Displays minutes [00-59]
{$M, fun(Input) -> parse_int_times(2, Input) end},
%% %S Displays seconds [00-59]
{$S, fun(Input) -> parse_second(Input) end},
%% %y Displays year YYYY [2021]
{$y, fun(Input) -> parse_int_times(4, Input) end},
%% %m Displays the number of the month [01-12]
{$m, fun(Input) -> parse_int_times(2, Input) end},
%% %d Displays the number of the month [01-12]
{$d, fun(Input) -> parse_int_times(2, Input) end},
%% %Z Displays Time zone
{$Z, fun(Input) -> parse_zone(Input) end}
]
).
update_result($H, Res, Str) -> Res#result{hour = Str};
update_result($M, Res, Str) -> Res#result{minute = Str};
update_result($S, Res, Str) -> Res#result{second = Str};
update_result($y, Res, Str) -> Res#result{year = Str};
update_result($m, Res, Str) -> Res#result{month = Str};
update_result($d, Res, Str) -> Res#result{day = Str};
update_result($Z, Res, Str) -> Res#result{zone = Str}.
parse_tag(Res, {st, St}, InputString) ->
{_A, B} = parse_string(St, InputString),
{Res, B};
parse_tag(Res, {tag, St}, InputString) ->
Fun = maps:get(St, mlist1()),
{A, B} = Fun(InputString),
NRes = update_result(St, Res, A),
{NRes, B}.
parse_tags(Res, [], _) ->
Res;
parse_tags(Res, [Tag | Others], InputString) ->
{NRes, B} = parse_tag(Res, Tag, InputString),
parse_tags(NRes, Others, B).
parse_input(FormatString, InputString) ->
[Head | Other] = string:split(FormatString, "%", all),
R = create_tag([{st, Head}], Other),
parse_tags(#result{}, R, InputString).

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

@ -2,7 +2,7 @@
{application, emqx_rule_engine, [
{description, "EMQX Rule Engine"},
% strict semver, bump manually!
{vsn, "5.0.22"},
{vsn, "5.0.23"},
{modules, []},
{registered, [emqx_rule_engine_sup, emqx_rule_engine]},
{applications, [kernel, stdlib, rulesql, getopt, emqx_ctl, uuid]},

2
apps/emqx_rule_engine/src/emqx_rule_engine_api.erl
View File

@ -514,7 +514,7 @@ format_rule_engine_resp(Config) ->
maps:remove(rules, Config).
format_datetime(Timestamp, Unit) ->
list_to_binary(calendar:system_time_to_rfc3339(Timestamp, [{unit, Unit}])).
emqx_utils_calendar:epoch_to_rfc3339(Timestamp, Unit).
format_action(Actions) ->
[do_format_action(Act) || Act <- Actions].

4
apps/emqx_rule_engine/src/emqx_rule_engine_cli.erl
View File

@ -74,8 +74,8 @@ pretty_print_rule(ID) ->
"Updated at:\n ~ts\n"
"Actions:\n ~s\n"
,[Id, Name, left_pad(Descr), Enable, left_pad(SQL),
calendar:system_time_to_rfc3339(CreatedAt, [{unit, millisecond}]),
calendar:system_time_to_rfc3339(UpdatedAt, [{unit, millisecond}]),
emqx_utils_calendar:epoch_to_rfc3339(CreatedAt, millisecond),
emqx_utils_calendar:epoch_to_rfc3339(UpdatedAt, millisecond),
[left_pad(format_action(A)) || A <- Actions]
]
);

33
apps/emqx_rule_engine/src/emqx_rule_funcs.erl
View File

@ -276,6 +276,8 @@
]}
).
-import(emqx_utils_calendar, [time_unit/1, now_to_rfc3339/0, now_to_rfc3339/1, epoch_to_rfc3339/2]).
%% @doc "msgid()" Func
msgid() ->
fun
@ -1077,23 +1079,19 @@ kv_store_del(Key) ->
%%--------------------------------------------------------------------
now_rfc3339() ->
now_rfc3339(<<"second">>).
now_to_rfc3339().
now_rfc3339(Unit) ->
unix_ts_to_rfc3339(now_timestamp(Unit), Unit).
now_to_rfc3339(time_unit(Unit)).
unix_ts_to_rfc3339(Epoch) ->
unix_ts_to_rfc3339(Epoch, <<"second">>).
epoch_to_rfc3339(Epoch, second).
unix_ts_to_rfc3339(Epoch, Unit) when is_integer(Epoch) ->
emqx_utils_conv:bin(
calendar:system_time_to_rfc3339(
Epoch, [{unit, time_unit(Unit)}]
)
).
epoch_to_rfc3339(Epoch, time_unit(Unit)).
rfc3339_to_unix_ts(DateTime) ->
rfc3339_to_unix_ts(DateTime, <<"second">>).
rfc3339_to_unix_ts(DateTime, second).
rfc3339_to_unix_ts(DateTime, Unit) when is_binary(DateTime) ->
calendar:rfc3339_to_system_time(
@ -1107,15 +1105,6 @@ now_timestamp() ->
now_timestamp(Unit) ->
erlang:system_time(time_unit(Unit)).
time_unit(<<"second">>) -> second;
time_unit(<<"millisecond">>) -> millisecond;
time_unit(<<"microsecond">>) -> microsecond;
time_unit(<<"nanosecond">>) -> nanosecond;
time_unit(second) -> second;
time_unit(millisecond) -> millisecond;
time_unit(microsecond) -> microsecond;
time_unit(nanosecond) -> nanosecond.
format_date(TimeUnit, Offset, FormatString) ->
Unit = time_unit(TimeUnit),
TimeEpoch = erlang:system_time(Unit),
@ -1125,17 +1114,17 @@ format_date(TimeUnit, Offset, FormatString, TimeEpoch) ->
Unit = time_unit(TimeUnit),
emqx_utils_conv:bin(
lists:concat(
emqx_calendar:format(TimeEpoch, Unit, Offset, FormatString)
emqx_utils_calendar:format(TimeEpoch, Unit, Offset, FormatString)
)
).
date_to_unix_ts(TimeUnit, FormatString, InputString) ->
Unit = time_unit(TimeUnit),
emqx_calendar:parse(InputString, Unit, FormatString).
emqx_utils_calendar:parse(InputString, Unit, FormatString).
date_to_unix_ts(TimeUnit, Offset, FormatString, InputString) ->
Unit = time_unit(TimeUnit),
OffsetSecond = emqx_calendar:offset_second(Offset),
OffsetSecond = emqx_utils_calendar:offset_second(Offset),
OffsetDelta = erlang:convert_time_unit(OffsetSecond, second, Unit),
date_to_unix_ts(Unit, FormatString, InputString) - OffsetDelta.
@ -1143,7 +1132,7 @@ timezone_to_second(TimeZone) ->
timezone_to_offset_seconds(TimeZone).
timezone_to_offset_seconds(TimeZone) ->
emqx_calendar:offset_second(TimeZone).
emqx_utils_calendar:offset_second(TimeZone).
'$handle_undefined_function'(sprintf, [Format | Args]) ->
erlang:apply(fun sprintf_s/2, [Format, Args]);

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

@ -2,7 +2,7 @@
{application, emqx_utils, [
{description, "Miscellaneous utilities for EMQX apps"},
% strict semver, bump manually!
{vsn, "5.0.6"},
{vsn, "5.0.7"},
{modules, [
emqx_utils,
emqx_utils_api,

199
apps/emqx/src/emqx_calendar.erl → apps/emqx_utils/src/emqx_utils_calendar.erl
View File

@ -1,5 +1,5 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2019-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
@ -14,7 +14,32 @@
%% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_calendar).
-module(emqx_utils_calendar).
-include_lib("typerefl/include/types.hrl").
-export([
formatter/1,
format/3,
format/4,
parse/3,
offset_second/1
]).
%% API
-export([
to_epoch_millisecond/1,
to_epoch_second/1,
human_readable_duration_string/1
]).
-export([
epoch_to_rfc3339/1,
epoch_to_rfc3339/2,
now_to_rfc3339/0,
now_to_rfc3339/1
]).
-export([time_unit/1]).
-define(SECONDS_PER_MINUTE, 60).
-define(SECONDS_PER_HOUR, 3600).
@ -24,13 +49,11 @@
-define(DAYS_FROM_0_TO_1970, 719528).
-define(SECONDS_FROM_0_TO_1970, (?DAYS_FROM_0_TO_1970 * ?SECONDS_PER_DAY)).
-export([
formatter/1,
format/3,
format/4,
parse/3,
offset_second/1
]).
%% the maximum value is the SECONDS_FROM_0_TO_10000 in the calendar.erl,
%% here minus SECONDS_PER_DAY to tolerate timezone time offset,
%% so the maximum date can reach 9999-12-31 which is ample.
-define(MAXIMUM_EPOCH, 253402214400).
-define(MAXIMUM_EPOCH_MILLI, 253402214400_000).
-define(DATE_PART, [
year,
@ -50,6 +73,72 @@
timezone2
]).
-reflect_type([
epoch_millisecond/0,
epoch_second/0
]).
-type epoch_second() :: non_neg_integer().
-type epoch_millisecond() :: non_neg_integer().
-typerefl_from_string({epoch_second/0, ?MODULE, to_epoch_second}).
-typerefl_from_string({epoch_millisecond/0, ?MODULE, to_epoch_millisecond}).
%%--------------------------------------------------------------------
%% Epoch <-> RFC 3339
%%--------------------------------------------------------------------
to_epoch_second(DateTime) ->
to_epoch(DateTime, second).
to_epoch_millisecond(DateTime) ->
to_epoch(DateTime, millisecond).
to_epoch(DateTime, Unit) ->
try
case string:to_integer(DateTime) of
{Epoch, []} -> validate_epoch(Epoch, Unit);
_ -> {ok, calendar:rfc3339_to_system_time(DateTime, [{unit, Unit}])}
end
catch
error:_ ->
{error, bad_rfc3339_timestamp}
end.
epoch_to_rfc3339(Timestamp) ->
epoch_to_rfc3339(Timestamp, millisecond).
epoch_to_rfc3339(Timestamp, Unit) when is_integer(Timestamp) ->
list_to_binary(calendar:system_time_to_rfc3339(Timestamp, [{unit, Unit}])).
now_to_rfc3339() ->
now_to_rfc3339(second).
now_to_rfc3339(Unit) ->
epoch_to_rfc3339(erlang:system_time(Unit), Unit).
-spec human_readable_duration_string(integer()) -> string().
human_readable_duration_string(Milliseconds) ->
Seconds = Milliseconds div 1000,
{D, {H, M, S}} = calendar:seconds_to_daystime(Seconds),
L0 = [{D, " days"}, {H, " hours"}, {M, " minutes"}, {S, " seconds"}],
L1 = lists:dropwhile(fun({K, _}) -> K =:= 0 end, L0),
L2 = lists:map(fun({Time, Unit}) -> [integer_to_list(Time), Unit] end, L1),
lists:flatten(lists:join(", ", L2)).
validate_epoch(Epoch, _Unit) when Epoch < 0 ->
{error, bad_epoch};
validate_epoch(Epoch, second) when Epoch =< ?MAXIMUM_EPOCH ->
{ok, Epoch};
validate_epoch(Epoch, millisecond) when Epoch =< ?MAXIMUM_EPOCH_MILLI ->
{ok, Epoch};
validate_epoch(_Epoch, _Unit) ->
{error, bad_epoch}.
%%--------------------------------------------------------------------
%% Timestamp <-> any format date string
%% Timestamp treat as a superset for epoch, it can be any positive integer
%%--------------------------------------------------------------------
formatter(FormatterStr) when is_list(FormatterStr) ->
formatter(list_to_binary(FormatterStr));
formatter(FormatterBin) when is_binary(FormatterBin) ->
@ -70,8 +159,10 @@ parse(DateStr, Unit, FormatterBin) when is_binary(FormatterBin) ->
parse(DateStr, Unit, formatter(FormatterBin));
parse(DateStr, Unit, Formatter) ->
do_parse(DateStr, Unit, Formatter).
%% -------------------------------------------------------------------------------------------------
%% internal
%%--------------------------------------------------------------------
%% Time unit
%%--------------------------------------------------------------------
time_unit(second) -> second;
time_unit(millisecond) -> millisecond;
@ -84,10 +175,12 @@ time_unit("nanosecond") -> nanosecond;
time_unit(<<"second">>) -> second;
time_unit(<<"millisecond">>) -> millisecond;
time_unit(<<"microsecond">>) -> microsecond;
time_unit(<<"nanosecond">>) -> nanosecond.
time_unit(<<"nanosecond">>) -> nanosecond;
time_unit(Any) -> error({invalid_time_unit, Any}).
%% -------------------------------------------------------------------------------------------------
%%--------------------------------------------------------------------
%% internal: format part
%%--------------------------------------------------------------------
do_formatter(<<>>, Formatter) ->
lists:reverse(Formatter);
@ -357,9 +450,9 @@ padding(Data, Len) when Len > 0 andalso erlang:length(Data) < Len ->
padding(Data, _Len) ->
Data.
%% -------------------------------------------------------------------------------------------------
%% internal
%% parse part
%%--------------------------------------------------------------------
%% internal: parse part
%%--------------------------------------------------------------------
do_parse(DateStr, Unit, Formatter) ->
DateInfo = do_parse_date_str(DateStr, Formatter, #{}),
@ -476,3 +569,77 @@ str_to_int_or_error(Str, Error) ->
_ ->
error(Error)
end.
%%--------------------------------------------------------------------
%% Unit Test
%%--------------------------------------------------------------------
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-compile(nowarn_export_all).
-compile(export_all).
roots() -> [bar].
fields(bar) ->
[
{second, ?MODULE:epoch_second()},
{millisecond, ?MODULE:epoch_millisecond()}
].
-define(FORMAT(_Sec_, _Ms_),
lists:flatten(
io_lib:format("bar={second=~w,millisecond=~w}", [_Sec_, _Ms_])
)
).
epoch_ok_test() ->
BigStamp = 1 bsl 37,
Args = [
{0, 0, 0, 0},
{1, 1, 1, 1},
{BigStamp, BigStamp * 1000, BigStamp, BigStamp * 1000},
{"2022-01-01T08:00:00+08:00", "2022-01-01T08:00:00+08:00", 1640995200, 1640995200000}
],
lists:foreach(
fun({Sec, Ms, EpochSec, EpochMs}) ->
check_ok(?FORMAT(Sec, Ms), EpochSec, EpochMs)
end,
Args
),
ok.
check_ok(Input, Sec, Ms) ->
{ok, Data} = hocon:binary(Input, #{}),
?assertMatch(
#{bar := #{second := Sec, millisecond := Ms}},
hocon_tconf:check_plain(?MODULE, Data, #{atom_key => true}, [bar])
),
ok.
epoch_failed_test() ->
BigStamp = 1 bsl 38,
Args = [
{-1, -1},
{"1s", "1s"},
{BigStamp, 0},
{0, BigStamp * 1000},
{"2022-13-13T08:00:00+08:00", "2022-13-13T08:00:00+08:00"}
],
lists:foreach(
fun({Sec, Ms}) ->
check_failed(?FORMAT(Sec, Ms))
end,
Args
),
ok.
check_failed(Input) ->
{ok, Data} = hocon:binary(Input, #{}),
?assertException(
throw,
_,
hocon_tconf:check_plain(?MODULE, Data, #{atom_key => true}, [bar])
),
ok.
-endif.

1
changes/ce/feat-11446.en.md Normal file
View File

@ -0,0 +1 @@
Refactored datetime-related modules and functions to simplify the code.