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

Update copyright

This commit is contained in:
Feng Lee 2019-06-18 14:27:06 +08:00
parent d065847665
commit 21162f88b8
77 changed files with 1118 additions and 1563 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/types.hrl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-type(maybe(T) :: undefined | T). -type(maybe(T) :: undefined | T).

33
src/emqx.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx). -module(emqx).
@ -59,9 +61,13 @@
-define(APP, ?MODULE). -define(APP, ?MODULE).
%%------------------------------------------------------------------------------ -define(COPYRIGHT, "Copyright (c) 2019 EMQ Technologies Co., Ltd").
-define(LICENSE_MESSAGE, "Licensed under the Apache License, Version 2.0").
%%--------------------------------------------------------------------
%% Bootstrap, is_running... %% Bootstrap, is_running...
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Start emqx application %% @doc Start emqx application
-spec(start() -> {ok, list(atom())} | {error, term()}). -spec(start() -> {ok, list(atom())} | {error, term()}).
@ -93,9 +99,9 @@ is_running(Node) ->
Pid when is_pid(Pid) -> true Pid when is_pid(Pid) -> true
end. end.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% PubSub API %% PubSub API
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(subscribe(emqx_topic:topic() | string()) -> ok). -spec(subscribe(emqx_topic:topic() | string()) -> ok).
subscribe(Topic) -> subscribe(Topic) ->
@ -120,9 +126,9 @@ publish(Msg) ->
unsubscribe(Topic) -> unsubscribe(Topic) ->
emqx_broker:unsubscribe(iolist_to_binary(Topic)). emqx_broker:unsubscribe(iolist_to_binary(Topic)).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% PubSub management API %% PubSub management API
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(topics() -> list(emqx_topic:topic())). -spec(topics() -> list(emqx_topic:topic())).
topics() -> emqx_router:topics(). topics() -> emqx_router:topics().
@ -141,9 +147,9 @@ subscribed(SubPid, Topic) when is_pid(SubPid) ->
subscribed(SubId, Topic) when is_atom(SubId); is_binary(SubId) -> subscribed(SubId, Topic) when is_atom(SubId); is_binary(SubId) ->
emqx_broker:subscribed(SubId, iolist_to_binary(Topic)). emqx_broker:subscribed(SubId, iolist_to_binary(Topic)).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Hooks API %% Hooks API
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(hook(emqx_hooks:hookpoint(), emqx_hooks:action()) -> ok | {error, already_exists}). -spec(hook(emqx_hooks:hookpoint(), emqx_hooks:action()) -> ok | {error, already_exists}).
hook(HookPoint, Action) -> hook(HookPoint, Action) ->
@ -175,9 +181,9 @@ run_hook(HookPoint, Args) ->
run_fold_hook(HookPoint, Args, Acc) -> run_fold_hook(HookPoint, Args, Acc) ->
emqx_hooks:run_fold(HookPoint, Args, Acc). emqx_hooks:run_fold(HookPoint, Args, Acc).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Shutdown and reboot %% Shutdown and reboot
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
shutdown() -> shutdown() ->
shutdown(normal). shutdown(normal).
@ -191,12 +197,13 @@ shutdown(Reason) ->
reboot() -> reboot() ->
lists:foreach(fun application:start/1, [gproc, esockd, ranch, cowboy, ekka, emqx]). lists:foreach(fun application:start/1, [gproc, esockd, ranch, cowboy, ekka, emqx]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
reload_config(ConfFile) -> reload_config(ConfFile) ->
{ok, [Conf]} = file:consult(ConfFile), {ok, [Conf]} = file:consult(ConfFile),
lists:foreach(fun({App, Vals}) -> lists:foreach(fun({App, Vals}) ->
[application:set_env(App, Par, Val) || {Par, Val} <- Vals] [application:set_env(App, Par, Val) || {Par, Val} <- Vals]
end, Conf). end, Conf).

15
src/emqx_access_control.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_access_control). -module(emqx_access_control).
@ -22,9 +24,10 @@
, reload_acl/0 , reload_acl/0
]). ]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(authenticate(emqx_types:credentials()) -spec(authenticate(emqx_types:credentials())
-> {ok, emqx_types:credentials()} | {error, term()}). -> {ok, emqx_types:credentials()} | {error, term()}).
authenticate(Credentials) -> authenticate(Credentials) ->
@ -36,7 +39,8 @@ authenticate(Credentials) ->
end. end.
%% @doc Check ACL %% @doc Check ACL
-spec(check_acl(emqx_types:credentials(), emqx_types:pubsub(), emqx_types:topic()) -> allow | deny). -spec(check_acl(emqx_types:credentials(), emqx_types:pubsub(), emqx_types:topic())
-> allow | deny).
check_acl(Credentials, PubSub, Topic) -> check_acl(Credentials, PubSub, Topic) ->
case emqx_acl_cache:is_enabled() of case emqx_acl_cache:is_enabled() of
false -> false ->
@ -47,8 +51,7 @@ check_acl(Credentials, PubSub, Topic) ->
AclResult = do_check_acl(Credentials, PubSub, Topic), AclResult = do_check_acl(Credentials, PubSub, Topic),
emqx_acl_cache:put_acl_cache(PubSub, Topic, AclResult), emqx_acl_cache:put_acl_cache(PubSub, Topic, AclResult),
AclResult; AclResult;
AclResult -> AclResult -> AclResult
AclResult
end end
end. end.

8
src/emqx_access_rule.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_access_rule). -module(emqx_access_rule).
@ -38,9 +40,9 @@
-define(ALLOW_DENY(A), ((A =:= allow) orelse (A =:= deny))). -define(ALLOW_DENY(A), ((A =:= allow) orelse (A =:= deny))).
-define(PUBSUB(A), ((A =:= subscribe) orelse (A =:= publish) orelse (A =:= pubsub))). -define(PUBSUB(A), ((A =:= subscribe) orelse (A =:= publish) orelse (A =:= pubsub))).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Compile Access Rule. %% @doc Compile Access Rule.
compile({A, all}) when ?ALLOW_DENY(A) -> compile({A, all}) when ?ALLOW_DENY(A) ->

5
src/emqx_acl_cache.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_acl_cache). -module(emqx_acl_cache).
@ -124,6 +126,7 @@ map_acl_cache(Fun) ->
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
add_acl(PubSub, Topic, AclResult) -> add_acl(PubSub, Topic, AclResult) ->
K = cache_k(PubSub, Topic), K = cache_k(PubSub, Topic),
V = cache_v(AclResult), V = cache_v(AclResult),

22
src/emqx_alarm_handler.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_alarm_handler). -module(emqx_alarm_handler).
@ -44,9 +46,9 @@
-define(ALARM_TAB, emqx_alarm). -define(ALARM_TAB, emqx_alarm).
-define(ALARM_HISTORY_TAB, emqx_alarm_history). -define(ALARM_HISTORY_TAB, emqx_alarm_history).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Mnesia bootstrap %% Mnesia bootstrap
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
mnesia(boot) -> mnesia(boot) ->
ok = ekka_mnesia:create_table(?ALARM_TAB, [ ok = ekka_mnesia:create_table(?ALARM_TAB, [
@ -61,13 +63,14 @@ mnesia(boot) ->
{local_content, true}, {local_content, true},
{record_name, alarm_history}, {record_name, alarm_history},
{attributes, record_info(fields, alarm_history)}]); {attributes, record_info(fields, alarm_history)}]);
mnesia(copy) -> mnesia(copy) ->
ok = ekka_mnesia:copy_table(?ALARM_TAB), ok = ekka_mnesia:copy_table(?ALARM_TAB),
ok = ekka_mnesia:copy_table(?ALARM_HISTORY_TAB). ok = ekka_mnesia:copy_table(?ALARM_HISTORY_TAB).
%%---------------------------------------------------------------------- %%--------------------------------------------------------------------
%% API %% API
%%---------------------------------------------------------------------- %%--------------------------------------------------------------------
load() -> load() ->
gen_event:swap_handler(alarm_handler, {alarm_handler, swap}, {?MODULE, []}). gen_event:swap_handler(alarm_handler, {alarm_handler, swap}, {?MODULE, []}).
@ -79,13 +82,14 @@ unload() ->
get_alarms() -> get_alarms() ->
gen_event:call(alarm_handler, ?MODULE, get_alarms). gen_event:call(alarm_handler, ?MODULE, get_alarms).
%%---------------------------------------------------------------------- %%--------------------------------------------------------------------
%% gen_event callbacks %% gen_event callbacks
%%---------------------------------------------------------------------- %%--------------------------------------------------------------------
init({_Args, {alarm_handler, ExistingAlarms}}) -> init({_Args, {alarm_handler, ExistingAlarms}}) ->
init_tables(ExistingAlarms), init_tables(ExistingAlarms),
{ok, []}; {ok, []};
init(_) -> init(_) ->
init_tables([]), init_tables([]),
{ok, []}. {ok, []}.
@ -171,6 +175,6 @@ get_alarms_() ->
[{Id, Desc} || #common_alarm{id = Id, desc = Desc} <- Alarms]. [{Id, Desc} || #common_alarm{id = Id, desc = Desc} <- Alarms].
set_alarm_history(Id) -> set_alarm_history(Id) ->
mnesia:dirty_write(?ALARM_HISTORY_TAB, #alarm_history{id = Id, His = #alarm_history{id = Id, clear_at = undefined},
clear_at = undefined}). mnesia:dirty_write(?ALARM_HISTORY_TAB, His).

10
src/emqx_app.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_app). -module(emqx_app).
@ -30,10 +32,10 @@ start(_Type, _Args) ->
print_banner(), print_banner(),
ekka:start(), ekka:start(),
{ok, Sup} = emqx_sup:start_link(), {ok, Sup} = emqx_sup:start_link(),
emqx_modules:load(), ok = emqx_modules:load(),
emqx_plugins:init(), ok = emqx_plugins:init(),
emqx_plugins:load(), emqx_plugins:load(),
emqx_listeners:start(), ok = emqx_listeners:start(),
start_autocluster(), start_autocluster(),
register(emqx, self()), register(emqx, self()),

12
src/emqx_base62.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_base62). -module(emqx_base62).
@ -21,9 +23,9 @@
, decode/2 , decode/2
]). ]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Encode any data to base62 binary %% @doc Encode any data to base62 binary
-spec encode(string() | integer() | binary()) -> binary(). -spec encode(string() | integer() | binary()) -> binary().
@ -43,9 +45,9 @@ decode(L) when is_list(L) ->
decode(B) when is_binary(B) -> decode(B) when is_binary(B) ->
decode(B, <<>>). decode(B, <<>>).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Interval Functions %% Interval Functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
encode(D, string) -> encode(D, string) ->
binary_to_list(encode(D)); binary_to_list(encode(D));

39
src/emqx_batch.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_batch). -module(emqx_batch).
@ -22,18 +24,18 @@
, items/1 , items/1
]). ]).
-record(batch, -record(batch, {
{ batch_size :: non_neg_integer() batch_size :: non_neg_integer(),
, batch_q :: list(any()) batch_q :: list(any()),
, linger_ms :: pos_integer() linger_ms :: pos_integer(),
, linger_timer :: reference() | undefined linger_timer :: reference() | undefined,
, commit_fun :: function() commit_fun :: function()
}). }).
-type(options() :: -type(options() :: #{
#{ batch_size => non_neg_integer() batch_size => non_neg_integer(),
, linger_ms => pos_integer() linger_ms => pos_integer(),
, commit_fun := function() commit_fun := function()
}). }).
-opaque(batch() :: #batch{}). -opaque(batch() :: #batch{}).
@ -42,9 +44,9 @@
-export_type([batch/0]). -export_type([batch/0]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(init(options()) -> batch()). -spec(init(options()) -> batch()).
init(Opts) when is_map(Opts) -> init(Opts) when is_map(Opts) ->
@ -54,14 +56,19 @@ init(Opts) when is_map(Opts) ->
commit_fun = maps:get(commit_fun, Opts)}. commit_fun = maps:get(commit_fun, Opts)}.
-spec(push(any(), batch()) -> batch()). -spec(push(any(), batch()) -> batch()).
push(El, Batch = #batch{batch_q = Q, linger_ms = Ms, linger_timer = undefined}) when length(Q) == 0 -> push(El, Batch = #batch{batch_q = Q,
Batch#batch{batch_q = [El], linger_timer = erlang:send_after(Ms, self(), batch_linger_expired)}; linger_ms = Ms,
linger_timer = undefined})
when length(Q) == 0 ->
TRef = erlang:send_after(Ms, self(), batch_linger_expired),
Batch#batch{batch_q = [El], linger_timer = TRef};
%% no limit. %% no limit.
push(El, Batch = #batch{batch_size = 0, batch_q = Q}) -> push(El, Batch = #batch{batch_size = 0, batch_q = Q}) ->
Batch#batch{batch_q = [El|Q]}; Batch#batch{batch_q = [El|Q]};
push(El, Batch = #batch{batch_size = MaxSize, batch_q = Q}) when length(Q) >= MaxSize -> push(El, Batch = #batch{batch_size = MaxSize, batch_q = Q})
when length(Q) >= MaxSize ->
commit(Batch#batch{batch_q = [El|Q]}); commit(Batch#batch{batch_q = [El|Q]});
push(El, Batch = #batch{batch_q = Q}) -> push(El, Batch = #batch{batch_q = Q}) ->

5
src/emqx_bridge.erl
View File

@ -1,3 +1,4 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved. %% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
@ -11,7 +12,9 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%%--------------------------------------------------------------------
%% @doc Bridge works in two layers (1) batching layer (2) transport layer %% @doc Bridge works in two layers (1) batching layer (2) transport layer
%% The `bridge' batching layer collects local messages in batches and sends over %% The `bridge' batching layer collects local messages in batches and sends over
%% to remote MQTT node/cluster via `connetion' transport layer. %% to remote MQTT node/cluster via `connetion' transport layer.
@ -56,8 +59,10 @@
%% %%
%% NOTES: %% NOTES:
%% * Local messages are all normalised to QoS-1 when exporting to remote %% * Local messages are all normalised to QoS-1 when exporting to remote
%%--------------------------------------------------------------------
-module(emqx_bridge). -module(emqx_bridge).
-behaviour(gen_statem). -behaviour(gen_statem).
%% APIs %% APIs

4
src/emqx_bridge_connect.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_bridge_connect). -module(emqx_bridge_connect).

7
src/emqx_bridge_mqtt.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,8 +12,10 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc This module implements EMQX Bridge transport layer on top of MQTT protocol %% @doc This module implements EMQX Bridge transport layer on top of
%% MQTT protocol.
-module(emqx_bridge_mqtt). -module(emqx_bridge_mqtt).

4
src/emqx_bridge_msg.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_bridge_msg). -module(emqx_bridge_msg).

5
src/emqx_bridge_rpc.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,10 +12,12 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc This module implements EMQX Bridge transport layer based on gen_rpc. %% @doc This module implements EMQX Bridge transport layer based on gen_rpc.
-module(emqx_bridge_rpc). -module(emqx_bridge_rpc).
-behaviour(emqx_bridge_connect). -behaviour(emqx_bridge_connect).
%% behaviour callbacks %% behaviour callbacks

5
src/emqx_bridge_sup.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,8 +12,10 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_bridge_sup). -module(emqx_bridge_sup).
-behavior(supervisor). -behavior(supervisor).
-include("logger.hrl"). -include("logger.hrl").

4
src/emqx_broker.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_broker). -module(emqx_broker).

12
src/emqx_broker_helper.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_broker_helper). -module(emqx_broker_helper).
@ -90,9 +92,9 @@ create_seq(Topic) ->
reclaim_seq(Topic) -> reclaim_seq(Topic) ->
emqx_sequence:reclaim(?SUBSEQ, Topic). emqx_sequence:reclaim(?SUBSEQ, Topic).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([]) -> init([]) ->
%% Helper table %% Helper table
@ -140,9 +142,9 @@ terminate(_Reason, _State) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
clean_down(SubPid) -> clean_down(SubPid) ->
case ets:lookup(?SUBMON, SubPid) of case ets:lookup(?SUBMON, SubPid) of

8
src/emqx_broker_sup.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_broker_sup). -module(emqx_broker_sup).
@ -23,9 +25,9 @@
start_link() -> start_link() ->
supervisor:start_link({local, ?MODULE}, ?MODULE, []). supervisor:start_link({local, ?MODULE}, ?MODULE, []).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Supervisor callbacks %% Supervisor callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([]) -> init([]) ->
%% Broker pool %% Broker pool

5
src/emqx_cli.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_cli). -module(emqx_cli).
@ -35,3 +37,4 @@ usage(CmdList) ->
usage(Format, Args) -> usage(Format, Args) ->
usage([{Format, Args}]). usage([{Format, Args}]).

5
src/emqx_client_sock.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_client_sock). -module(emqx_client_sock).
@ -105,3 +107,4 @@ default_ciphers(TlsVersions) ->
fun(TlsVer, Ciphers) -> fun(TlsVer, Ciphers) ->
Ciphers ++ ssl:cipher_suites(all, TlsVer) Ciphers ++ ssl:cipher_suites(all, TlsVer)
end, [], TlsVersions). end, [], TlsVersions).

4
src/emqx_config.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc Hot Configuration %% @doc Hot Configuration
%% %%

4
src/emqx_ctl.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_ctl). -module(emqx_ctl).

12
src/emqx_gc.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,9 +12,10 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc This module manages an opaque collection of statistics data used to %% @doc This module manages an opaque collection of statistics data used
%% force garbage collection on `self()' process when hitting thresholds. %% to force garbage collection on `self()' process when hitting thresholds.
%% Namely: %% Namely:
%% (1) Total number of messages passed through %% (1) Total number of messages passed through
%% (2) Total data volume passed through %% (2) Total data volume passed through
@ -85,9 +87,9 @@ reset(?GCS(St)) ->
reset(undefined) -> reset(undefined) ->
undefined. undefined.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(dec(cnt | oct, pos_integer(), st()) -> {boolean(), st()}). -spec(dec(cnt | oct, pos_integer(), st()) -> {boolean(), st()}).
dec(Key, Num, St) -> dec(Key, Num, St) ->

4
src/emqx_gen_mod.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_gen_mod). -module(emqx_gen_mod).

4
src/emqx_guid.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc Generate global unique id for mqtt message. %% @doc Generate global unique id for mqtt message.
%% %%

11
src/emqx_inflight.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_inflight). -module(emqx_inflight).
@ -38,17 +40,18 @@
-opaque(inflight() :: {?MODULE, max_size(), gb_trees:tree()}). -opaque(inflight() :: {?MODULE, max_size(), gb_trees:tree()}).
-define(Inflight(Tree), {?MODULE, _MaxSize, Tree}). -define(Inflight(Tree), {?MODULE, _MaxSize, Tree}).
-define(Inflight(MaxSize, Tree), {?MODULE, MaxSize, (Tree)}). -define(Inflight(MaxSize, Tree), {?MODULE, MaxSize, (Tree)}).
-export_type([inflight/0]). -export_type([inflight/0]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(new(non_neg_integer()) -> inflight()). -spec(new(non_neg_integer()) -> inflight()).
new(MaxSize) when MaxSize >= 0 -> new(MaxSize) when MaxSize >= 0 ->
{?MODULE, MaxSize, gb_trees:empty()}. ?Inflight(MaxSize, gb_trees:empty()).
-spec(contain(key(), inflight()) -> boolean()). -spec(contain(key(), inflight()) -> boolean()).
contain(Key, ?Inflight(Tree)) -> contain(Key, ?Inflight(Tree)) ->

12
src/emqx_json.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_json). -module(emqx_json).
@ -35,12 +37,12 @@ encode(Term, Opts) ->
jsx:encode(Term, Opts). jsx:encode(Term, Opts).
-spec(safe_encode(jsx:json_term()) -spec(safe_encode(jsx:json_term())
-> {ok, jsx:json_text()} | {error, term()}). -> {ok, jsx:json_text()} | {error, Reason :: term()}).
safe_encode(Term) -> safe_encode(Term) ->
safe_encode(Term, []). safe_encode(Term, []).
-spec(safe_encode(jsx:json_term(), jsx_to_json:config()) -spec(safe_encode(jsx:json_term(), jsx_to_json:config())
-> {ok, jsx:json_text()} | {error, term()}). -> {ok, jsx:json_text()} | {error, Reason :: term()}).
safe_encode(Term, Opts) -> safe_encode(Term, Opts) ->
try encode(Term, Opts) of try encode(Term, Opts) of
Json -> {ok, Json} Json -> {ok, Json}
@ -58,12 +60,12 @@ decode(Json, Opts) ->
jsx:decode(Json, Opts). jsx:decode(Json, Opts).
-spec(safe_decode(jsx:json_text()) -spec(safe_decode(jsx:json_text())
-> {ok, jsx:json_term()} | {error, term()}). -> {ok, jsx:json_term()} | {error, Reason :: term()}).
safe_decode(Json) -> safe_decode(Json) ->
safe_decode(Json, []). safe_decode(Json, []).
-spec(safe_decode(jsx:json_text(), jsx_to_json:config()) -spec(safe_decode(jsx:json_text(), jsx_to_json:config())
-> {ok, jsx:json_term()} | {error, term()}). -> {ok, jsx:json_term()} | {error, Reason :: term()}).
safe_decode(Json, Opts) -> safe_decode(Json, Opts) ->
try decode(Json, Opts) of try decode(Json, Opts) of
Term -> {ok, Term} Term -> {ok, Term}

4
src/emqx_kernel_sup.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_kernel_sup). -module(emqx_kernel_sup).

9
src/emqx_listeners.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc Start/Stop MQTT listeners. %% @doc Start/Stop MQTT listeners.
-module(emqx_listeners). -module(emqx_listeners).
@ -33,9 +35,9 @@
-type(listener() :: {esockd:proto(), esockd:listen_on(), [esockd:option()]}). -type(listener() :: {esockd:proto(), esockd:listen_on(), [esockd:option()]}).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Start all listeners. %% @doc Start all listeners.
-spec(start() -> ok). -spec(start() -> ok).
@ -167,3 +169,4 @@ format({Addr, Port}) when is_list(Addr) ->
io_lib:format("~s:~w", [Addr, Port]); io_lib:format("~s:~w", [Addr, Port]);
format({Addr, Port}) when is_tuple(Addr) -> format({Addr, Port}) when is_tuple(Addr) ->
io_lib:format("~s:~w", [esockd_net:ntoab(Addr), Port]). io_lib:format("~s:~w", [esockd_net:ntoab(Addr), Port]).

12
src/emqx_logger.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_logger). -module(emqx_logger).
@ -48,9 +50,9 @@
, get_log_handler/1 , get_log_handler/1
]). ]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
debug(Msg) -> debug(Msg) ->
logger:debug(Msg). logger:debug(Msg).
@ -120,9 +122,9 @@ set_log_level(Level) ->
{error, Error} -> {error, {primary_logger_level, Error}} {error, Error} -> {error, {primary_logger_level, Error}}
end. end.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal Functions %% Internal Functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
log_hanlder_info(#{id := Id, level := Level, module := logger_std_h, log_hanlder_info(#{id := Id, level := Level, module := logger_std_h,
config := #{type := Type}}) when Type =:= standard_io; config := #{type := Type}}) when Type =:= standard_io;

5
src/emqx_logger_handler.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_logger_handler). -module(emqx_logger_handler).
@ -41,3 +43,4 @@ filter_by_level(LogEvent = #{level := error}, _Extra) ->
LogEvent; LogEvent;
filter_by_level(_LogEvent, _Extra) -> filter_by_level(_LogEvent, _Extra) ->
stop. stop.

4
src/emqx_message.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_message). -module(emqx_message).

20
src/emqx_metrics.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_metrics). -module(emqx_metrics).
@ -143,9 +145,9 @@ start_link() ->
stop() -> stop() ->
gen_server:stop(?SERVER). gen_server:stop(?SERVER).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Metrics API %% Metrics API
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(new(metric_name()) -> ok). -spec(new(metric_name()) -> ok).
new(Name) -> new(Name) ->
@ -249,9 +251,9 @@ update_counter(Name, Value) ->
end, end,
counters:add(CRef, CIdx, Value). counters:add(CRef, CIdx, Value).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Inc Received/Sent metrics %% Inc Received/Sent metrics
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Inc packets received. %% @doc Inc packets received.
-spec(inc_recv(emqx_mqtt_types:packet()) -> ok). -spec(inc_recv(emqx_mqtt_types:packet()) -> ok).
@ -337,9 +339,9 @@ do_inc_sent(?PACKET(?AUTH)) ->
do_inc_sent(_Packet) -> do_inc_sent(_Packet) ->
ignore. ignore.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([]) -> init([]) ->
% Create counters array % Create counters array
@ -389,9 +391,9 @@ terminate(_Reason, _State) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
reserved_idx('bytes.received') -> 01; reserved_idx('bytes.received') -> 01;
reserved_idx('bytes.sent') -> 02; reserved_idx('bytes.sent') -> 02;

4
src/emqx_misc.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_misc). -module(emqx_misc).

16
src/emqx_mod_acl_internal.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_mod_acl_internal). -module(emqx_mod_acl_internal).
@ -35,9 +37,9 @@
-type(acl_rules() :: #{publish => [emqx_access_rule:rule()], -type(acl_rules() :: #{publish => [emqx_access_rule:rule()],
subscribe => [emqx_access_rule:rule()]}). subscribe => [emqx_access_rule:rule()]}).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% API %% API
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
load(_Env) -> load(_Env) ->
Rules = rules_from_file(acl_file()), Rules = rules_from_file(acl_file()),
@ -52,9 +54,9 @@ unload(_Env) ->
all_rules() -> all_rules() ->
rules_from_file(acl_file()). rules_from_file(acl_file()).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% ACL callbacks %% ACL callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Check ACL %% @doc Check ACL
-spec(check_acl(emqx_types:credentials(), emqx_types:pubsub(), emqx_topic:topic(), -spec(check_acl(emqx_types:credentials(), emqx_types:pubsub(), emqx_topic:topic(),
@ -71,9 +73,9 @@ check_acl(Credentials, PubSub, Topic, _AclResult, Rules) ->
reload_acl() -> reload_acl() ->
unload([]), load([]). unload([]), load([]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal Functions %% Internal Functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
acl_file() -> acl_file() ->
emqx_config:get_env(acl_file). emqx_config:get_env(acl_file).

8
src/emqx_mod_presence.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_mod_presence). -module(emqx_mod_presence).
@ -31,9 +33,9 @@
-define(ATTR_KEYS, [clean_start, proto_ver, proto_name, keepalive]). -define(ATTR_KEYS, [clean_start, proto_ver, proto_name, keepalive]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
load(Env) -> load(Env) ->
emqx_hooks:add('client.connected', fun ?MODULE:on_client_connected/4, [Env]), emqx_hooks:add('client.connected', fun ?MODULE:on_client_connected/4, [Env]),

4
src/emqx_mod_rewrite.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_mod_rewrite). -module(emqx_mod_rewrite).

4
src/emqx_mod_subscription.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_mod_subscription). -module(emqx_mod_subscription).

19
src/emqx_mod_sup.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,11 +12,14 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_mod_sup). -module(emqx_mod_sup).
-behaviour(supervisor). -behaviour(supervisor).
-include("types.hrl").
-export([ start_link/0 -export([ start_link/0
, start_child/1 , start_child/1
, start_child/2 , start_child/2
@ -25,8 +29,14 @@
-export([init/1]). -export([init/1]).
%% Helper macro for declaring children of supervisor %% Helper macro for declaring children of supervisor
-define(CHILD(Mod, Type), {Mod, {Mod, start_link, []}, permanent, 5000, Type, [Mod]}). -define(CHILD(Mod, Type), #{id => Mod,
start => {Mod, start_link, []},
restart => permanent,
shutdown => 5000,
type => Type,
modules => [Mod]}).
-spec(start_link() -> startlink_ret()).
start_link() -> start_link() ->
supervisor:start_link({local, ?MODULE}, ?MODULE, []). supervisor:start_link({local, ?MODULE}, ?MODULE, []).
@ -43,9 +53,10 @@ stop_child(ChildId) ->
Error -> Error Error -> Error
end. end.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Supervisor callbacks %% Supervisor callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([]) -> init([]) ->
{ok, {{one_for_one, 10, 100}, []}}. {ok, {{one_for_one, 10, 100}, []}}.

8
src/emqx_mountpoint.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_mountpoint). -module(emqx_mountpoint).
@ -27,9 +29,9 @@
-export_type([mountpoint/0]). -export_type([mountpoint/0]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
mount(undefined, Any) -> mount(undefined, Any) ->
Any; Any;

4
src/emqx_mqtt_caps.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc MQTTv5 capabilities %% @doc MQTTv5 capabilities
-module(emqx_mqtt_caps). -module(emqx_mqtt_caps).

4
src/emqx_mqtt_props.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc MQTT5 Properties %% @doc MQTT5 Properties
-module(emqx_mqtt_props). -module(emqx_mqtt_props).

4
src/emqx_mqtt_types.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_mqtt_types). -module(emqx_mqtt_types).

6
src/emqx_mqueue.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,7 +12,9 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%%--------------------------------------------------------------------
%% @doc A Simple in-memory message queue. %% @doc A Simple in-memory message queue.
%% %%
%% Notice that MQTT is not a (on-disk) persistent messaging queue. %% Notice that MQTT is not a (on-disk) persistent messaging queue.
@ -42,6 +45,7 @@
%% unless `max_len' is set to `0' which implies (`infinity'). %% unless `max_len' is set to `0' which implies (`infinity').
%% %%
%% @end %% @end
%%--------------------------------------------------------------------
-module(emqx_mqueue). -module(emqx_mqueue).

82
src/emqx_os_mon.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_os_mon). -module(emqx_os_mon).
@ -20,15 +22,6 @@
-export([start_link/1]). -export([start_link/1]).
%% gen_server callbacks
-export([ init/1
, handle_call/3
, handle_cast/2
, handle_info/2
, terminate/2
, code_change/3
]).
-export([ get_cpu_check_interval/0 -export([ get_cpu_check_interval/0
, set_cpu_check_interval/1 , set_cpu_check_interval/1
, get_cpu_high_watermark/0 , get_cpu_high_watermark/0
@ -43,15 +36,24 @@
, set_procmem_high_watermark/1 , set_procmem_high_watermark/1
]). ]).
-define(OS_MON, ?MODULE). %% gen_server callbacks
-export([ init/1
, handle_call/3
, handle_cast/2
, handle_info/2
, terminate/2
, code_change/3
]).
%%------------------------------------------------------------------------------ -define(OS_MON, ?MODULE).
%% API
%%------------------------------------------------------------------------------
start_link(Opts) -> start_link(Opts) ->
gen_server:start_link({local, ?OS_MON}, ?MODULE, [Opts], []). gen_server:start_link({local, ?OS_MON}, ?MODULE, [Opts], []).
%%--------------------------------------------------------------------
%% API
%%--------------------------------------------------------------------
get_cpu_check_interval() -> get_cpu_check_interval() ->
call(get_cpu_check_interval). call(get_cpu_check_interval).
@ -88,9 +90,12 @@ get_procmem_high_watermark() ->
set_procmem_high_watermark(Float) -> set_procmem_high_watermark(Float) ->
memsup:set_procmem_high_watermark(Float). memsup:set_procmem_high_watermark(Float).
%%------------------------------------------------------------------------------ call(Req) ->
gen_server:call(?OS_MON, Req, infinity).
%%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([Opts]) -> init([Opts]) ->
_ = cpu_sup:util(), _ = cpu_sup:util(),
@ -105,47 +110,58 @@ init([Opts]) ->
handle_call(get_cpu_check_interval, _From, State) -> handle_call(get_cpu_check_interval, _From, State) ->
{reply, maps:get(cpu_check_interval, State, undefined), State}; {reply, maps:get(cpu_check_interval, State, undefined), State};
handle_call({set_cpu_check_interval, Seconds}, _From, State) -> handle_call({set_cpu_check_interval, Seconds}, _From, State) ->
{reply, ok, State#{cpu_check_interval := Seconds}}; {reply, ok, State#{cpu_check_interval := Seconds}};
handle_call(get_cpu_high_watermark, _From, State) -> handle_call(get_cpu_high_watermark, _From, State) ->
{reply, maps:get(cpu_high_watermark, State, undefined), State}; {reply, maps:get(cpu_high_watermark, State, undefined), State};
handle_call({set_cpu_high_watermark, Float}, _From, State) -> handle_call({set_cpu_high_watermark, Float}, _From, State) ->
{reply, ok, State#{cpu_high_watermark := Float}}; {reply, ok, State#{cpu_high_watermark := Float}};
handle_call(get_cpu_low_watermark, _From, State) -> handle_call(get_cpu_low_watermark, _From, State) ->
{reply, maps:get(cpu_low_watermark, State, undefined), State}; {reply, maps:get(cpu_low_watermark, State, undefined), State};
handle_call({set_cpu_low_watermark, Float}, _From, State) -> handle_call({set_cpu_low_watermark, Float}, _From, State) ->
{reply, ok, State#{cpu_low_watermark := Float}}; {reply, ok, State#{cpu_low_watermark := Float}};
handle_call(_Request, _From, State) -> handle_call(Req, _From, State) ->
{reply, ok, State}. ?LOG(error, "[OS_MON] Unexpected call: ~p", [Req]),
{reply, ignored, State}.
handle_cast(_Request, State) -> handle_cast(Msg, State) ->
?LOG(error, "[OS_MON] Unexpected cast: ~p", [Msg]),
{noreply, State}. {noreply, State}.
handle_info({timeout, Timer, check}, State = #{timer := Timer, handle_info({timeout, Timer, check},
State = #{timer := Timer,
cpu_high_watermark := CPUHighWatermark, cpu_high_watermark := CPUHighWatermark,
cpu_low_watermark := CPULowWatermark, cpu_low_watermark := CPULowWatermark,
is_cpu_alarm_set := IsCPUAlarmSet}) -> is_cpu_alarm_set := IsCPUAlarmSet}) ->
case cpu_sup:util() of NState = case cpu_sup:util() of
0 -> 0 ->
{noreply, State#{timer := undefined}}; State#{timer := undefined};
{error, Reason} -> {error, Reason} ->
?LOG(error, "[OS Monitor] Failed to get cpu utilization: ~p", [Reason]), ?LOG(error, "[OS_MON] Failed to get cpu utilization: ~p", [Reason]),
{noreply, ensure_check_timer(State)}; ensure_check_timer(State);
Busy when Busy / 100 >= CPUHighWatermark -> Busy when (Busy / 100) >= CPUHighWatermark ->
alarm_handler:set_alarm({cpu_high_watermark, Busy}), alarm_handler:set_alarm({cpu_high_watermark, Busy}),
{noreply, ensure_check_timer(State#{is_cpu_alarm_set := true})}; ensure_check_timer(State#{is_cpu_alarm_set := true});
Busy when Busy / 100 < CPULowWatermark -> Busy when (Busy / 100) < CPULowWatermark ->
case IsCPUAlarmSet of case IsCPUAlarmSet of
true -> alarm_handler:clear_alarm(cpu_high_watermark); true -> alarm_handler:clear_alarm(cpu_high_watermark);
false -> ok false -> ok
end, end,
{noreply, ensure_check_timer(State#{is_cpu_alarm_set := false})}; ensure_check_timer(State#{is_cpu_alarm_set := false});
_Busy -> _Busy ->
{noreply, ensure_check_timer(State)} ensure_check_timer(State)
end. end,
{noreply, NState};
handle_info(Info, State) ->
?LOG(error, "[OS_MON] Unexpected info: ~p", [Info]),
{noreply, State}.
terminate(_Reason, #{timer := Timer}) -> terminate(_Reason, #{timer := Timer}) ->
emqx_misc:cancel_timer(Timer). emqx_misc:cancel_timer(Timer).
@ -153,11 +169,9 @@ terminate(_Reason, #{timer := Timer}) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
call(Req) ->
gen_server:call(?OS_MON, Req, infinity).
ensure_check_timer(State = #{cpu_check_interval := Interval}) -> ensure_check_timer(State = #{cpu_check_interval := Interval}) ->
State#{timer := emqx_misc:start_timer(timer:seconds(Interval), check)}. State#{timer := emqx_misc:start_timer(timer:seconds(Interval), check)}.

8
src/emqx_packet.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_packet). -module(emqx_packet).
@ -40,9 +42,9 @@ protocol_name(?MQTT_PROTO_V5) ->
type_name(Type) when Type > ?RESERVED andalso Type =< ?AUTH -> type_name(Type) when Type > ?RESERVED andalso Type =< ?AUTH ->
lists:nth(Type, ?TYPE_NAMES). lists:nth(Type, ?TYPE_NAMES).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Validate MQTT Packet %% Validate MQTT Packet
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
validate(?SUBSCRIBE_PACKET(_PacketId, _Properties, [])) -> validate(?SUBSCRIBE_PACKET(_PacketId, _Properties, [])) ->
error(topic_filters_invalid); error(topic_filters_invalid);

4
src/emqx_pd.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc The utility functions for erlang process dictionary. %% @doc The utility functions for erlang process dictionary.
-module(emqx_pd). -module(emqx_pd).

12
src/emqx_plugins.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_plugins). -module(emqx_plugins).
@ -28,9 +30,9 @@
, load_expand_plugin/1 , load_expand_plugin/1
]). ]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Init plugins' config %% @doc Init plugins' config
-spec(init() -> ok). -spec(init() -> ok).
@ -45,8 +47,8 @@ init() ->
init_config(CfgFile) -> init_config(CfgFile) ->
{ok, [AppsEnv]} = file:consult(CfgFile), {ok, [AppsEnv]} = file:consult(CfgFile),
lists:foreach(fun({AppName, Envs}) -> lists:foreach(fun({App, Envs}) ->
[application:set_env(AppName, Par, Val) || {Par, Val} <- Envs] [application:set_env(App, Par, Val) || {Par, Val} <- Envs]
end, AppsEnv). end, AppsEnv).
%% @doc Load all plugins when the broker started. %% @doc Load all plugins when the broker started.

8
src/emqx_pmon.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_pmon). -module(emqx_pmon).
@ -33,9 +35,9 @@
-type(pmon() :: {?MODULE, map()}). -type(pmon() :: {?MODULE, map()}).
-export_type([pmon/0]). -export_type([pmon/0]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(new() -> pmon()). -spec(new() -> pmon()).
new() -> new() ->

16
src/emqx_pool.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_pool). -module(emqx_pool).
@ -45,9 +47,9 @@
-type(task() :: fun() | mfa() | {fun(), Args :: list(any())}). -type(task() :: fun() | mfa() | {fun(), Args :: list(any())}).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Start pool. %% @doc Start pool.
-spec(start_link(atom(), pos_integer()) -> startlink_ret()). -spec(start_link(atom(), pos_integer()) -> startlink_ret()).
@ -85,9 +87,9 @@ cast(Msg) ->
worker() -> worker() ->
gproc_pool:pick_worker(?POOL). gproc_pool:pick_worker(?POOL).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([Pool, Id]) -> init([Pool, Id]) ->
true = gproc_pool:connect_worker(Pool, {Pool, Id}), true = gproc_pool:connect_worker(Pool, {Pool, Id}),
@ -121,9 +123,9 @@ terminate(_Reason, #{pool := Pool, id := Id}) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
run({M, F, A}) -> run({M, F, A}) ->
erlang:apply(M, F, A); erlang:apply(M, F, A);

16
src/emqx_pool_sup.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_pool_sup). -module(emqx_pool_sup).
@ -44,7 +46,8 @@ spec(ChildId, Args) ->
start_link() -> start_link() ->
start_link(?POOL, random, {?POOL, start_link, []}). start_link(?POOL, random, {?POOL, start_link, []}).
-spec(start_link(atom() | tuple(), atom(), mfa()) -> {ok, pid()} | {error, term()}). -spec(start_link(atom() | tuple(), atom(), mfa())
-> {ok, pid()} | {error, term()}).
start_link(Pool, Type, MFA) -> start_link(Pool, Type, MFA) ->
start_link(Pool, Type, emqx_vm:schedulers(), MFA). start_link(Pool, Type, emqx_vm:schedulers(), MFA).
@ -54,11 +57,16 @@ start_link(Pool, Type, Size, MFA) ->
supervisor:start_link(?MODULE, [Pool, Type, Size, MFA]). supervisor:start_link(?MODULE, [Pool, Type, Size, MFA]).
init([Pool, Type, Size, {M, F, Args}]) -> init([Pool, Type, Size, {M, F, Args}]) ->
ensure_pool(Pool, Type, [{size, Size}]), ok = ensure_pool(Pool, Type, [{size, Size}]),
{ok, {{one_for_one, 10, 3600}, [ {ok, {{one_for_one, 10, 3600}, [
begin begin
ensure_pool_worker(Pool, {Pool, I}, I), ensure_pool_worker(Pool, {Pool, I}, I),
{{M, I}, {M, F, [Pool, I | Args]}, transient, 5000, worker, [M]} #{id => {M, I},
start => {M, F, [Pool, I | Args]},
restart => transient,
shutdown => 5000,
type => worker,
modules => [M]}
end || I <- lists:seq(1, Size)]}}. end || I <- lists:seq(1, Size)]}}.
ensure_pool(Pool, Type, Opts) -> ensure_pool(Pool, Type, Opts) ->

138
src/emqx_protocol.erl
View File

@ -49,7 +49,6 @@
proto_name, proto_name,
client_id, client_id,
is_assigned, is_assigned,
conn_pid,
conn_props, conn_props,
ack_props, ack_props,
username, username,
@ -81,15 +80,15 @@
-define(NO_PROPS, undefined). -define(NO_PROPS, undefined).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Init %% Init
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(init(map(), list()) -> state()). -spec(init(map(), list()) -> state()).
init(SocketOpts = #{ sockname := Sockname init(SocketOpts = #{sockname := Sockname,
, peername := Peername peername := Peername,
, peercert := Peercert peercert := Peercert,
, sendfun := SendFun}, Options) -> sendfun := SendFun}, Options) ->
Zone = proplists:get_value(zone, Options), Zone = proplists:get_value(zone, Options),
#pstate{zone = Zone, #pstate{zone = Zone,
sendfun = SendFun, sendfun = SendFun,
@ -100,7 +99,6 @@ init(SocketOpts = #{ sockname := Sockname
proto_name = <<"MQTT">>, proto_name = <<"MQTT">>,
client_id = <<>>, client_id = <<>>,
is_assigned = false, is_assigned = false,
conn_pid = self(),
username = init_username(Peercert, Options), username = init_username(Peercert, Options),
clean_start = false, clean_start = false,
topic_aliases = #{}, topic_aliases = #{},
@ -108,6 +106,7 @@ init(SocketOpts = #{ sockname := Sockname
recv_stats = #{msg => 0, pkt => 0}, recv_stats = #{msg => 0, pkt => 0},
send_stats = #{msg => 0, pkt => 0}, send_stats = #{msg => 0, pkt => 0},
connected = false, connected = false,
%% TODO: ...?
topic_alias_maximum = #{to_client => 0, from_client => 0}, topic_alias_maximum = #{to_client => 0, from_client => 0},
conn_mod = maps:get(conn_mod, SocketOpts, undefined), conn_mod = maps:get(conn_mod, SocketOpts, undefined),
credentials = #{}, credentials = #{},
@ -126,9 +125,9 @@ set_username(Username, PState = #pstate{username = undefined}) ->
set_username(_Username, PState) -> set_username(_Username, PState) ->
PState. PState.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% API %% API
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
info(PState = #pstate{zone = Zone, info(PState = #pstate{zone = Zone,
conn_props = ConnProps, conn_props = ConnProps,
@ -157,19 +156,19 @@ attrs(#pstate{zone = Zone,
connected_at = ConnectedAt, connected_at = ConnectedAt,
conn_mod = ConnMod, conn_mod = ConnMod,
credentials = Credentials}) -> credentials = Credentials}) ->
#{ zone => Zone #{zone => Zone,
, client_id => ClientId client_id => ClientId,
, username => Username username => Username,
, peername => Peername peername => Peername,
, peercert => Peercert peercert => Peercert,
, proto_ver => ProtoVer proto_ver => ProtoVer,
, proto_name => ProtoName proto_name => ProtoName,
, clean_start => CleanStart clean_start => CleanStart,
, keepalive => Keepalive keepalive => Keepalive,
, is_bridge => IsBridge is_bridge => IsBridge,
, connected_at => ConnectedAt connected_at => ConnectedAt,
, conn_mod => ConnMod conn_mod => ConnMod,
, credentials => Credentials credentials => Credentials
}. }.
attr(proto_ver, #pstate{proto_ver = ProtoVer}) -> attr(proto_ver, #pstate{proto_ver = ProtoVer}) ->
@ -238,8 +237,8 @@ stats(#pstate{recv_stats = #{pkt := RecvPkt, msg := RecvMsg},
{send_pkt, SendPkt}, {send_pkt, SendPkt},
{send_msg, SendMsg}]. {send_msg, SendMsg}].
session(#pstate{session = SPid}) -> session(#pstate{session = Session}) ->
SPid. Session.
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% Packet Received %% Packet Received
@ -364,6 +363,7 @@ preprocess_properties(Packet, PState) ->
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% Process MQTT Packet %% Process MQTT Packet
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
process(?CONNECT_PACKET( process(?CONNECT_PACKET(
#mqtt_packet_connect{proto_name = ProtoName, #mqtt_packet_connect{proto_name = ProtoName,
proto_ver = ProtoVer, proto_ver = ProtoVer,
@ -403,11 +403,11 @@ process(?CONNECT_PACKET(
%% Open session %% Open session
SessAttrs = #{will_msg => make_will_msg(ConnPkt)}, SessAttrs = #{will_msg => make_will_msg(ConnPkt)},
case try_open_session(SessAttrs, PState3) of case try_open_session(SessAttrs, PState3) of
{ok, SPid, SP} -> {ok, Session, SP} ->
PState4 = PState3#pstate{session = SPid, connected = true, PState4 = PState3#pstate{session = Session, connected = true,
credentials = keepsafety(Credentials0)}, credentials = keepsafety(Credentials0)},
ok = emqx_cm:register_connection(client_id(PState4)), ok = emqx_cm:register_channel(client_id(PState4)),
true = emqx_cm:set_conn_attrs(client_id(PState4), attrs(PState4)), ok = emqx_cm:set_conn_attrs(client_id(PState4), attrs(PState4)),
%% Start keepalive %% Start keepalive
start_keepalive(Keepalive, PState4), start_keepalive(Keepalive, PState4),
%% Success %% Success
@ -470,36 +470,43 @@ process(Packet = ?PUBLISH_PACKET(?QOS_2, Topic, PacketId, _Payload),
end end
end; end;
process(?PUBACK_PACKET(PacketId, ReasonCode), PState = #pstate{session = SPid}) -> process(?PUBACK_PACKET(PacketId, ReasonCode), PState = #pstate{session = Session}) ->
{ok = emqx_session:puback(SPid, PacketId, ReasonCode), PState}; NSession = emqx_session:puback(PacketId, ReasonCode, Session),
{ok, PState#pstate{session = NSession}};
process(?PUBREC_PACKET(PacketId, ReasonCode), PState = #pstate{session = SPid}) -> process(?PUBREC_PACKET(PacketId, ReasonCode), PState = #pstate{session = Session}) ->
case emqx_session:pubrec(SPid, PacketId, ReasonCode) of case emqx_session:pubrec(PacketId, ReasonCode, Session) of
ok -> {ok, NSession} ->
send(?PUBREL_PACKET(PacketId), PState); send(?PUBREL_PACKET(PacketId), PState#pstate{session = NSession});
{error, NotFound} -> {error, NotFound} ->
send(?PUBREL_PACKET(PacketId, NotFound), PState) send(?PUBREL_PACKET(PacketId, NotFound), PState)
end; end;
process(?PUBREL_PACKET(PacketId, ReasonCode), PState = #pstate{session = SPid}) -> process(?PUBREL_PACKET(PacketId, ReasonCode), PState = #pstate{session = Session}) ->
case emqx_session:pubrel(SPid, PacketId, ReasonCode) of case emqx_session:pubrel(PacketId, ReasonCode, Session) of
ok -> {ok, NSession} ->
send(?PUBCOMP_PACKET(PacketId), PState); send(?PUBCOMP_PACKET(PacketId), PState#pstate{session = NSession});
{error, NotFound} -> {error, NotFound} ->
send(?PUBCOMP_PACKET(PacketId, NotFound), PState) send(?PUBCOMP_PACKET(PacketId, NotFound), PState)
end; end;
process(?PUBCOMP_PACKET(PacketId, ReasonCode), PState = #pstate{session = SPid}) -> process(?PUBCOMP_PACKET(PacketId, ReasonCode), PState = #pstate{session = Session}) ->
{ok = emqx_session:pubcomp(SPid, PacketId, ReasonCode), PState}; case emqx_session:pubcomp(PacketId, ReasonCode, Session) of
{ok, NSession} ->
{ok, PState#pstate{session = NSession}};
{error, _NotFound} ->
%% TODO: How to handle NotFound?
{ok, PState}
end;
process(Packet = ?SUBSCRIBE_PACKET(PacketId, Properties, RawTopicFilters), process(Packet = ?SUBSCRIBE_PACKET(PacketId, Properties, RawTopicFilters),
PState = #pstate{zone = Zone, proto_ver = ProtoVer, session = SPid, credentials = Credentials}) -> PState = #pstate{zone = Zone, proto_ver = ProtoVer, session = Session, credentials = Credentials}) ->
case check_subscribe(parse_topic_filters(?SUBSCRIBE, raw_topic_filters(PState, RawTopicFilters)), PState) of case check_subscribe(parse_topic_filters(?SUBSCRIBE, raw_topic_filters(PState, RawTopicFilters)), PState) of
{ok, TopicFilters} -> {ok, TopicFilters} ->
TopicFilters0 = emqx_hooks:run_fold('client.subscribe', [Credentials], TopicFilters), TopicFilters0 = emqx_hooks:run_fold('client.subscribe', [Credentials], TopicFilters),
TopicFilters1 = emqx_mountpoint:mount(mountpoint(Credentials), TopicFilters0), TopicFilters1 = emqx_mountpoint:mount(mountpoint(Credentials), TopicFilters0),
ok = emqx_session:subscribe(SPid, PacketId, Properties, TopicFilters1), {ok, ReasonCodes, NSession} = emqx_session:subscribe(TopicFilters1, Session),
{ok, PState}; deliver({suback, PacketId, ReasonCodes}, PState#pstate{session = NSession});
{error, TopicFilters} -> {error, TopicFilters} ->
{SubTopics, ReasonCodes} = {SubTopics, ReasonCodes} =
lists:foldr(fun({Topic, #{rc := ?RC_SUCCESS}}, {Topics, Codes}) -> lists:foldr(fun({Topic, #{rc := ?RC_SUCCESS}}, {Topics, Codes}) ->
@ -520,26 +527,26 @@ process(Packet = ?SUBSCRIBE_PACKET(PacketId, Properties, RawTopicFilters),
end; end;
process(?UNSUBSCRIBE_PACKET(PacketId, Properties, RawTopicFilters), process(?UNSUBSCRIBE_PACKET(PacketId, Properties, RawTopicFilters),
PState = #pstate{session = SPid, credentials = Credentials}) -> PState = #pstate{session = Session, credentials = Credentials}) ->
TopicFilters = emqx_hooks:run_fold('client.unsubscribe', [Credentials], TopicFilters = emqx_hooks:run_fold('client.unsubscribe', [Credentials],
parse_topic_filters(?UNSUBSCRIBE, RawTopicFilters)), parse_topic_filters(?UNSUBSCRIBE, RawTopicFilters)),
ok = emqx_session:unsubscribe(SPid, PacketId, Properties, TopicFilters1 = emqx_mountpoint:mount(mountpoint(Credentials), TopicFilters),
emqx_mountpoint:mount(mountpoint(Credentials), TopicFilters)), {ok, ReasonCodes, NSession} = emqx_session:unsubscribe(TopicFilters1, Session),
{ok, PState}; deliver({unsuback, PacketId, ReasonCodes}, PState#pstate{session = NSession});
process(?PACKET(?PINGREQ), PState) -> process(?PACKET(?PINGREQ), PState) ->
send(?PACKET(?PINGRESP), PState); send(?PACKET(?PINGRESP), PState);
process(?DISCONNECT_PACKET(?RC_SUCCESS, #{'Session-Expiry-Interval' := Interval}), process(?DISCONNECT_PACKET(?RC_SUCCESS, #{'Session-Expiry-Interval' := Interval}),
PState = #pstate{session = SPid, conn_props = #{'Session-Expiry-Interval' := OldInterval}}) -> PState = #pstate{session = Session, conn_props = #{'Session-Expiry-Interval' := OldInterval}}) ->
case Interval =/= 0 andalso OldInterval =:= 0 of case Interval =/= 0 andalso OldInterval =:= 0 of
true -> true ->
deliver({disconnect, ?RC_PROTOCOL_ERROR}, PState), deliver({disconnect, ?RC_PROTOCOL_ERROR}, PState),
{error, protocol_error, PState#pstate{will_msg = undefined}}; {error, protocol_error, PState#pstate{will_msg = undefined}};
false -> false ->
emqx_session:update_expiry_interval(SPid, Interval), NSession = emqx_session:update_expiry_interval(Interval, Session),
%% Clean willmsg %% Clean willmsg
{stop, normal, PState#pstate{will_msg = undefined}} {stop, normal, PState#pstate{will_msg = undefined, session = NSession}}
end; end;
process(?DISCONNECT_PACKET(?RC_SUCCESS), PState) -> process(?DISCONNECT_PACKET(?RC_SUCCESS), PState) ->
@ -562,19 +569,27 @@ connack({ReasonCode, PState = #pstate{proto_ver = ProtoVer, credentials = Creden
_ = deliver({connack, ReasonCode1}, PState), _ = deliver({connack, ReasonCode1}, PState),
{error, emqx_reason_codes:name(ReasonCode1, ProtoVer), PState}. {error, emqx_reason_codes:name(ReasonCode1, ProtoVer), PState}.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Publish Message -> Broker %% Publish Message -> Broker
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
do_publish(Packet = ?PUBLISH_PACKET(QoS, PacketId), do_publish(Packet = ?PUBLISH_PACKET(QoS, PacketId),
PState = #pstate{session = SPid, credentials = Credentials}) -> PState = #pstate{session = Session, credentials = Credentials}) ->
Msg = emqx_mountpoint:mount(mountpoint(Credentials), Msg = emqx_mountpoint:mount(mountpoint(Credentials),
emqx_packet:to_message(Credentials, Packet)), emqx_packet:to_message(Credentials, Packet)),
puback(QoS, PacketId, emqx_session:publish(SPid, PacketId, emqx_message:set_flag(dup, false, Msg)), PState). Msg1 = emqx_message:set_flag(dup, false, Msg),
case emqx_session:publish(PacketId, Msg1, Session) of
{ok, Result} ->
puback(QoS, PacketId, {ok, Result}, PState);
{ok, Result, NSession} ->
puback(QoS, PacketId, {ok, Result}, PState#pstate{session = NSession});
{error, ReasonCode} ->
puback(QoS, PacketId, {error, ReasonCode}, PState)
end.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Puback -> Client %% Puback -> Client
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
puback(?QOS_0, _PacketId, _Result, PState) -> puback(?QOS_0, _PacketId, _Result, PState) ->
{ok, PState}; {ok, PState};
@ -734,21 +749,19 @@ maybe_assign_client_id(PState) ->
try_open_session(SessAttrs, PState = #pstate{zone = Zone, try_open_session(SessAttrs, PState = #pstate{zone = Zone,
client_id = ClientId, client_id = ClientId,
conn_pid = ConnPid,
username = Username, username = Username,
clean_start = CleanStart}) -> clean_start = CleanStart}) ->
case emqx_sm:open_session( case emqx_cm:open_session(
maps:merge(#{zone => Zone, maps:merge(#{zone => Zone,
client_id => ClientId, client_id => ClientId,
conn_pid => ConnPid,
username => Username, username => Username,
clean_start => CleanStart, clean_start => CleanStart,
max_inflight => attr(max_inflight, PState), max_inflight => attr(max_inflight, PState),
expiry_interval => attr(expiry_interval, PState), expiry_interval => attr(expiry_interval, PState),
topic_alias_maximum => attr(topic_alias_maximum, PState)}, topic_alias_maximum => attr(topic_alias_maximum, PState)},
SessAttrs)) of SessAttrs)) of
{ok, SPid} -> {ok, Session} ->
{ok, SPid, false}; {ok, Session, false};
Other -> Other Other -> Other
end. end.
@ -1035,3 +1048,4 @@ do_acl_check(Action, Credentials, Topic, AllowTerm, DenyTerm) ->
allow -> AllowTerm; allow -> AllowTerm;
deny -> DenyTerm deny -> DenyTerm
end. end.

4
src/emqx_psk.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_psk). -module(emqx_psk).

4
src/emqx_reason_codes.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc MQTT5 reason codes %% @doc MQTT5 reason codes
-module(emqx_reason_codes). -module(emqx_reason_codes).

24
src/emqx_router.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_router). -module(emqx_router).
@ -66,9 +68,9 @@
-define(ROUTE, emqx_route). -define(ROUTE, emqx_route).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Mnesia bootstrap %% Mnesia bootstrap
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
mnesia(boot) -> mnesia(boot) ->
ok = ekka_mnesia:create_table(?ROUTE, [ ok = ekka_mnesia:create_table(?ROUTE, [
@ -81,18 +83,18 @@ mnesia(boot) ->
mnesia(copy) -> mnesia(copy) ->
ok = ekka_mnesia:copy_table(?ROUTE). ok = ekka_mnesia:copy_table(?ROUTE).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Start a router %% Start a router
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(start_link(atom(), pos_integer()) -> startlink_ret()). -spec(start_link(atom(), pos_integer()) -> startlink_ret()).
start_link(Pool, Id) -> start_link(Pool, Id) ->
gen_server:start_link({local, emqx_misc:proc_name(?MODULE, Id)}, gen_server:start_link({local, emqx_misc:proc_name(?MODULE, Id)},
?MODULE, [Pool, Id], [{hibernate_after, 1000}]). ?MODULE, [Pool, Id], [{hibernate_after, 1000}]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Route APIs %% Route APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(add_route(emqx_topic:topic()) -> ok | {error, term()}). -spec(add_route(emqx_topic:topic()) -> ok | {error, term()}).
add_route(Topic) when is_binary(Topic) -> add_route(Topic) when is_binary(Topic) ->
@ -181,9 +183,9 @@ call(Router, Msg) ->
pick(Topic) -> pick(Topic) ->
gproc_pool:pick_worker(router_pool, Topic). gproc_pool:pick_worker(router_pool, Topic).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([Pool, Id]) -> init([Pool, Id]) ->
true = gproc_pool:connect_worker(Pool, {Pool, Id}), true = gproc_pool:connect_worker(Pool, {Pool, Id}),
@ -215,9 +217,9 @@ terminate(_Reason, #{pool := Pool, id := Id}) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
insert_direct_route(Route) -> insert_direct_route(Route) ->
mnesia:async_dirty(fun mnesia:write/3, [?ROUTE, Route, sticky_write]). mnesia:async_dirty(fun mnesia:write/3, [?ROUTE, Route, sticky_write]).

20
src/emqx_router_helper.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_router_helper). -module(emqx_router_helper).
@ -49,9 +51,9 @@
-define(ROUTING_NODE, emqx_routing_node). -define(ROUTING_NODE, emqx_routing_node).
-define(LOCK, {?MODULE, cleanup_routes}). -define(LOCK, {?MODULE, cleanup_routes}).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Mnesia bootstrap %% Mnesia bootstrap
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
mnesia(boot) -> mnesia(boot) ->
ok = ekka_mnesia:create_table(?ROUTING_NODE, [ ok = ekka_mnesia:create_table(?ROUTING_NODE, [
@ -64,9 +66,9 @@ mnesia(boot) ->
mnesia(copy) -> mnesia(copy) ->
ok = ekka_mnesia:copy_table(?ROUTING_NODE). ok = ekka_mnesia:copy_table(?ROUTING_NODE).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% API %% API
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Starts the router helper %% @doc Starts the router helper
-spec(start_link() -> startlink_ret()). -spec(start_link() -> startlink_ret()).
@ -84,9 +86,9 @@ monitor(Node) when is_atom(Node) ->
false -> mnesia:dirty_write(?ROUTING_NODE, #routing_node{name = Node}) false -> mnesia:dirty_write(?ROUTING_NODE, #routing_node{name = Node})
end. end.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([]) -> init([]) ->
ok = ekka:monitor(membership), ok = ekka:monitor(membership),
@ -152,9 +154,9 @@ terminate(_Reason, _State) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
stats_fun() -> stats_fun() ->
case ets:info(?ROUTE, size) of case ets:info(?ROUTE, size) of

4
src/emqx_router_sup.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_router_sup). -module(emqx_router_sup).

4
src/emqx_rpc.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
%% @doc wrap gen_rpc? %% @doc wrap gen_rpc?
-module(emqx_rpc). -module(emqx_rpc).

8
src/emqx_sequence.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_sequence). -module(emqx_sequence).
@ -29,9 +31,9 @@
-export_type([seqid/0]). -export_type([seqid/0]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Create a sequence. %% @doc Create a sequence.
-spec(create(name()) -> ok). -spec(create(name()) -> ok).

1132
src/emqx_session.erl
View File

File diff suppressed because it is too large Load Diff

23
src/emqx_shared_sub.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_shared_sub). -module(emqx_shared_sub).
@ -65,11 +67,12 @@
-define(no_ack, no_ack). -define(no_ack, no_ack).
-record(state, {pmon}). -record(state, {pmon}).
-record(emqx_shared_subscription, {group, topic, subpid}). -record(emqx_shared_subscription, {group, topic, subpid}).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Mnesia bootstrap %% Mnesia bootstrap
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
mnesia(boot) -> mnesia(boot) ->
ok = ekka_mnesia:create_table(?TAB, [ ok = ekka_mnesia:create_table(?TAB, [
@ -81,9 +84,9 @@ mnesia(boot) ->
mnesia(copy) -> mnesia(copy) ->
ok = ekka_mnesia:copy_table(?TAB). ok = ekka_mnesia:copy_table(?TAB).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% API %% API
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(start_link() -> startlink_ret()). -spec(start_link() -> startlink_ret()).
start_link() -> start_link() ->
@ -273,12 +276,12 @@ do_pick_subscriber(Group, Topic, round_robin, _ClientId, Count) ->
subscribers(Group, Topic) -> subscribers(Group, Topic) ->
ets:select(?TAB, [{{emqx_shared_subscription, Group, Topic, '$1'}, [], ['$1']}]). ets:select(?TAB, [{{emqx_shared_subscription, Group, Topic, '$1'}, [], ['$1']}]).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([]) -> init([]) ->
mnesia:subscribe({table, ?TAB, simple}), {ok, _} = mnesia:subscribe({table, ?TAB, simple}),
{atomic, PMon} = mnesia:transaction(fun init_monitors/0), {atomic, PMon} = mnesia:transaction(fun init_monitors/0),
ok = emqx_tables:new(?SHARED_SUBS, [protected, bag]), ok = emqx_tables:new(?SHARED_SUBS, [protected, bag]),
ok = emqx_tables:new(?ALIVE_SUBS, [protected, set, {read_concurrency, true}]), ok = emqx_tables:new(?ALIVE_SUBS, [protected, set, {read_concurrency, true}]),
@ -343,9 +346,9 @@ terminate(_Reason, _State) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% keep track of alive remote pids %% keep track of alive remote pids
maybe_insert_alive_tab(Pid) when ?IS_LOCAL_PID(Pid) -> ok; maybe_insert_alive_tab(Pid) when ?IS_LOCAL_PID(Pid) -> ok;

295
src/emqx_sm.erl
View File

@ -1,295 +0,0 @@
%% Copyright (c) 2013-2019 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_sm).
-behaviour(gen_server).
-include("emqx.hrl").
-include("logger.hrl").
-include("types.hrl").
%% APIs
-export([start_link/0]).
-export([ open_session/1
, close_session/1
, resume_session/2
, discard_session/1
, discard_session/2
, register_session/1
, register_session/2
, unregister_session/1
, unregister_session/2
]).
-export([ get_session_attrs/1
, get_session_attrs/2
, set_session_attrs/2
, set_session_attrs/3
, get_session_stats/1
, get_session_stats/2
, set_session_stats/2
, set_session_stats/3
]).
-export([lookup_session_pids/1]).
%% Internal functions for rpc
-export([dispatch/3]).
%% Internal function for stats
-export([stats_fun/0]).
%% Internal function for emqx_session_sup
-export([clean_down/1]).
%% gen_server callbacks
-export([ init/1
, handle_call/3
, handle_cast/2
, handle_info/2
, terminate/2
, code_change/3
]).
-define(SM, ?MODULE).
%% ETS Tables for session management.
-define(SESSION_TAB, emqx_session).
-define(SESSION_P_TAB, emqx_session_p).
-define(SESSION_ATTRS_TAB, emqx_session_attrs).
-define(SESSION_STATS_TAB, emqx_session_stats).
-define(BATCH_SIZE, 100000).
%%------------------------------------------------------------------------------
%% APIs
%%------------------------------------------------------------------------------
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?SM}, ?MODULE, [], []).
%% @doc Open a session.
-spec(open_session(map()) -> {ok, pid()} | {ok, pid(), boolean()} | {error, term()}).
open_session(SessAttrs = #{clean_start := true, client_id := ClientId, conn_pid := ConnPid}) ->
CleanStart = fun(_) ->
ok = discard_session(ClientId, ConnPid),
emqx_session_sup:start_session(SessAttrs)
end,
emqx_sm_locker:trans(ClientId, CleanStart);
open_session(SessAttrs = #{clean_start := false, client_id := ClientId}) ->
ResumeStart = fun(_) ->
case resume_session(ClientId, SessAttrs) of
{ok, SessPid} ->
{ok, SessPid, true};
{error, not_found} ->
emqx_session_sup:start_session(SessAttrs)
end
end,
emqx_sm_locker:trans(ClientId, ResumeStart).
%% @doc Discard all the sessions identified by the ClientId.
-spec(discard_session(emqx_types:client_id()) -> ok).
discard_session(ClientId) when is_binary(ClientId) ->
discard_session(ClientId, self()).
-spec(discard_session(emqx_types:client_id(), pid()) -> ok).
discard_session(ClientId, ConnPid) when is_binary(ClientId) ->
lists:foreach(
fun(SessPid) ->
try emqx_session:discard(SessPid, ConnPid)
catch
_:Error:_Stk ->
?LOG(warning, "[SM] Failed to discard ~p: ~p", [SessPid, Error])
end
end, lookup_session_pids(ClientId)).
%% @doc Try to resume a session.
-spec(resume_session(emqx_types:client_id(), map()) -> {ok, pid()} | {error, term()}).
resume_session(ClientId, SessAttrs = #{conn_pid := ConnPid}) ->
case lookup_session_pids(ClientId) of
[] -> {error, not_found};
[SessPid] ->
ok = emqx_session:resume(SessPid, SessAttrs),
{ok, SessPid};
SessPids ->
[SessPid|StalePids] = lists:reverse(SessPids),
?LOG(error, "[SM] More than one session found: ~p", [SessPids]),
lists:foreach(fun(StalePid) ->
catch emqx_session:discard(StalePid, ConnPid)
end, StalePids),
ok = emqx_session:resume(SessPid, SessAttrs),
{ok, SessPid}
end.
%% @doc Close a session.
-spec(close_session(emqx_types:client_id() | pid()) -> ok).
close_session(ClientId) when is_binary(ClientId) ->
case lookup_session_pids(ClientId) of
[] -> ok;
[SessPid] -> close_session(SessPid);
SessPids -> lists:foreach(fun close_session/1, SessPids)
end;
close_session(SessPid) when is_pid(SessPid) ->
emqx_session:close(SessPid).
%% @doc Register a session.
-spec(register_session(emqx_types:client_id()) -> ok).
register_session(ClientId) when is_binary(ClientId) ->
register_session(ClientId, self()).
-spec(register_session(emqx_types:client_id(), pid()) -> ok).
register_session(ClientId, SessPid) when is_binary(ClientId), is_pid(SessPid) ->
Session = {ClientId, SessPid},
true = ets:insert(?SESSION_TAB, Session),
emqx_sm_registry:register_session(Session).
%% @doc Unregister a session
-spec(unregister_session(emqx_types:client_id()) -> ok).
unregister_session(ClientId) when is_binary(ClientId) ->
unregister_session(ClientId, self()).
-spec(unregister_session(emqx_types:client_id(), pid()) -> ok).
unregister_session(ClientId, SessPid) when is_binary(ClientId), is_pid(SessPid) ->
Session = {ClientId, SessPid},
true = ets:delete(?SESSION_STATS_TAB, Session),
true = ets:delete(?SESSION_ATTRS_TAB, Session),
true = ets:delete_object(?SESSION_P_TAB, Session),
true = ets:delete_object(?SESSION_TAB, Session),
emqx_sm_registry:unregister_session(Session).
%% @doc Get session attrs
-spec(get_session_attrs(emqx_types:client_id()) -> list(emqx_session:attr())).
get_session_attrs(ClientId) when is_binary(ClientId) ->
case lookup_session_pids(ClientId) of
[] -> [];
[SessPid|_] -> get_session_attrs(ClientId, SessPid)
end.
-spec(get_session_attrs(emqx_types:client_id(), pid()) -> list(emqx_session:attr())).
get_session_attrs(ClientId, SessPid) when is_binary(ClientId), is_pid(SessPid) ->
emqx_tables:lookup_value(?SESSION_ATTRS_TAB, {ClientId, SessPid}, []).
%% @doc Set session attrs
-spec(set_session_attrs(emqx_types:client_id(), list(emqx_session:attr())) -> true).
set_session_attrs(ClientId, SessAttrs) when is_binary(ClientId) ->
set_session_attrs(ClientId, self(), SessAttrs).
-spec(set_session_attrs(emqx_types:client_id(), pid(), list(emqx_session:attr())) -> true).
set_session_attrs(ClientId, SessPid, SessAttrs) when is_binary(ClientId), is_pid(SessPid) ->
Session = {ClientId, SessPid},
true = ets:insert(?SESSION_ATTRS_TAB, {Session, SessAttrs}),
proplists:get_value(clean_start, SessAttrs, true) orelse ets:insert(?SESSION_P_TAB, Session).
%% @doc Get session stats
-spec(get_session_stats(emqx_types:client_id()) -> list(emqx_stats:stats())).
get_session_stats(ClientId) when is_binary(ClientId) ->
case lookup_session_pids(ClientId) of
[] -> [];
[SessPid|_] ->
get_session_stats(ClientId, SessPid)
end.
-spec(get_session_stats(emqx_types:client_id(), pid()) -> list(emqx_stats:stats())).
get_session_stats(ClientId, SessPid) when is_binary(ClientId) ->
emqx_tables:lookup_value(?SESSION_STATS_TAB, {ClientId, SessPid}, []).
%% @doc Set session stats
-spec(set_session_stats(emqx_types:client_id(), emqx_stats:stats()) -> true).
set_session_stats(ClientId, Stats) when is_binary(ClientId) ->
set_session_stats(ClientId, self(), Stats).
-spec(set_session_stats(emqx_types:client_id(), pid(), emqx_stats:stats()) -> true).
set_session_stats(ClientId, SessPid, Stats) when is_binary(ClientId), is_pid(SessPid) ->
ets:insert(?SESSION_STATS_TAB, {{ClientId, SessPid}, Stats}).
%% @doc Lookup session pid.
-spec(lookup_session_pids(emqx_types:client_id()) -> list(pid())).
lookup_session_pids(ClientId) ->
case emqx_sm_registry:is_enabled() of
true -> emqx_sm_registry:lookup_session(ClientId);
false ->
case emqx_tables:lookup_value(?SESSION_TAB, ClientId) of
undefined -> [];
SessPid when is_pid(SessPid) -> [SessPid]
end
end.
%% @doc Dispatch a message to the session.
-spec(dispatch(emqx_types:client_id(), emqx_topic:topic(), emqx_types:message()) -> any()).
dispatch(ClientId, Topic, Msg) ->
case lookup_session_pids(ClientId) of
[SessPid|_] when is_pid(SessPid) ->
SessPid ! {dispatch, Topic, Msg};
[] ->
emqx_hooks:run('message.dropped', [#{client_id => ClientId}, Msg])
end.
%%------------------------------------------------------------------------------
%% gen_server callbacks
%%------------------------------------------------------------------------------
init([]) ->
TabOpts = [public, set, {write_concurrency, true}],
ok = emqx_tables:new(?SESSION_TAB, [{read_concurrency, true} | TabOpts]),
ok = emqx_tables:new(?SESSION_P_TAB, TabOpts),
ok = emqx_tables:new(?SESSION_ATTRS_TAB, TabOpts),
ok = emqx_tables:new(?SESSION_STATS_TAB, TabOpts),
ok = emqx_stats:update_interval(sess_stats, fun ?MODULE:stats_fun/0),
{ok, #{}}.
handle_call(Req, _From, State) ->
?LOG(error, "[SM] Unexpected call: ~p", [Req]),
{reply, ignored, State}.
handle_cast(Msg, State) ->
?LOG(error, "[SM] Unexpected cast: ~p", [Msg]),
{noreply, State}.
handle_info(Info, State) ->
?LOG(error, "[SM] Unexpected info: ~p", [Info]),
{noreply, State}.
terminate(_Reason, _State) ->
emqx_stats:cancel_update(sess_stats).
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%%------------------------------------------------------------------------------
%% Internal functions
%%------------------------------------------------------------------------------
clean_down(Session = {ClientId, SessPid}) ->
case ets:member(?SESSION_TAB, ClientId)
orelse ets:member(?SESSION_ATTRS_TAB, Session) of
true ->
unregister_session(ClientId, SessPid);
false -> ok
end.
stats_fun() ->
safe_update_stats(?SESSION_TAB, 'sessions.count', 'sessions.max'),
safe_update_stats(?SESSION_P_TAB, 'sessions.persistent.count', 'sessions.persistent.max').
safe_update_stats(Tab, Stat, MaxStat) ->
case ets:info(Tab, size) of
undefined -> ok;
Size -> emqx_stats:setstat(Stat, MaxStat, Size)
end.

68
src/emqx_sm_locker.erl
View File

@ -1,68 +0,0 @@
%% Copyright (c) 2013-2019 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_sm_locker).
-include("emqx.hrl").
-include("types.hrl").
-export([start_link/0]).
-export([ trans/2
, trans/3
, lock/1
, lock/2
, unlock/1
]).
%%------------------------------------------------------------------------------
%% APIs
%%------------------------------------------------------------------------------
-spec(start_link() -> startlink_ret()).
start_link() ->
ekka_locker:start_link(?MODULE).
-spec(trans(emqx_types:client_id(), fun(([node()]) -> any())) -> any()).
trans(ClientId, Fun) ->
trans(ClientId, Fun, undefined).
-spec(trans(maybe(emqx_types:client_id()),
fun(([node()])-> any()), ekka_locker:piggyback()) -> any()).
trans(undefined, Fun, _Piggyback) ->
Fun([]);
trans(ClientId, Fun, Piggyback) ->
case lock(ClientId, Piggyback) of
{true, Nodes} ->
try Fun(Nodes) after unlock(ClientId) end;
{false, _Nodes} ->
{error, client_id_unavailable}
end.
-spec(lock(emqx_types:client_id()) -> ekka_locker:lock_result()).
lock(ClientId) ->
ekka_locker:acquire(?MODULE, ClientId, strategy()).
-spec(lock(emqx_types:client_id(), ekka_locker:piggyback()) -> ekka_locker:lock_result()).
lock(ClientId, Piggyback) ->
ekka_locker:acquire(?MODULE, ClientId, strategy(), Piggyback).
-spec(unlock(emqx_types:client_id()) -> {boolean(), [node()]}).
unlock(ClientId) ->
ekka_locker:release(?MODULE, ClientId, strategy()).
-spec(strategy() -> local | one | quorum | all).
strategy() ->
emqx_config:get_env(session_locking_strategy, quorum).

22
src/emqx_stats.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_stats). -module(emqx_stats).
@ -49,7 +51,11 @@
-record(update, {name, countdown, interval, func}). -record(update, {name, countdown, interval, func}).
-record(state, {timer, updates :: [#update{}], tick_ms :: timeout()}). -record(state, {
timer :: reference(),
updates :: [#update{}],
tick_ms :: timeout()
}).
-type(stats() :: list({atom(), non_neg_integer()})). -type(stats() :: list({atom(), non_neg_integer()})).
@ -166,9 +172,9 @@ rec(Name, Secs, UpFun) ->
cast(Msg) -> cast(Msg) ->
gen_server:cast(?SERVER, Msg). gen_server:cast(?SERVER, Msg).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init(#{tick_ms := TickMs}) -> init(#{tick_ms := TickMs}) ->
ok = emqx_tables:new(?TAB, [public, set, {write_concurrency, true}]), ok = emqx_tables:new(?TAB, [public, set, {write_concurrency, true}]),
@ -199,7 +205,8 @@ handle_cast({setstat, Stat, MaxStat, Val}, State) ->
safe_update_element(Stat, Val), safe_update_element(Stat, Val),
{noreply, State}; {noreply, State};
handle_cast({update_interval, Update = #update{name = Name}}, State = #state{updates = Updates}) -> handle_cast({update_interval, Update = #update{name = Name}},
State = #state{updates = Updates}) ->
case lists:keyfind(Name, #update.name, Updates) of case lists:keyfind(Name, #update.name, Updates) of
#update{} -> #update{} ->
?LOG(warning, "[Stats] Duplicated update: ~s", [Name]), ?LOG(warning, "[Stats] Duplicated update: ~s", [Name]),
@ -240,9 +247,9 @@ terminate(_Reason, #state{timer = TRef}) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
safe_update_element(Key, Val) -> safe_update_element(Key, Val) ->
try ets:update_element(?TAB, Key, {2, Val}) of try ets:update_element(?TAB, Key, {2, Val}) of
@ -254,3 +261,4 @@ safe_update_element(Key, Val) ->
error:badarg -> error:badarg ->
?LOG(warning, "[Stats] Update ~p to ~p failed", [Key, Val]) ?LOG(warning, "[Stats] Update ~p to ~p failed", [Key, Val])
end. end.

70
src/emqx_sup.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,11 +12,14 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_sup). -module(emqx_sup).
-behaviour(supervisor). -behaviour(supervisor).
-include("types.hrl").
-export([ start_link/0 -export([ start_link/0
, start_child/1 , start_child/1
, start_child/2 , start_child/2
@ -28,29 +32,28 @@
| {ok, supervisor:child(), term()} | {ok, supervisor:child(), term()}
| {error, term()}). | {error, term()}).
-define(SUPERVISOR, ?MODULE). -define(SUP, ?MODULE).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% API %% API
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-spec(start_link() -> startlink_ret()).
start_link() -> start_link() ->
supervisor:start_link({local, ?SUPERVISOR}, ?MODULE, []). supervisor:start_link({local, ?SUP}, ?MODULE, []).
-spec(start_child(supervisor:child_spec()) -> startchild_ret()). -spec(start_child(supervisor:child_spec()) -> startchild_ret()).
start_child(ChildSpec) when is_tuple(ChildSpec) -> start_child(ChildSpec) when is_tuple(ChildSpec) ->
supervisor:start_child(?SUPERVISOR, ChildSpec). supervisor:start_child(?SUP, ChildSpec).
-spec(start_child(module(), worker | supervisor) -> startchild_ret()). -spec(start_child(module(), worker | supervisor) -> startchild_ret()).
start_child(Mod, worker) -> start_child(Mod, Type) ->
start_child(worker_spec(Mod)); start_child(child_spec(Mod, Type)).
start_child(Mod, supervisor) ->
start_child(supervisor_spec(Mod)).
-spec(stop_child(supervisor:child_id()) -> ok | {error, term()}). -spec(stop_child(supervisor:child_id()) -> ok | {error, term()}).
stop_child(ChildId) -> stop_child(ChildId) ->
case supervisor:terminate_child(?SUPERVISOR, ChildId) of case supervisor:terminate_child(?SUP, ChildId) of
ok -> supervisor:delete_child(?SUPERVISOR, ChildId); ok -> supervisor:delete_child(?SUP, ChildId);
Error -> Error Error -> Error
end. end.
@ -60,32 +63,39 @@ stop_child(ChildId) ->
init([]) -> init([]) ->
%% Kernel Sup %% Kernel Sup
KernelSup = supervisor_spec(emqx_kernel_sup), KernelSup = child_spec(emqx_kernel_sup, supervisor),
%% Router Sup %% Router Sup
RouterSup = supervisor_spec(emqx_router_sup), RouterSup = child_spec(emqx_router_sup, supervisor),
%% Broker Sup %% Broker Sup
BrokerSup = supervisor_spec(emqx_broker_sup), BrokerSup = child_spec(emqx_broker_sup, supervisor),
BridgeSup = supervisor_spec(emqx_bridge_sup), %% Bridge Sup
%% Session Manager BridgeSup = child_spec(emqx_bridge_sup, supervisor),
SMSup = supervisor_spec(emqx_sm_sup), %% CM Sup
%% Connection Manager CMSup = child_spec(emqx_cm_sup, supervisor),
CMSup = supervisor_spec(emqx_cm_sup),
%% Sys Sup %% Sys Sup
SysSup = supervisor_spec(emqx_sys_sup), SysSup = child_spec(emqx_sys_sup, supervisor),
{ok, {{one_for_all, 0, 1}, {ok, {{one_for_all, 0, 1},
[KernelSup, [KernelSup, RouterSup, BrokerSup, BridgeSup, CMSup, SysSup]}}.
RouterSup,
BrokerSup,
BridgeSup,
SMSup,
CMSup,
SysSup]}}.
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
worker_spec(M) -> child_spec(Mod, supervisor) ->
{M, {M, start_link, []}, permanent, 30000, worker, [M]}. #{id => Mod,
supervisor_spec(M) -> start => {Mod, start_link, []},
{M, {M, start_link, []}, permanent, infinity, supervisor, [M]}. restart => permanent,
shutdown => infinity,
type => supervisor,
modules => [Mod]
};
child_spec(Mod, worker) ->
#{id => Mod,
start => {Mod, start_link, []},
restart => permanent,
shutdown => 15000,
type => worker,
modules => [Mod]
}.

4
src/emqx_sys.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_sys). -module(emqx_sys).

14
src/emqx_sys_mon.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_sys_mon). -module(emqx_sys_mon).
@ -41,18 +43,14 @@
-define(SYSMON, ?MODULE). -define(SYSMON, ?MODULE).
%%------------------------------------------------------------------------------ %% @doc Start the system monitor.
%% APIs
%%------------------------------------------------------------------------------
%% @doc Start system monitor
-spec(start_link(list(option())) -> startlink_ret()). -spec(start_link(list(option())) -> startlink_ret()).
start_link(Opts) -> start_link(Opts) ->
gen_server:start_link({local, ?SYSMON}, ?MODULE, [Opts], []). gen_server:start_link({local, ?SYSMON}, ?MODULE, [Opts], []).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([Opts]) -> init([Opts]) ->
erlang:system_monitor(self(), parse_opt(Opts)), erlang:system_monitor(self(), parse_opt(Opts)),

29
src/emqx_sys_sup.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_sys_sup). -module(emqx_sys_sup).
@ -24,23 +26,28 @@ start_link() ->
supervisor:start_link({local, ?MODULE}, ?MODULE, []). supervisor:start_link({local, ?MODULE}, ?MODULE, []).
init([]) -> init([]) ->
{ok, {{one_for_one, 10, 100}, [child_spec(emqx_sys, worker), Childs = [child_spec(emqx_sys),
child_spec(emqx_sys_mon, worker, [emqx_config:get_env(sysmon, [])]), child_spec(emqx_sys_mon, [config(sysmon)]),
child_spec(emqx_os_mon, worker, [emqx_config:get_env(os_mon, [])]), child_spec(emqx_os_mon, [config(os_mon)]),
child_spec(emqx_vm_mon, worker, [emqx_config:get_env(vm_mon, [])])]}}. child_spec(emqx_vm_mon, [config(vm_mon)])],
{ok, {{one_for_one, 10, 100}, Childs}}.
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
child_spec(M, worker) -> child_spec(Mod) ->
child_spec(M, worker, []). child_spec(Mod, []).
child_spec(M, worker, A) -> child_spec(Mod, Args) ->
#{id => M, #{id => Mod,
start => {M, start_link, A}, start => {Mod, start_link, Args},
restart => permanent, restart => permanent,
shutdown => 5000, shutdown => 5000,
type => worker, type => worker,
modules => [M]}. modules => [Mod]
}.
config(Name) ->
emqx_config:get_env(Name, []).

5
src/emqx_tables.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_tables). -module(emqx_tables).
@ -52,3 +54,4 @@ lookup_value(Tab, Key, Def) ->
catch catch
error:badarg -> Def error:badarg -> Def
end. end.

5
src/emqx_time.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_time). -module(emqx_time).
@ -39,3 +41,4 @@ now_ms({MegaSecs, Secs, MicroSecs}) ->
ts_from_ms(Ms) -> ts_from_ms(Ms) ->
{Ms div 1000000, Ms rem 1000000, 0}. {Ms div 1000000, Ms rem 1000000, 0}.

8
src/emqx_topic.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_topic). -module(emqx_topic).
@ -43,9 +45,9 @@
-define(MAX_TOPIC_LEN, 4096). -define(MAX_TOPIC_LEN, 4096).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Is wildcard topic? %% @doc Is wildcard topic?
-spec(wildcard(topic() | words()) -> true | false). -spec(wildcard(topic() | words()) -> true | false).

4
src/emqx_tracer.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_tracer). -module(emqx_tracer).

16
src/emqx_trie.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_trie). -module(emqx_trie).
@ -35,9 +37,9 @@
-define(TRIE, emqx_trie). -define(TRIE, emqx_trie).
-define(TRIE_NODE, emqx_trie_node). -define(TRIE_NODE, emqx_trie_node).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Mnesia bootstrap %% Mnesia bootstrap
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Create or replicate trie tables. %% @doc Create or replicate trie tables.
-spec(mnesia(boot | copy) -> ok). -spec(mnesia(boot | copy) -> ok).
@ -64,9 +66,9 @@ mnesia(copy) ->
%% Copy trie_node table %% Copy trie_node table
ok = ekka_mnesia:copy_table(?TRIE_NODE). ok = ekka_mnesia:copy_table(?TRIE_NODE).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Trie APIs %% Trie APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @doc Insert a topic filter into the trie. %% @doc Insert a topic filter into the trie.
-spec(insert(emqx_topic:topic()) -> ok). -spec(insert(emqx_topic:topic()) -> ok).
@ -111,9 +113,9 @@ delete(Topic) when is_binary(Topic) ->
empty() -> empty() ->
ets:info(?TRIE, size) == 0. ets:info(?TRIE, size) == 0.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% @private %% @private
%% @doc Add a path to the trie. %% @doc Add a path to the trie.

10
src/emqx_types.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_types). -module(emqx_types).
@ -32,9 +34,7 @@
, protocol/0 , protocol/0
]). ]).
-export_type([ credentials/0 -export_type([credentials/0]).
, session/0
]).
-export_type([ subscription/0 -export_type([ subscription/0
, subscriber/0 , subscriber/0
@ -65,7 +65,6 @@
share => binary(), share => binary(),
atom() => term() atom() => term()
}). }).
-type(session() :: #session{}).
-type(client_id() :: binary() | atom()). -type(client_id() :: binary() | atom()).
-type(username() :: maybe(binary())). -type(username() :: maybe(binary())).
-type(password() :: maybe(binary())). -type(password() :: maybe(binary())).
@ -105,3 +104,4 @@
-type(alarm() :: #alarm{}). -type(alarm() :: #alarm{}).
-type(plugin() :: #plugin{}). -type(plugin() :: #plugin{}).
-type(command() :: #command{}). -type(command() :: #command{}).

4
src/emqx_vm.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_vm). -module(emqx_vm).

58
src/emqx_vm_mon.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,11 +12,14 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_vm_mon). -module(emqx_vm_mon).
-behaviour(gen_server). -behaviour(gen_server).
-include("logger.hrl").
%% APIs %% APIs
-export([start_link/1]). -export([start_link/1]).
@ -38,13 +42,13 @@
-define(VM_MON, ?MODULE). -define(VM_MON, ?MODULE).
%%----------------------------------------------------------------------
%% API
%%----------------------------------------------------------------------
start_link(Opts) -> start_link(Opts) ->
gen_server:start_link({local, ?VM_MON}, ?MODULE, [Opts], []). gen_server:start_link({local, ?VM_MON}, ?MODULE, [Opts], []).
%%--------------------------------------------------------------------
%% API
%%--------------------------------------------------------------------
get_check_interval() -> get_check_interval() ->
call(get_check_interval). call(get_check_interval).
@ -63,9 +67,12 @@ get_process_low_watermark() ->
set_process_low_watermark(Float) -> set_process_low_watermark(Float) ->
call({set_process_low_watermark, Float}). call({set_process_low_watermark, Float}).
%%---------------------------------------------------------------------- call(Req) ->
gen_server:call(?VM_MON, Req, infinity).
%%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%---------------------------------------------------------------------- %%--------------------------------------------------------------------
init([Opts]) -> init([Opts]) ->
{ok, ensure_check_timer(#{check_interval => proplists:get_value(check_interval, Opts, 30), {ok, ensure_check_timer(#{check_interval => proplists:get_value(check_interval, Opts, 30),
@ -76,43 +83,53 @@ init([Opts]) ->
handle_call(get_check_interval, _From, State) -> handle_call(get_check_interval, _From, State) ->
{reply, maps:get(check_interval, State, undefined), State}; {reply, maps:get(check_interval, State, undefined), State};
handle_call({set_check_interval, Seconds}, _From, State) -> handle_call({set_check_interval, Seconds}, _From, State) ->
{reply, ok, State#{check_interval := Seconds}}; {reply, ok, State#{check_interval := Seconds}};
handle_call(get_process_high_watermark, _From, State) -> handle_call(get_process_high_watermark, _From, State) ->
{reply, maps:get(process_high_watermark, State, undefined), State}; {reply, maps:get(process_high_watermark, State, undefined), State};
handle_call({set_process_high_watermark, Float}, _From, State) -> handle_call({set_process_high_watermark, Float}, _From, State) ->
{reply, ok, State#{process_high_watermark := Float}}; {reply, ok, State#{process_high_watermark := Float}};
handle_call(get_process_low_watermark, _From, State) -> handle_call(get_process_low_watermark, _From, State) ->
{reply, maps:get(process_low_watermark, State, undefined), State}; {reply, maps:get(process_low_watermark, State, undefined), State};
handle_call({set_process_low_watermark, Float}, _From, State) -> handle_call({set_process_low_watermark, Float}, _From, State) ->
{reply, ok, State#{process_low_watermark := Float}}; {reply, ok, State#{process_low_watermark := Float}};
handle_call(_Request, _From, State) -> handle_call(Req, _From, State) ->
{reply, ok, State}. ?LOG(error, "[VM_MON] Unexpected call: ~p", [Req]),
{reply, ignored, State}.
handle_cast(_Request, State) -> handle_cast(Msg, State) ->
?LOG(error, "[VM_MON] Unexpected cast: ~p", [Msg]),
{noreply, State}. {noreply, State}.
handle_info({timeout, Timer, check}, State = #{timer := Timer, handle_info({timeout, Timer, check},
State = #{timer := Timer,
process_high_watermark := ProcHighWatermark, process_high_watermark := ProcHighWatermark,
process_low_watermark := ProcLowWatermark, process_low_watermark := ProcLowWatermark,
is_process_alarm_set := IsProcessAlarmSet}) -> is_process_alarm_set := IsProcessAlarmSet}) ->
ProcessCount = erlang:system_info(process_count), ProcessCount = erlang:system_info(process_count),
case ProcessCount / erlang:system_info(process_limit) of NState = case ProcessCount / erlang:system_info(process_limit) of
Percent when Percent >= ProcHighWatermark -> Percent when Percent >= ProcHighWatermark ->
alarm_handler:set_alarm({too_many_processes, ProcessCount}), alarm_handler:set_alarm({too_many_processes, ProcessCount}),
{noreply, ensure_check_timer(State#{is_process_alarm_set := true})}; State#{is_process_alarm_set := true};
Percent when Percent < ProcLowWatermark -> Percent when Percent < ProcLowWatermark ->
case IsProcessAlarmSet of case IsProcessAlarmSet of
true -> alarm_handler:clear_alarm(too_many_processes); true -> alarm_handler:clear_alarm(too_many_processes);
false -> ok false -> ok
end, end,
{noreply, ensure_check_timer(State#{is_process_alarm_set := false})}; State#{is_process_alarm_set := false};
_Precent -> _Precent -> State
{noreply, ensure_check_timer(State)} end,
end. {noreply, ensure_check_timer(NState)};
handle_info(Info, State) ->
?LOG(error, "[VM_MON] Unexpected info: ~p", [Info]),
{noreply, State}.
terminate(_Reason, #{timer := Timer}) -> terminate(_Reason, #{timer := Timer}) ->
emqx_misc:cancel_timer(Timer). emqx_misc:cancel_timer(Timer).
@ -120,11 +137,10 @@ terminate(_Reason, #{timer := Timer}) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%---------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%---------------------------------------------------------------------- %%--------------------------------------------------------------------
call(Req) ->
gen_server:call(?VM_MON, Req, infinity).
ensure_check_timer(State = #{check_interval := Interval}) -> ensure_check_timer(State = #{check_interval := Interval}) ->
State#{timer := emqx_misc:start_timer(timer:seconds(Interval), check)}. State#{timer := emqx_misc:start_timer(timer:seconds(Interval), check)}.

16
src/emqx_zone.erl
View File

@ -1,4 +1,5 @@
%% Copyright (c) 2013-2019 EMQ Technologies Co., Ltd. All Rights Reserved. %%--------------------------------------------------------------------
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -11,6 +12,7 @@
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and %% See the License for the specific language governing permissions and
%% limitations under the License. %% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_zone). -module(emqx_zone).
@ -48,9 +50,9 @@
-define(SERVER, ?MODULE). -define(SERVER, ?MODULE).
-define(KEY(Zone, Key), {?MODULE, Zone, Key}). -define(KEY(Zone, Key), {?MODULE, Zone, Key}).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
-spec(start_link() -> startlink_ret()). -spec(start_link() -> startlink_ret()).
start_link() -> start_link() ->
@ -83,9 +85,9 @@ force_reload() ->
stop() -> stop() ->
gen_server:stop(?SERVER, normal, infinity). gen_server:stop(?SERVER, normal, infinity).
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
init([]) -> init([]) ->
_ = do_reload(), _ = do_reload(),
@ -117,9 +119,9 @@ terminate(_Reason, _State) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
%% Internal functions %% Internal functions
%%------------------------------------------------------------------------------ %%--------------------------------------------------------------------
do_reload() -> do_reload() ->
[ persistent_term:put(?KEY(Zone, Key), Val) [ persistent_term:put(?KEY(Zone, Key), Val)