Merge pull request #6 from emqx/emqx30

Emqx30

.gitignore

@@ -32,3 +32,5 @@ rebar3.crashdump
 .DS_Store
 rebar.config
 emqx.iml
+bbmustache/
+etc/gen.emqx.conf

Makefile

@@ -37,7 +37,7 @@ EUNIT_OPTS = verbose
 
 CT_SUITES = emqx emqx_banned emqx_connection emqx_session emqx_access emqx_broker emqx_cm emqx_frame emqx_guid emqx_inflight \
 			emqx_json emqx_keepalive emqx_lib emqx_metrics emqx_misc emqx_mod emqx_mqtt_caps \
-			emqx_mqtt_compat emqx_mqtt_properties emqx_mqueue emqx_net emqx_pqueue emqx_router emqx_sm \
+			emqx_mqtt_compat emqx_mqtt_props emqx_mqueue emqx_net emqx_pqueue emqx_router emqx_sm \
 			emqx_stats emqx_tables emqx_time emqx_topic emqx_trie emqx_vm emqx_zone \
 		 	emqx_mountpoint emqx_listeners emqx_protocol
 
@@ -51,6 +51,29 @@ DIALYZER_OPTS := --verbose --statistics -Werror_handling -Wrace_conditions #-Wun
 
 include erlang.mk
 
-app.config::
+clean:: gen-clean
-	./deps/cuttlefish/cuttlefish -l info -e etc/ -c etc/emqx.conf -i priv/emqx.schema -d data/
+
+.PHONY: gen-clean
+gen-clean:
+	@rm -rf bbmustache
+	@rm -f etc/gen.emqx.conf
+
+bbmustache:
+	$(verbose) git clone https://github.com/soranoba/bbmustache.git && pushd bbmustache && ./rebar3 compile && popd
+
+# This hack is to generate a conf file for testing
+# relx overlay is used for release
+etc/gen.emqx.conf: bbmustache etc/emqx.conf
+	$(verbose) erl -noshell -pa bbmustache/_build/default/lib/bbmustache/ebin -eval \
+		"{ok, Temp} = file:read_file('etc/emqx.conf'), \
+		{ok, Vars0} = file:consult('vars'), \
+		Vars = [{atom_to_list(N), list_to_binary(V)} || {N, V} <- Vars0], \
+		Targ = bbmustache:render(Temp, Vars), \
+		ok = file:write_file('etc/gen.emqx.conf', Targ), \
+		halt(0)."
+
+app.config: etc/gen.emqx.conf
+	$(verbose) ./deps/cuttlefish/cuttlefish -l info -e etc/ -c etc/gen.emqx.conf -i priv/emqx.schema -d data/
+
+ct: app.config
 

README.md

@@ -1,7 +1,6 @@
-# *EMQ X* - MQTT Broker
+# EMQ X Broker
 
-
+*EMQ X* broker is a fully open source, highly scalable, highly available distributed MQTT messaging broker for IoT, M2M and Mobile applications that can handle tens of millions of concurrent clients.
-*EMQ X* broker is a fully open source, highly scalable, highly available distributed message broker for IoT, M2M and Mobile applications that can handle tens of millions of concurrent clients.
 
 Starting from 3.0 release, *EMQ X* broker fully supports MQTT V5.0 protocol specifications and backward compatible with MQTT V3.1 and V3.1.1,  as well as other communication protocols such as MQTT-SN, CoAP, LwM2M, WebSocket and STOMP. The 3.0 release of the *EMQ X* broker can scaled to 10+ million concurrent MQTT connections on one cluster.
 

include/emqx.hrl

@@ -148,13 +148,16 @@
 %%--------------------------------------------------------------------
 %% Banned
 %%--------------------------------------------------------------------
+-type(banned_who() ::  {client_id,  binary()}
+                     | {username,   binary()}
+                     | {ip_address, inet:ip_address()}).
 
 -record(banned, {
-          key, 
+          who    :: banned_who(),
-          reason, 
+          reason :: binary(),
-          by, 
+          by     :: binary(),
-          desc, 
+          desc   :: binary(),
-          until}).
+          until  :: integer()
+        }).
 
 -endif.
-

src/emqx.erl

@@ -29,16 +29,16 @@
 -export([get_subopts/2, set_subopts/3]).
 
 %% Hooks API
--export([hook/4, hook/3, unhook/2, run_hooks/2, run_hooks/3]).
+-export([hook/2, hook/3, hook/4, unhook/2, run_hooks/2, run_hooks/3]).
 
 %% Shutdown and reboot
 -export([shutdown/0, shutdown/1, reboot/0]).
 
 -define(APP, ?MODULE).
 
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 %% Bootstrap, is_running...
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
 %% @doc Start emqx application
 -spec(start() -> {ok, list(atom())} | {error, term()}).
@@ -62,9 +62,9 @@ is_running(Node) ->
         Pid when is_pid(Pid) -> true
     end.
 
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 %% PubSub API
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
 -spec(subscribe(emqx_topic:topic() | string()) -> ok).
 subscribe(Topic) ->
@@ -97,9 +97,9 @@ unsubscribe(Topic, SubId) when is_atom(SubId); is_binary(SubId) ->
 unsubscribe(Topic, SubPid) when is_pid(SubPid) ->
     emqx_broker:unsubscribe(iolist_to_binary(Topic), SubPid).
 
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 %% PubSub management API
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
 -spec(get_subopts(emqx_topic:topic() | string(), emqx_types:subscriber())
       -> emqx_types:subopts()).
@@ -128,36 +128,43 @@ subscribed(Topic, SubPid) when is_pid(SubPid) ->
 subscribed(Topic, SubId) when is_atom(SubId); is_binary(SubId) ->
     emqx_broker:subscribed(iolist_to_binary(Topic), SubId).
 
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 %% Hooks API
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
--spec(hook(atom(), function() | {emqx_hooks:hooktag(), function()}, list(any()))
+-spec(hook(emqx_hooks:hookpoint(), emqx_hooks:action()) -> ok | {error, already_exists}).
-      -> ok | {error, term()}).
+hook(HookPoint, Action) ->
-hook(Hook, TagFunction, InitArgs) ->
+    emqx_hooks:add(HookPoint, Action).
-    emqx_hooks:add(Hook, TagFunction, InitArgs).
 
--spec(hook(atom(), function() | {emqx_hooks:hooktag(), function()}, list(any()), integer())
+-spec(hook(emqx_hooks:hookpoint(), emqx_hooks:action(), emqx_hooks:filter() | integer())
-      -> ok | {error, term()}).
+      -> ok | {error, already_exists}).
-hook(Hook, TagFunction, InitArgs, Priority) ->
+hook(HookPoint, Action, Priority) when is_integer(Priority) ->
-    emqx_hooks:add(Hook, TagFunction, InitArgs, Priority).
+    emqx_hooks:add(HookPoint, Action, Priority);
+hook(HookPoint, Action, Filter) when is_function(Filter); is_tuple(Filter) ->
+    emqx_hooks:add(HookPoint, Action, Filter);
+hook(HookPoint, Action, InitArgs) when is_list(InitArgs) ->
+    emqx_hooks:add(HookPoint, Action, InitArgs).
 
--spec(unhook(atom(), function() | {emqx_hooks:hooktag(), function()})
+-spec(hook(emqx_hooks:hookpoint(), emqx_hooks:action(), emqx_hooks:filter(), integer())
-      -> ok | {error, term()}).
+      -> ok | {error, already_exists}).
-unhook(Hook, TagFunction) ->
+hook(HookPoint, Action, Filter, Priority) ->
-    emqx_hooks:delete(Hook, TagFunction).
+    emqx_hooks:add(HookPoint, Action, Filter, Priority).
 
--spec(run_hooks(atom(), list(any())) -> ok | stop).
+-spec(unhook(emqx_hooks:hookpoint(), emqx_hooks:action()) -> ok).
-run_hooks(Hook, Args) ->
+unhook(HookPoint, Action) ->
-    emqx_hooks:run(Hook, Args).
+    emqx_hooks:del(HookPoint, Action).
 
--spec(run_hooks(atom(), list(any()), any()) -> {ok | stop, any()}).
+-spec(run_hooks(emqx_hooks:hookpoint(), list(any())) -> ok | stop).
-run_hooks(Hook, Args, Acc) ->
+run_hooks(HookPoint, Args) ->
-    emqx_hooks:run(Hook, Args, Acc).
+    emqx_hooks:run(HookPoint, Args).
 
-%%--------------------------------------------------------------------
+-spec(run_hooks(emqx_hooks:hookpoint(), list(any()), any()) -> {ok | stop, any()}).
+run_hooks(HookPoint, Args, Acc) ->
+    emqx_hooks:run(HookPoint, Args, Acc).
+
+%%------------------------------------------------------------------------------
 %% Shutdown and reboot
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
 shutdown() ->
     shutdown(normal).

src/emqx_access_control.erl

@@ -154,7 +154,7 @@ init([]) ->
 handle_call({register_mod, Type, Mod, Opts, Seq}, _From, State) ->
     Mods = lookup_mods(Type),
     reply(case lists:keymember(Mod, 1, Mods) of
-              true  -> {error, already_existed};
+              true  -> {error, already_exists};
               false ->
                   case catch Mod:init(Opts) of
                       {ok, ModState} ->
@@ -183,7 +183,7 @@ handle_call(stop, _From, State) ->
 
 handle_call(Req, _From, State) ->
     emqx_logger:error("[AccessControl] unexpected request: ~p", [Req]),
-    {reply, ignore, State}.
+    {reply, ignored, State}.
 
 handle_cast(Msg, State) ->
     emqx_logger:error("[AccessControl] unexpected msg: ~p", [Msg]),

src/emqx_banned.erl

@@ -85,7 +85,7 @@ handle_cast(Msg, State) ->
     {noreply, State}.
 
 handle_info({timeout, TRef, expire}, State = #{expiry_timer := TRef}) ->
-    mnesia:async_dirty(fun expire_banned_items/1, [erlang:timestamp()]),
+    mnesia:async_dirty(fun expire_banned_items/1, [erlang:system_time(second)]),
     {noreply, ensure_expiry_timer(State), hibernate};
 
 handle_info(Info, State) ->
@@ -106,17 +106,8 @@ ensure_expiry_timer(State) ->
     State#{expiry_timer := emqx_misc:start_timer(timer:minutes(5), expire)}.
 
 expire_banned_items(Now) ->
-    expire_banned_item(mnesia:first(?TAB), Now).
+    mnesia:foldl(fun
-
+            (B = #banned{until = Until}, _Acc) when Until < Now ->
-expire_banned_item('$end_of_table', _Now) ->
+                mnesia:delete_object(?TAB, B, sticky_write);
-    ok;
+            (_, _Acc) -> ok
-expire_banned_item(Key, Now) ->
+        end, ok, ?TAB).
-    case mnesia:read(?TAB, Key) of
-        [#banned{until = undefined}] ->
-            ok;
-        [B = #banned{until = Until}] when Until < Now ->
-           mnesia:delete_object(?TAB, B, sticky_write);
-        _ -> ok
-    end,
-    expire_banned_item(mnesia:next(?TAB, Key), Now).
-

src/emqx_client.erl

@@ -159,7 +159,7 @@ start_link() -> start_link([]).
 start_link(Options) when is_map(Options) ->
     start_link(maps:to_list(Options));
 start_link(Options) when is_list(Options) ->
-    ok  = emqx_mqtt_properties:validate(
+    ok  = emqx_mqtt_props:validate(
             proplists:get_value(properties, Options, #{})),
     case start_client(with_owner(Options)) of
         {ok, Client} ->
@@ -265,7 +265,7 @@ publish(Client, Topic, Payload, Opts) when is_binary(Topic), is_list(Opts) ->
       -> ok | {ok, packet_id()} | {error, term()}).
 publish(Client, Topic, Properties, Payload, Opts)
     when is_binary(Topic), is_map(Properties), is_list(Opts) ->
-    ok = emqx_mqtt_properties:validate(Properties),
+    ok = emqx_mqtt_props:validate(Properties),
     Retain = proplists:get_bool(retain, Opts),
     QoS = ?QOS_I(proplists:get_value(qos, Opts, ?QOS_0)),
     publish(Client, #mqtt_msg{qos     = QoS,
@@ -541,7 +541,7 @@ mqtt_connect(State = #state{client_id   = ClientId,
                             will_msg    = WillMsg,
                             properties  = Properties}) ->
     ?WILL_MSG(WillQoS, WillRetain, WillTopic, WillProps, WillPayload) = WillMsg,
-    ConnProps = emqx_mqtt_properties:filter(?CONNECT, Properties),
+    ConnProps = emqx_mqtt_props:filter(?CONNECT, Properties),
     send(?CONNECT_PACKET(
             #mqtt_packet_connect{proto_ver    = ProtoVer,
                                  proto_name   = ProtoName,

src/emqx_hooks.erl

@@ -16,142 +16,160 @@
 
 -behaviour(gen_server).
 
--export([start_link/0]).
+-export([start_link/0, stop/0]).
 
 %% Hooks API
--export([add/3, add/4, delete/2, run/2, run/3, lookup/1]).
+-export([add/2, add/3, add/4, del/2, run/2, run/3, lookup/1]).
 
 %% gen_server Function Exports
--export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
-         terminate/2, code_change/3]).
+         code_change/3]).
 
--record(state, {}).
+-type(hookpoint() :: atom()).
+-type(action() :: function() | mfa()).
+-type(filter() :: function() | mfa()).
 
--type(hooktag() :: atom() | string() | binary()).
+-record(callback, {action   :: action(),
+                   filter   :: filter(),
+                   priority :: integer()}).
 
--export_type([hooktag/0]).
+-record(hook, {name :: hookpoint(), callbacks :: list(#callback{})}).
 
--record(callback, {tag            :: hooktag(),
+-export_type([hookpoint/0, action/0, filter/0]).
-                   function       :: function(),
-                   init_args = [] :: list(any()),
-                   priority  = 0  :: integer()}).
-
--record(hook, {name :: atom(), callbacks = [] :: list(#callback{})}).
 
 -define(TAB, ?MODULE).
+-define(SERVER, ?MODULE).
 
+-spec(start_link() -> emqx_types:startlink_ret()).
 start_link() ->
-    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], [{hibernate_after, 60000}]).
 
-%%--------------------------------------------------------------------
+-spec(stop() -> ok).
+stop() ->
+    gen_server:stop(?SERVER, normal, infinity).
+
+%%------------------------------------------------------------------------------
 %% Hooks API
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
--spec(add(atom(), function() | {hooktag(), function()}, list(any())) -> ok).
+%% @doc Register a callback
-add(HookPoint, Function, InitArgs) when is_function(Function) ->
+-spec(add(hookpoint(), action() | #callback{}) -> emqx_types:ok_or_error(already_exists)).
-    add(HookPoint, {undefined, Function}, InitArgs, 0);
+add(HookPoint, Callback) when is_record(Callback, callback) ->
+    gen_server:call(?SERVER, {add, HookPoint, Callback}, infinity);
+add(HookPoint, Action) when is_function(Action); is_tuple(Action) ->
+    add(HookPoint, #callback{action = Action, priority = 0}).
 
-add(HookPoint, {Tag, Function}, InitArgs) when is_function(Function) ->
+-spec(add(hookpoint(), action(), filter() | integer() | list())
-    add(HookPoint, {Tag, Function}, InitArgs, 0).
+      -> emqx_types:ok_or_error(already_exists)).
+add(HookPoint, Action, InitArgs) when is_function(Action), is_list(InitArgs) ->
+    add(HookPoint, #callback{action = {Action, InitArgs}, priority = 0});
+add(HookPoint, Action, Filter) when is_function(Filter); is_tuple(Filter) ->
+    add(HookPoint, #callback{action = Action, filter = Filter, priority = 0});
+add(HookPoint, Action, Priority) when is_integer(Priority) ->
+    add(HookPoint, #callback{action = Action, priority = Priority}).
 
--spec(add(atom(), function() | {hooktag(), function()}, list(any()), integer()) -> ok).
+-spec(add(hookpoint(), action(), filter(), integer())
-add(HookPoint, Function, InitArgs, Priority) when is_function(Function) ->
+      -> emqx_types:ok_or_error(already_exists)).
-    add(HookPoint, {undefined, Function}, InitArgs, Priority);
+add(HookPoint, Action, Filter, Priority) ->
-add(HookPoint, {Tag, Function}, InitArgs, Priority) when is_function(Function) ->
+    add(HookPoint, #callback{action = Action, filter = Filter, priority = Priority}).
-    gen_server:call(?MODULE, {add, HookPoint, {Tag, Function}, InitArgs, Priority}).
 
--spec(delete(atom(), function() | {hooktag(), function()}) -> ok).
+%% @doc Unregister a callback.
-delete(HookPoint, Function) when is_function(Function) ->
+-spec(del(hookpoint(), action()) -> ok).
-    delete(HookPoint, {undefined, Function});
+del(HookPoint, Action) ->
-delete(HookPoint, {Tag, Function}) when is_function(Function) ->
+    gen_server:cast(?SERVER, {del, HookPoint, Action}).
-    gen_server:call(?MODULE, {delete, HookPoint, {Tag, Function}}).
 
-%% @doc Run hooks without Acc.
+%% @doc Run hooks.
 -spec(run(atom(), list(Arg :: any())) -> ok | stop).
 run(HookPoint, Args) ->
     run_(lookup(HookPoint), Args).
 
+%% @doc Run hooks with Accumulator.
 -spec(run(atom(), list(Arg :: any()), any()) -> any()).
 run(HookPoint, Args, Acc) ->
     run_(lookup(HookPoint), Args, Acc).
 
 %% @private
-run_([#callback{function = Fun, init_args = InitArgs} | Callbacks], Args) ->
+run_([#callback{action = Action, filter = Filter} | Callbacks], Args) ->
-    case apply(Fun, lists:append([Args, InitArgs])) of
+    case filtered(Filter, Args) orelse execute(Action, Args) of
+        true -> run_(Callbacks, Args);
         ok   -> run_(Callbacks, Args);
         stop -> stop;
         _Any -> run_(Callbacks, Args)
     end;
-
 run_([], _Args) ->
     ok.
 
 %% @private
-run_([#callback{function = Fun, init_args = InitArgs} | Callbacks], Args, Acc) ->
+run_([#callback{action = Action, filter = Filter} | Callbacks], Args, Acc) ->
-    case apply(Fun, lists:append([Args, [Acc], InitArgs])) of
+    Args1 = Args ++ [Acc],
+    case filtered(Filter, Args1) orelse execute(Action, Args1) of
+        true           -> run_(Callbacks, Args, Acc);
         ok             -> run_(Callbacks, Args, Acc);
         {ok, NewAcc}   -> run_(Callbacks, Args, NewAcc);
         stop           -> {stop, Acc};
         {stop, NewAcc} -> {stop, NewAcc};
         _Any           -> run_(Callbacks, Args, Acc)
     end;
-
 run_([], _Args, Acc) ->
     {ok, Acc}.
 
--spec(lookup(atom()) -> [#callback{}]).
+filtered(undefined, _Args) ->
+    false;
+filtered(Filter, Args) ->
+    execute(Filter, Args).
+
+execute(Action, Args) when is_function(Action) ->
+    erlang:apply(Action, Args);
+execute({Fun, InitArgs}, Args) when is_function(Fun) ->
+    erlang:apply(Fun, Args ++ InitArgs);
+execute({M, F, A}, Args) ->
+    erlang:apply(M, F, Args ++ A).
+
+%% @doc Lookup callbacks.
+-spec(lookup(hookpoint()) -> [#callback{}]).
 lookup(HookPoint) ->
     case ets:lookup(?TAB, HookPoint) of
-        [#hook{callbacks = Callbacks}] -> Callbacks;
+        [#hook{callbacks = Callbacks}] ->
+            Callbacks;
         [] -> []
     end.
 
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 %% gen_server callbacks
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
 init([]) ->
-    _ = emqx_tables:new(?TAB, [set, protected, {keypos, #hook.name},
+    _ = emqx_tables:new(?TAB, [{keypos, #hook.name}, {read_concurrency, true}]),
-                               {read_concurrency, true}]),
+    {ok, #{}}.
-    {ok, #state{}}.
 
-handle_call({add, HookPoint, {Tag, Function}, InitArgs, Priority}, _From, State) ->
+handle_call({add, HookPoint, Callback = #callback{action = Action}}, _From, State) ->
-    Callback = #callback{tag = Tag, function = Function,
+    Reply = case lists:keymember(Action, 2, Callbacks = lookup(HookPoint)) of
-                         init_args = InitArgs, priority = Priority},
+                true ->
-    {reply,
+                    {error, already_exists};
-     case ets:lookup(?TAB, HookPoint) of
+                false ->
-         [#hook{callbacks = Callbacks}] ->
+                    insert_hook(HookPoint, add_callback(Callback, Callbacks))
-             case contain_(Tag, Function, Callbacks) of
+            end,
-                 false ->
+    {reply, Reply, State};
-                     insert_hook_(HookPoint, add_callback_(Callback, Callbacks));
-                 true  ->
-                     {error, already_hooked}
-             end;
-         [] ->
-             insert_hook_(HookPoint, [Callback])
-     end, State};
-
-handle_call({delete, HookPoint, {Tag, Function}}, _From, State) ->
-    {reply,
-     case ets:lookup(?TAB, HookPoint) of
-         [#hook{callbacks = Callbacks}] ->
-             case contain_(Tag, Function, Callbacks) of
-                 true  ->
-                     insert_hook_(HookPoint, del_callback_(Tag, Function, Callbacks));
-                 false ->
-                     {error, not_found}
-             end;
-         [] ->
-             {error, not_found}
-     end, State};
 
 handle_call(Req, _From, State) ->
-    {reply, {error, {unexpected_request, Req}}, State}.
+    emqx_logger:error("[Hooks] unexpected call: ~p", [Req]),
+    {reply, ignored, State}.
 
-handle_cast(_Msg, State) ->
+handle_cast({del, HookPoint, Action}, State) ->
+    case del_callback(Action, lookup(HookPoint)) of
+        [] ->
+            ets:delete(?TAB, HookPoint);
+        Callbacks ->
+            insert_hook(HookPoint, Callbacks)
+    end,
+    {noreply, State};
+
+handle_cast(Msg, State) ->
+    emqx_logger:error("[Hooks] unexpected msg: ~p", [Msg]),
     {noreply, State}.
 
-handle_info(_Info, State) ->
+handle_info(Info, State) ->
+    emqx_logger:error("[Hooks] unexpected info: ~p", [Info]),
     {noreply, State}.
 
 terminate(_Reason, _State) ->
@@ -160,26 +178,35 @@ terminate(_Reason, _State) ->
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 %% Internal functions
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
-insert_hook_(HookPoint, Callbacks) ->
+insert_hook(HookPoint, Callbacks) ->
     ets:insert(?TAB, #hook{name = HookPoint, callbacks = Callbacks}), ok.
 
-add_callback_(Callback, Callbacks) ->
+add_callback(C, Callbacks) ->
-    lists:keymerge(#callback.priority, Callbacks, [Callback]).
+    add_callback(C, Callbacks, []).
 
-del_callback_(Tag, Function, Callbacks) ->
+add_callback(C, [], Acc) ->
-    lists:filter(
+    lists:reverse([C|Acc]);
-      fun(#callback{tag = Tag1, function = Func1}) ->
+add_callback(C1 = #callback{priority = P1}, [C2 = #callback{priority = P2}|More], Acc)
-        not ((Tag =:= Tag1) andalso (Function =:= Func1))
+    when P1 =< P2 ->
-      end, Callbacks).
+    add_callback(C1, More, [C2|Acc]);
+add_callback(C1, More, Acc) ->
+    lists:append(lists:reverse(Acc), [C1 | More]).
 
-contain_(_Tag, _Function, []) ->
+del_callback(Action, Callbacks) ->
-    false;
+    del_callback(Action, Callbacks, []).
-contain_(Tag, Function, [#callback{tag = Tag, function = Function}|_Callbacks]) ->
+
-    true;
+del_callback(_Action, [], Acc) ->
-contain_(Tag, Function, [_Callback | Callbacks]) ->
+    lists:reverse(Acc);
-    contain_(Tag, Function, Callbacks).
+del_callback(Action, [#callback{action = Action} | Callbacks], Acc) ->
+    del_callback(Action, Callbacks, Acc);
+del_callback(Action = {M, F}, [#callback{action = {M, F, _A}} | Callbacks], Acc) ->
+    del_callback(Action, Callbacks, Acc);
+del_callback(Func, [#callback{action = {Func, _A}} | Callbacks], Acc) ->
+    del_callback(Func, Callbacks, Acc);
+del_callback(Action, [Callback | Callbacks], Acc) ->
+    del_callback(Action, Callbacks, [Callback | Acc]).
 

src/emqx_mod_presence.erl

@@ -19,24 +19,24 @@
 -include("emqx.hrl").
 
 -export([load/1, unload/1]).
+
 -export([on_client_connected/4, on_client_disconnected/3]).
 
+-define(ATTR_KEYS, [clean_start, proto_ver, proto_name, keepalive]).
+
 load(Env) ->
     emqx_hooks:add('client.connected',    fun ?MODULE:on_client_connected/4, [Env]),
     emqx_hooks:add('client.disconnected', fun ?MODULE:on_client_disconnected/3, [Env]).
 
 on_client_connected(#{client_id := ClientId,
                       username  := Username,
-                      peername  := {IpAddr, _}}, ConnAck, ConnInfo, Env) ->
+                      peername  := {IpAddr, _}}, ConnAck, ConnAttrs, Env) ->
+    Attrs = lists:filter(fun({K, _}) -> lists:member(K, ?ATTR_KEYS) end, ConnAttrs),
     case emqx_json:safe_encode([{clientid, ClientId},
                                 {username, Username},
                                 {ipaddress, iolist_to_binary(esockd_net:ntoa(IpAddr))},
-                                {clean_start, proplists:get_value(clean_start, ConnInfo)},
-                                {proto_ver, proplists:get_value(proto_ver, ConnInfo)},
-                                {proto_name, proplists:get_value(proto_name, ConnInfo)},
-                                {keepalive, proplists:get_value(keepalive, ConnInfo)},
                                 {connack, ConnAck},
-                                {ts, os:system_time(second)}]) of
+                                {ts, os:system_time(second)} | Attrs]) of
         {ok, Payload} ->
             emqx:publish(message(qos(Env), topic(connected, ClientId), Payload));
         {error, Reason} ->
@@ -55,20 +55,20 @@ on_client_disconnected(#{client_id := ClientId, username := Username}, Reason, E
     end.
 
 unload(_Env) ->
-    emqx_hooks:delete('client.connected',    fun ?MODULE:on_client_connected/4),
+    emqx_hooks:del('client.connected',    fun ?MODULE:on_client_connected/4),
-    emqx_hooks:delete('client.disconnected', fun ?MODULE:on_client_disconnected/3).
+    emqx_hooks:del('client.disconnected', fun ?MODULE:on_client_disconnected/3).
 
 message(QoS, Topic, Payload) ->
-    Msg = emqx_message:make(?MODULE, QoS, Topic, iolist_to_binary(Payload)),
+    emqx_message:set_flag(
-    emqx_message:set_flag(sys, Msg).
+      sys, emqx_message:make(
+             ?MODULE, QoS, Topic, iolist_to_binary(Payload))).
 
 topic(connected, ClientId) ->
     emqx_topic:systop(iolist_to_binary(["clients/", ClientId, "/connected"]));
 topic(disconnected, ClientId) ->
     emqx_topic:systop(iolist_to_binary(["clients/", ClientId, "/disconnected"])).
 
-qos(Env) ->
+qos(Env) -> proplists:get_value(qos, Env, 0).
-    proplists:get_value(qos, Env, 0).
 
 reason(Reason) when is_atom(Reason) -> Reason;
 reason({Error, _}) when is_atom(Error) -> Error;

src/emqx_mod_rewrite.erl

@@ -21,11 +21,15 @@
 
 -export([rewrite_subscribe/3, rewrite_unsubscribe/3, rewrite_publish/2]).
 
-load(Rules0) ->
+%%------------------------------------------------------------------------------
-    Rules = compile(Rules0),
+%% Load/Unload
-    emqx_hooks:add('client.subscribe',  fun ?MODULE:rewrite_subscribe/3, [Rules]),
+%%------------------------------------------------------------------------------
-    emqx_hooks:add('client.unsubscribe',fun ?MODULE:rewrite_unsubscribe/3, [Rules]),
+
-    emqx_hooks:add('message.publish',   fun ?MODULE:rewrite_publish/2, [Rules]).
+load(RawRules) ->
+    Rules = compile(RawRules),
+    emqx_hooks:add('client.subscribe',   fun ?MODULE:rewrite_subscribe/3, [Rules]),
+    emqx_hooks:add('client.unsubscribe', fun ?MODULE:rewrite_unsubscribe/3, [Rules]),
+    emqx_hooks:add('message.publish',    fun ?MODULE:rewrite_publish/2, [Rules]).
 
 rewrite_subscribe(_Credentials, TopicTable, Rules) ->
     {ok, [{match_rule(Topic, Rules), Opts} || {Topic, Opts} <- TopicTable]}.
@@ -37,13 +41,13 @@ rewrite_publish(Message = #message{topic = Topic}, Rules) ->
     {ok, Message#message{topic = match_rule(Topic, Rules)}}.
 
 unload(_) ->
-    emqx_hooks:delete('client.subscribe',  fun ?MODULE:rewrite_subscribe/3),
+    emqx_hooks:del('client.subscribe',   fun ?MODULE:rewrite_subscribe/3),
-    emqx_hooks:delete('client.unsubscribe',fun ?MODULE:rewrite_unsubscribe/3),
+    emqx_hooks:del('client.unsubscribe', fun ?MODULE:rewrite_unsubscribe/3),
-    emqx_hooks:delete('message.publish',   fun ?MODULE:rewrite_publish/2).
+    emqx_hooks:del('message.publish',    fun ?MODULE:rewrite_publish/2).
 
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 %% Internal functions
-%%--------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
 match_rule(Topic, []) ->
     Topic;

src/emqx_mqtt_caps.erl

@@ -43,7 +43,9 @@
                        {mqtt_wildcard_subscription, true}]).
 
 -define(PUBCAP_KEYS, [max_qos_allowed,
-                      mqtt_retain_available]).
+                      mqtt_retain_available,
+                      mqtt_topic_alias
+                     ]).
 -define(SUBCAP_KEYS, [max_qos_allowed,
                       max_topic_levels,
                       mqtt_shared_subscription,
@@ -60,8 +62,15 @@ do_check_pub(Props = #{qos := QoS}, [{max_qos_allowed, MaxQoS}|Caps]) ->
         true  -> {error, ?RC_QOS_NOT_SUPPORTED};
         false -> do_check_pub(Props, Caps)
     end;
+do_check_pub(Props = #{ topic_alias := TopicAlias}, [{max_topic_alias, MaxTopicAlias}| Caps]) ->
+    case TopicAlias =< MaxTopicAlias andalso TopicAlias > 0 of
+        false -> {error, ?RC_TOPIC_ALIAS_INVALID};
+        true -> do_check_pub(Props, Caps)
+    end;
 do_check_pub(#{retain := true}, [{mqtt_retain_available, false}|_Caps]) ->
     {error, ?RC_RETAIN_NOT_SUPPORTED};
+do_check_pub(Props, [{max_topic_alias, _} | Caps]) ->
+    do_check_pub(Props, Caps);
 do_check_pub(Props, [{mqtt_retain_available, _}|Caps]) ->
     do_check_pub(Props, Caps).
 
@@ -136,4 +145,3 @@ with_env(Zone, Key, InitFun) ->
                      Caps;
         ZoneCaps  -> ZoneCaps
     end.
-

src/emqx_mqtt_props.erl

@@ -13,7 +13,7 @@
 %% limitations under the License.
 
 %% @doc MQTT5 Properties
--module(emqx_mqtt_properties).
+-module(emqx_mqtt_props).
 
 -include("emqx_mqtt.hrl").
 

src/emqx_packet.erl

@@ -72,9 +72,11 @@ validate_packet_id(_) ->
 validate_properties(?SUBSCRIBE, #{'Subscription-Identifier' := I})
     when I =< 0; I >= 16#FFFFFFF ->
     error(subscription_identifier_invalid);
-validate_properties(?PUBLISH, # {'Topic-Alias':= I})
+validate_properties(?PUBLISH, #{'Topic-Alias':= I})
     when I =:= 0 ->
     error(topic_alias_invalid);
+validate_properties(?PUBLISH, #{'Subscription-Identifier' := _I}) ->
+    error(protocol_error);
 validate_properties(_, _) ->
     true.
 
@@ -236,4 +238,3 @@ format_password(_Password) -> '******'.
 i(true)  -> 1;
 i(false) -> 0;
 i(I) when is_integer(I) -> I.
-

src/emqx_pool.erl

@@ -23,16 +23,14 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
          code_change/3]).
 
--record(state, {pool, id}).
-
 -define(POOL, ?MODULE).
 
 %% @doc Start pooler supervisor.
 start_link() ->
     emqx_pool_sup:start_link(?POOL, random, {?MODULE, start_link, []}).
 
-%% @doc Start pool
+%% @doc Start pool.
--spec(start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, term()}).
+-spec(start_link(atom(), pos_integer()) -> emqx_types:startlink_ret()).
 start_link(Pool, Id) ->
     gen_server:start_link({local, emqx_misc:proc_name(?MODULE, Id)}, ?MODULE, [Pool, Id], []).
 
@@ -49,13 +47,13 @@ async_submit(Fun) ->
 worker() ->
     gproc_pool:pick_worker(pool).
 
-%%-----------------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 %% gen_server callbacks
-%%-----------------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
 init([Pool, Id]) ->
     true = gproc_pool:connect_worker(Pool, {Pool, Id}),
-    {ok, #state{pool = Pool, id = Id}}.
+    {ok, #{pool => Pool, id => Id}}.
 
 handle_call({submit, Fun}, _From, State) ->
     {reply, catch run(Fun), State};
@@ -79,15 +77,15 @@ handle_info(Info, State) ->
     emqx_logger:error("[Pool] unexpected info: ~p", [Info]),
     {noreply, State}.
 
-terminate(_Reason, #state{pool = Pool, id = Id}) ->
+terminate(_Reason, #{pool := Pool, id := Id}) ->
     true = gproc_pool:disconnect_worker(Pool, {Pool, Id}).
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
-%%-----------------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 %% Internal functions
-%%-----------------------------------------------------------------------------
+%%------------------------------------------------------------------------------
 
 run({M, F, A}) ->
     erlang:apply(M, F, A);

src/emqx_pool_sup.erl

@@ -26,8 +26,12 @@ spec(Args) ->
 
 -spec(spec(any(), list()) -> supervisor:child_spec()).
 spec(ChildId, Args) ->
-    {ChildId, {?MODULE, start_link, Args},
+    #{id       => ChildId,
-     transient, infinity, supervisor, [?MODULE]}.
+      start    => {?MODULE, start_link, Args},
+      restart  => transient,
+      shutdown => infinity,
+      type     => supervisor,
+      modules  => [?MODULE]}.
 
 -spec(start_link(atom() | tuple(), atom(), mfa()) -> {ok, pid()} | {error, term()}).
 start_link(Pool, Type, MFA) ->

src/emqx_protocol.erl

@@ -33,36 +33,36 @@
 -export([shutdown/2]).
 
 -record(pstate, {
-          zone,
+         zone,
-          sendfun,
+         sendfun,
-          peername,
+         peername,
-          peercert,
+         peercert,
-          proto_ver,
+         proto_ver,
-          proto_name,
+         proto_name,
-          ackprops,
+         ackprops,
-          client_id,
+         client_id,
-          is_assigned,
+         is_assigned,
-          conn_pid,
+         conn_pid,
-          conn_props,
+         conn_props,
-          ack_props,
+         ack_props,
-          username,
+         username,
-          session,
+         session,
-          clean_start,
+         clean_start,
-          topic_aliases,
+         topic_aliases,
-          packet_size,
+         packet_size,
-          will_topic,
+         will_topic,
-          will_msg,
+         will_msg,
-          keepalive,
+         keepalive,
-          mountpoint,
+         mountpoint,
-          is_super,
+         is_super,
-          is_bridge,
+         is_bridge,
-          enable_ban,
+         enable_ban,
-          enable_acl,
+         enable_acl,
-          recv_stats,
+         recv_stats,
-          send_stats,
+         send_stats,
-          connected,
+         connected,
-          connected_at
+         connected_at
-         }).
+        }).
 
 -type(state() :: #pstate{}).
 -export_type([state/0]).
@@ -315,9 +315,12 @@ process_packet(Packet = ?PUBLISH_PACKET(?QOS_0, Topic, _PacketId, _Payload), PSt
     case check_publish(Packet, PState) of
         {ok, PState1} ->
             do_publish(Packet, PState1);
+        {error, ?RC_TOPIC_ALIAS_INVALID} ->
+            ?LOG(error, "Protocol error - ~p", [?RC_TOPIC_ALIAS_INVALID], PState),
+            {error, ?RC_TOPIC_ALIAS_INVALID, PState};
         {error, ReasonCode} ->
             ?LOG(warning, "Cannot publish qos0 message to ~s for ~s", [Topic, ReasonCode], PState),
-            {ok, PState}
+            {error, ReasonCode, PState}
     end;
 
 process_packet(Packet = ?PUBLISH_PACKET(?QOS_1, PacketId), PState) ->
@@ -644,9 +647,11 @@ check_publish(Packet, PState) ->
     run_check_steps([fun check_pub_caps/2,
                      fun check_pub_acl/2], Packet, PState).
 
-check_pub_caps(#mqtt_packet{header = #mqtt_packet_header{qos = QoS, retain = R}},
+check_pub_caps(#mqtt_packet{header = #mqtt_packet_header{qos = QoS, retain = Retain},
+                            variable = #mqtt_packet_publish{ properties = Properties}},
                #pstate{zone = Zone}) ->
-    emqx_mqtt_caps:check_pub(Zone, #{qos => QoS, retain => R}).
+    #{'Topic-Alias' := TopicAlias} = Properties,
+    emqx_mqtt_caps:check_pub(Zone, #{qos => QoS, retain => Retain, topic_alias => TopicAlias}).
 
 check_pub_acl(_Packet, #pstate{is_super = IsSuper, enable_acl = EnableAcl})
     when IsSuper orelse (not EnableAcl) ->

src/emqx_tracer.erl

@@ -78,7 +78,7 @@ init([]) ->
 handle_call({start_trace, Who, LogFile}, _From, State = #state{level = Level, traces = Traces}) ->
     case catch lager:trace_file(LogFile, [Who], Level, ?OPTIONS) of
         {ok, exists} ->
-            {reply, {error, already_existed}, State};
+            {reply, {error, already_exists}, State};
         {ok, Trace} ->
             {reply, ok, State#state{traces = maps:put(Who, {Trace, LogFile}, Traces)}};
         {error, Reason} ->

src/emqx_types.erl

@@ -17,7 +17,7 @@
 -include("emqx.hrl").
 
 -export_type([zone/0]).
--export_type([startlink_ret/0]).
+-export_type([startlink_ret/0, ok_or_error/1]).
 -export_type([pubsub/0, topic/0, subid/0, subopts/0]).
 -export_type([client_id/0, username/0, password/0, peername/0, protocol/0]).
 -export_type([credentials/0, session/0]).
@@ -29,6 +29,7 @@
 
 -type(zone() :: atom()).
 -type(startlink_ret() :: {ok, pid()} | ignore | {error, term()}).
+-type(ok_or_error(Reason) :: ok | {error, Reason}).
 -type(pubsub() :: publish | subscribe).
 -type(topic() :: binary()).
 -type(subid() :: binary() | atom()).

src/emqx_zone.erl

@@ -21,17 +21,20 @@
 -export([start_link/0]).
 -export([get_env/2, get_env/3]).
 -export([set_env/3]).
+-export([force_reload/0]).
+%% for test
+-export([stop/0]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
          code_change/3]).
 
--record(state, {timer}).
-
 -define(TAB, ?MODULE).
+-define(SERVER, ?MODULE).
 
+-spec(start_link() -> emqx_types:startlink_ret()).
 start_link() ->
-    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
 -spec(get_env(emqx_types:zone() | undefined, atom()) -> undefined | term()).
 get_env(undefined, Key) ->
@@ -50,7 +53,15 @@ get_env(Zone, Key, Def) ->
 
 -spec(set_env(emqx_types:zone(), atom(), term()) -> ok).
 set_env(Zone, Key, Val) ->
-    gen_server:cast(?MODULE, {set_env, Zone, Key, Val}).
+    gen_server:cast(?SERVER, {set_env, Zone, Key, Val}).
+
+-spec(force_reload() -> ok).
+force_reload() ->
+    gen_server:call(?SERVER, force_reload).
+
+-spec(stop() -> ok).
+stop() ->
+    gen_server:stop(?SERVER, normal, infinity).
 
 %%------------------------------------------------------------------------------
 %% gen_server callbacks
@@ -58,7 +69,11 @@ set_env(Zone, Key, Val) ->
 
 init([]) ->
     _ = emqx_tables:new(?TAB, [set, {read_concurrency, true}]),
-    {ok, element(2, handle_info(reload, #state{}))}.
+    {ok, element(2, handle_info(reload, #{timer => undefined}))}.
+
+handle_call(force_reload, _From, State) ->
+    _ = do_reload(),
+    {reply, ok, State};
 
 handle_call(Req, _From, State) ->
     emqx_logger:error("[Zone] unexpected call: ~p", [Req]),
@@ -73,11 +88,8 @@ handle_cast(Msg, State) ->
     {noreply, State}.
 
 handle_info(reload, State) ->
-    lists:foreach(
+    _ = do_reload(),
-      fun({Zone, Opts}) ->
+    {noreply, ensure_reload_timer(State#{timer := undefined}), hibernate};
-          [ets:insert(?TAB, {{Zone, Key}, Val}) || {Key, Val} <- Opts]
-      end, emqx_config:get_env(zones, [])),
-    {noreply, ensure_reload_timer(State), hibernate};
 
 handle_info(Info, State) ->
     emqx_logger:error("[Zone] unexpected info: ~p", [Info]),
@@ -93,6 +105,12 @@ code_change(_OldVsn, State, _Extra) ->
 %% Internal functions
 %%------------------------------------------------------------------------------
 
-ensure_reload_timer(State) ->
+do_reload() ->
-    State#state{timer = erlang:send_after(10000, self(), reload)}.
+    [ets:insert(?TAB, [{{Zone, Key}, Val} || {Key, Val} <- Opts])
+     || {Zone, Opts} <- emqx_config:get_env(zones, [])].
+
+ensure_reload_timer(State = #{timer := undefined}) ->
+    State#{timer := erlang:send_after(timer:minutes(5), self(), reload)};
+ensure_reload_timer(State) ->
+    State.
 

test/emqx_SUITE.erl

@@ -17,8 +17,6 @@
 -compile(export_all).
 -compile(nowarn_export_all).
 
--include("emqx_mqtt.hrl").
-
 -define(APP, emqx).
 
 -include_lib("eunit/include/eunit.hrl").
@@ -52,9 +50,7 @@
 -define(PUBPACKET, ?PUBLISH_PACKET(?PUBQOS, <<"sub/topic">>, ?PACKETID, <<"publish">>)).
 
 all() ->
-    [{group, connect}%,
+    [{group, connect}].
-    % {group, cleanSession}
-    ].
 
 groups() ->
     [{connect, [non_parallel_tests],
@@ -64,11 +60,7 @@ groups() ->
       mqtt_connect_with_ssl_oneway,
       mqtt_connect_with_ssl_twoway,
       mqtt_connect_with_ws
-      ]},
+      ]}].
-     {cleanSession, [sequence],
-      [cleanSession_validate]
-     }
-    ].
 
 init_per_suite(Config) ->
     emqx_ct_broker_helpers:run_setup_steps(),
@@ -109,15 +101,17 @@ mqtt_connect_with_ssl_oneway(_) ->
     emqx_ct_broker_helpers:change_opts(ssl_oneway),
     emqx:start(),
     ClientSsl = emqx_ct_broker_helpers:client_ssl(),
-    {ok, #ssl_socket{tcp = Sock, ssl = SslSock}}
+    {ok, #ssl_socket{tcp = _Sock1, ssl = SslSock} = Sock}
     = emqx_client_sock:connect("127.0.0.1", 8883, [{ssl_opts, ClientSsl}], 3000),
-%%     Packet = raw_send_serialise(?CLIENT),
+    Packet = raw_send_serialise(?CLIENT),
-%%     ssl:send(SslSock, Packet),
+    emqx_client_sock:setopts(Sock, [{active, once}]),
-%%     receive Data  ->
+    emqx_client_sock:send(Sock, Packet),
-%%         ct:log("Data:~p~n", [Data])
+    ?assert(
-%%     after 30000 ->
+    receive {ssl, _, ConAck}->
-%%               ok
+        {ok, ?CONNACK_PACKET(?CONNACK_ACCEPT), _} = raw_recv_pase(ConAck), true
-%%     end,
+    after 1000 ->
+        false
+    end),
     ssl:close(SslSock).
 
 mqtt_connect_with_ssl_twoway(_Config) ->
@@ -131,11 +125,12 @@ mqtt_connect_with_ssl_twoway(_Config) ->
     emqx_client_sock:setopts(Sock, [{active, once}]),
     emqx_client_sock:send(Sock, Packet),
     timer:sleep(500),
+    ?assert(
     receive {ssl, _, Data}->
-        {ok, ?CONNACK_PACKET(?CONNACK_ACCEPT), _} = raw_recv_pase(Data)
+        {ok, ?CONNACK_PACKET(?CONNACK_ACCEPT), _} = raw_recv_pase(Data), true
     after 1000 ->
-        ok
+        false
-    end,
+    end),
     emqx_client_sock:close(Sock).
 
 mqtt_connect_with_ws(_Config) ->
@@ -162,32 +157,6 @@ mqtt_connect_with_ws(_Config) ->
     {close, _} = rfc6455_client:close(WS),
     ok.
 
-cleanSession_validate(_) ->
-    {ok, C1} = emqttc:start_link([{host, "localhost"},
-                                         {port, 1883},
-                                         {client_id, <<"c1">>},
-                                         {clean_sess, false}]),
-    timer:sleep(10),
-    emqttc:subscribe(C1, <<"topic">>, qos0),
-    emqttc:disconnect(C1),
-    {ok, Pub} = emqttc:start_link([{host, "localhost"},
-                                         {port, 1883},
-                                         {client_id, <<"pub">>}]),
-
-    emqttc:publish(Pub, <<"topic">>, <<"m1">>, [{qos, 0}]),
-    timer:sleep(10),
-    {ok, C11} = emqttc:start_link([{host, "localhost"},
-                                   {port, 1883},
-                                   {client_id, <<"c1">>},
-                                   {clean_sess, false}]),
-    timer:sleep(100),
-    receive {publish, _Topic, M1} ->
-        ?assertEqual(<<"m1">>, M1)
-    after 1000 -> false
-    end,
-    emqttc:disconnect(Pub),
-    emqttc:disconnect(C11).
-
 raw_send_serialise(Packet) ->
     emqx_frame:serialize(Packet).
 

test/emqx_banned_SUITE.erl

@@ -28,14 +28,14 @@ all() -> [t_banned_all].
 t_banned_all(_) ->
     emqx_ct_broker_helpers:run_setup_steps(),
     emqx_banned:start_link(),
-    {MegaSecs, Secs, MicroSecs} = erlang:timestamp(),
+    TimeNow = erlang:system_time(second),
-    ok = emqx_banned:add(#banned{key = {client_id, <<"TestClient">>}, 
+    ok = emqx_banned:add(#banned{who = {client_id, <<"TestClient">>},
                                  reason = <<"test">>,
                                  by = <<"banned suite">>,
-                                 desc = <<"test">>, 
+                                 desc = <<"test">>,
-                                 until = {MegaSecs, Secs + 10, MicroSecs}}),
+                                 until = TimeNow + 10}),
     % here is not expire banned test because its check interval is greater than 5 mins, but its effect has been confirmed
     timer:sleep(100),
     ?assert(emqx_banned:check(#{client_id => <<"TestClient">>, username => undefined, peername => {undefined, undefined}})),
     emqx_banned:del({client_id, <<"TestClient">>}),
-    ?assertNot(emqx_banned:check(#{client_id => <<"TestClient">>, username => undefined, peername => {undefined, undefined}})).
+    ?assertNot(emqx_banned:check(#{client_id => <<"TestClient">>, username => undefined, peername => {undefined, undefined}})).

test/emqx_broker_SUITE.erl

@@ -20,7 +20,6 @@
 -define(APP, emqx).
 
 -include_lib("eunit/include/eunit.hrl").
-
 -include_lib("common_test/include/ct.hrl").
 
 -include("emqx.hrl").
@@ -32,7 +31,6 @@ all() ->
      {group, broker},
      {group, metrics},
      {group, stats},
-     {group, hook},
      {group, alarms}].
 
 groups() ->
@@ -43,10 +41,8 @@ groups() ->
                            t_shared_subscribe,
                            'pubsub#', 'pubsub+']},
      {session, [sequence], [start_session]},
-     {broker, [sequence], [hook_unhook]},
      {metrics, [sequence], [inc_dec_metric]},
      {stats, [sequence], [set_get_stat]},
-     {hook, [sequence], [add_delete_hook, run_hooks]},
      {alarms, [sequence], [set_alarms]}
     ].
 
@@ -165,8 +161,6 @@ start_session(_) ->
 %%--------------------------------------------------------------------
 %% Broker Group
 %%--------------------------------------------------------------------
-hook_unhook(_) ->
-    ok.
 
 %%--------------------------------------------------------------------
 %% Metric Group
@@ -178,61 +172,11 @@ inc_dec_metric(_) ->
 %%--------------------------------------------------------------------
 %% Stats Group
 %%--------------------------------------------------------------------
+
 set_get_stat(_) ->
     emqx_stats:setstat('retained/max', 99),
     99 = emqx_stats:getstat('retained/max').
 
-%%--------------------------------------------------------------------
-%% Hook Test
-%%--------------------------------------------------------------------
-
-add_delete_hook(_) ->
-    ok = emqx:hook(test_hook, fun ?MODULE:hook_fun1/1, []),
-    ok = emqx:hook(test_hook, {tag, fun ?MODULE:hook_fun2/1}, []),
-    {error, already_hooked} = emqx:hook(test_hook, {tag, fun ?MODULE:hook_fun2/1}, []),
-    Callbacks = [{callback, undefined, fun ?MODULE:hook_fun1/1, [], 0},
-                 {callback, tag, fun ?MODULE:hook_fun2/1, [], 0}],
-    Callbacks = emqx_hooks:lookup(test_hook),
-    ok = emqx:unhook(test_hook, fun ?MODULE:hook_fun1/1),
-    ct:print("Callbacks: ~p~n", [emqx_hooks:lookup(test_hook)]),
-    ok = emqx:unhook(test_hook, {tag, fun ?MODULE:hook_fun2/1}),
-    {error, not_found} = emqx:unhook(test_hook1, {tag, fun ?MODULE:hook_fun2/1}),
-    [] = emqx_hooks:lookup(test_hook),
-
-    ok = emqx:hook(emqx_hook, fun ?MODULE:hook_fun1/1, [], 9),
-    ok = emqx:hook(emqx_hook, {"tag", fun ?MODULE:hook_fun2/1}, [], 8),
-    Callbacks2 = [{callback, "tag", fun ?MODULE:hook_fun2/1, [], 8},
-                  {callback, undefined, fun ?MODULE:hook_fun1/1, [], 9}],
-    Callbacks2 = emqx_hooks:lookup(emqx_hook),
-    ok = emqx:unhook(emqx_hook, fun ?MODULE:hook_fun1/1),
-    ok = emqx:unhook(emqx_hook, {"tag", fun ?MODULE:hook_fun2/1}),
-    [] = emqx_hooks:lookup(emqx_hook).
-
-run_hooks(_) ->
-    ok = emqx:hook(foldl_hook, fun ?MODULE:hook_fun3/4, [init]),
-    ok = emqx:hook(foldl_hook, {tag, fun ?MODULE:hook_fun3/4}, [init]),
-    ok = emqx:hook(foldl_hook, fun ?MODULE:hook_fun4/4, [init]),
-    ok = emqx:hook(foldl_hook, fun ?MODULE:hook_fun5/4, [init]),
-    {stop, [r3, r2]} = emqx:run_hooks(foldl_hook, [arg1, arg2], []),
-    {ok, []} = emqx:run_hooks(unknown_hook, [], []),
-
-    ok = emqx:hook(foreach_hook, fun ?MODULE:hook_fun6/2, [initArg]),
-    ok = emqx:hook(foreach_hook, {tag, fun ?MODULE:hook_fun6/2}, [initArg]),
-    ok = emqx:hook(foreach_hook, fun ?MODULE:hook_fun7/2, [initArg]),
-    ok = emqx:hook(foreach_hook, fun ?MODULE:hook_fun8/2, [initArg]),
-    stop = emqx:run_hooks(foreach_hook, [arg]).
-
-hook_fun1([]) -> ok.
-hook_fun2([]) -> {ok, []}.
-
-hook_fun3(arg1, arg2, _Acc, init) -> ok.
-hook_fun4(arg1, arg2, Acc, init)  -> {ok, [r2 | Acc]}.
-hook_fun5(arg1, arg2, Acc, init)  -> {stop, [r3 | Acc]}.
-
-hook_fun6(arg, initArg) -> ok.
-hook_fun7(arg, initArg) -> any.
-hook_fun8(arg, initArg) -> stop.
-
 set_alarms(_) ->
     AlarmTest = #alarm{id = <<"1">>, severity = error, title="alarm title", summary="alarm summary"},
     emqx_alarm_mgr:set_alarm(AlarmTest),

test/emqx_ct_broker_helpers.erl

@@ -63,7 +63,7 @@ run_teardown_steps() ->
 
 generate_config() ->
     Schema = cuttlefish_schema:files([local_path(["priv", "emqx.schema"])]),
-    Conf = conf_parse:file([local_path(["etc", "emqx.conf"])]),
+    Conf = conf_parse:file([local_path(["etc", "gen.emqx.conf"])]),
     cuttlefish_generator:map(Schema, Conf).
 
 get_base_dir(Module) ->

test/emqx_hooks_SUITE.erl

@@ -0,0 +1,77 @@
+%% Copyright (c) 2018 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_hooks_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+
+all() ->
+    [add_delete_hook, run_hooks].
+
+add_delete_hook(_) ->
+    {ok, _} = emqx_hooks:start_link(),
+    ok = emqx:hook(test_hook, fun ?MODULE:hook_fun1/1, []),
+    ok = emqx:hook(test_hook, fun ?MODULE:hook_fun2/1, []),
+    ?assertEqual({error, already_exists},
+                 emqx:hook(test_hook, fun ?MODULE:hook_fun2/1, [])),
+    Callbacks = [{callback, {fun ?MODULE:hook_fun1/1, []}, undefined, 0},
+                 {callback, {fun ?MODULE:hook_fun2/1, []}, undefined, 0}],
+    ?assertEqual(Callbacks, emqx_hooks:lookup(test_hook)),
+    ok = emqx:unhook(test_hook, fun ?MODULE:hook_fun1/1),
+    ok = emqx:unhook(test_hook, fun ?MODULE:hook_fun2/1),
+    timer:sleep(1000),
+    ?assertEqual([], emqx_hooks:lookup(test_hook)),
+
+    ok = emqx:hook(emqx_hook, {?MODULE, hook_fun2, []}, 8),
+    ok = emqx:hook(emqx_hook, {?MODULE, hook_fun1, []}, 9),
+    Callbacks2 = [{callback, {?MODULE, hook_fun1, []}, undefined, 9},
+                  {callback, {?MODULE, hook_fun2, []}, undefined, 8}],
+    ?assertEqual(Callbacks2, emqx_hooks:lookup(emqx_hook)),
+    ok = emqx:unhook(emqx_hook, {?MODULE, hook_fun1, []}),
+    ok = emqx:unhook(emqx_hook, {?MODULE, hook_fun2, []}),
+    timer:sleep(1000),
+    ?assertEqual([], emqx_hooks:lookup(emqx_hook)),
+    ok = emqx_hooks:stop().
+
+run_hooks(_) ->
+    {ok, _} = emqx_hooks:start_link(),
+    ok = emqx:hook(foldl_hook, fun ?MODULE:hook_fun3/4, [init]),
+    ok = emqx:hook(foldl_hook, {?MODULE, hook_fun3, [init]}),
+    ok = emqx:hook(foldl_hook, fun ?MODULE:hook_fun4/4, [init]),
+    ok = emqx:hook(foldl_hook, fun ?MODULE:hook_fun5/4, [init]),
+    {stop, [r3, r2]} = emqx:run_hooks(foldl_hook, [arg1, arg2], []),
+    {ok, []} = emqx:run_hooks(unknown_hook, [], []),
+
+    ok = emqx:hook(foreach_hook, fun ?MODULE:hook_fun6/2, [initArg]),
+    {error, already_exists} = emqx:hook(foreach_hook, fun ?MODULE:hook_fun6/2, [initArg]),
+    ok = emqx:hook(foreach_hook, fun ?MODULE:hook_fun7/2, [initArg]),
+    ok = emqx:hook(foreach_hook, fun ?MODULE:hook_fun8/2, [initArg]),
+    stop = emqx:run_hooks(foreach_hook, [arg]),
+    ok = emqx_hooks:stop().
+
+hook_fun1([]) -> ok.
+hook_fun2([]) -> {ok, []}.
+
+hook_fun3(arg1, arg2, _Acc, init) -> ok.
+hook_fun4(arg1, arg2, Acc, init)  -> {ok, [r2 | Acc]}.
+hook_fun5(arg1, arg2, Acc, init)  -> {stop, [r3 | Acc]}.
+
+hook_fun6(arg, initArg) -> ok.
+hook_fun7(arg, initArg) -> any.
+hook_fun8(arg, initArg) -> stop.
+

test/emqx_listeners_SUITE.erl

@@ -49,7 +49,7 @@ restart_listeners(_) ->
     
 generate_config() ->
     Schema = cuttlefish_schema:files([local_path(["priv", "emqx.schema"])]),
-    Conf = conf_parse:file([local_path(["etc", "emqx.conf"])]),
+    Conf = conf_parse:file([local_path(["etc", "gen.emqx.conf"])]),
     cuttlefish_generator:map(Schema, Conf).
 
 set_app_env({App, Lists}) ->

test/emqx_mqtt_caps_SUITE.erl

@@ -38,7 +38,7 @@ t_get_set_caps(_) ->
         mqtt_wildcard_subscription => true
     },
     Caps2 = Caps#{max_packet_size => 1048576},
-    case emqx_mqtt_caps:get_caps(zone) of 
+    case emqx_mqtt_caps:get_caps(zone) of
         Caps -> ok;
         Caps2 -> ok
     end,
@@ -57,31 +57,41 @@ t_get_set_caps(_) ->
         mqtt_shared_subscription => true,
         mqtt_wildcard_subscription => true
     },
-    SubCaps = emqx_mqtt_caps:get_caps(zone, subscribe).
+    SubCaps = emqx_mqtt_caps:get_caps(zone, subscribe),
+    emqx_zone:stop().
 
 t_check_pub(_) ->
     {ok, _} = emqx_zone:start_link(),
     PubCaps = #{
         max_qos_allowed => ?QOS_1,
-        mqtt_retain_available => false
+        mqtt_retain_available => false,
+        max_topic_alias => 4
     },
     emqx_zone:set_env(zone, '$mqtt_pub_caps', PubCaps),
     timer:sleep(100),
+    ct:log("~p", [emqx_mqtt_caps:get_caps(zone, publish)]),
     BadPubProps1 = #{
         qos => ?QOS_2,
         retain => false
-    }, 
+    },
     {error, ?RC_QOS_NOT_SUPPORTED} = emqx_mqtt_caps:check_pub(zone, BadPubProps1),
     BadPubProps2 = #{
         qos => ?QOS_1,
         retain => true
-    }, 
+    },
     {error, ?RC_RETAIN_NOT_SUPPORTED} = emqx_mqtt_caps:check_pub(zone, BadPubProps2),
+    BadPubProps3 = #{
+        qos => ?QOS_1,
+        retain => false,
+        topic_alias => 5
+    },
+    {error, ?RC_TOPIC_ALIAS_INVALID} = emqx_mqtt_caps:check_pub(zone, BadPubProps3),
     PubProps = #{
         qos => ?QOS_1,
         retain => false
     },
-    ok = emqx_mqtt_caps:check_pub(zone, PubProps).
+    ok = emqx_mqtt_caps:check_pub(zone, PubProps),
+    emqx_zone:stop().
 
 t_check_sub(_) ->
     {ok, _} = emqx_zone:start_link(),
@@ -94,18 +104,19 @@ t_check_sub(_) ->
         mqtt_wildcard_subscription => true
     },
 
-    ok = do_check_sub([{<<"client/stat">>, Opts}], [{<<"client/stat">>, Opts}]), 
+    ok = do_check_sub([{<<"client/stat">>, Opts}], [{<<"client/stat">>, Opts}]),
     ok = do_check_sub(Caps#{max_qos_allowed => ?QOS_1}, [{<<"client/stat">>, Opts}], [{<<"client/stat">>, Opts#{qos => ?QOS_1}}]),
-    ok = do_check_sub(Caps#{max_topic_levels => 1}, 
+    ok = do_check_sub(Caps#{max_topic_levels => 1},
-                        [{<<"client/stat">>, Opts}], 
+                        [{<<"client/stat">>, Opts}],
                         [{<<"client/stat">>, Opts#{rc => ?RC_TOPIC_FILTER_INVALID}}]),
-    ok = do_check_sub(Caps#{mqtt_shared_subscription => false}, 
+    ok = do_check_sub(Caps#{mqtt_shared_subscription => false},
-                        [{<<"client/stat">>, Opts}], 
+                        [{<<"client/stat">>, Opts}],
                         [{<<"client/stat">>, Opts#{rc => ?RC_SHARED_SUBSCRIPTIONS_NOT_SUPPORTED}}]),
+
     ok = do_check_sub(Caps#{mqtt_wildcard_subscription => false}, 
                         [{<<"vlient/+/dsofi">>, Opts}], 
-                        [{<<"vlient/+/dsofi">>, Opts#{rc => ?RC_WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED}}]).
+                        [{<<"vlient/+/dsofi">>, Opts#{rc => ?RC_WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED}}]),
-        
+    emqx_zone:stop().
 
 
 

test/emqx_mqtt_packet_SUITE.erl

@@ -0,0 +1,117 @@
+%%%===================================================================
+%%% Copyright (c) 2013-2018 EMQ Inc. 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_mqtt_packet_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-import(emqx_frame, [serialize/1]).
+
+-include("emqx_mqtt.hrl").
+
+-include_lib("eunit/include/eunit.hrl").
+
+-define(INVALID_RESERVED, 1).
+
+-define(CONNECT_INVALID_PACKET(Var),
+    #mqtt_packet{header   = #mqtt_packet_header{type = ?INVALID_RESERVED},
+                 variable = Var}).
+
+-define(CASE1_PROTOCOL_NAME, ?CONNECT_PACKET(#mqtt_packet_connect{
+                                proto_name = <<"MQTC">>,
+                                client_id = <<"mqtt_protocol_name">>,
+                                username  = <<"admin">>,
+                                password  = <<"public">>})).
+
+-define(CASE2_PROTOCAL_VER, ?CONNECT_PACKET(#mqtt_packet_connect{
+                                client_id = <<"mqtt_client">>,
+                                proto_ver = 6,
+                                username  = <<"admin">>,
+                                password  = <<"public">>})).
+
+-define(CASE3_PROTOCAL_INVALID_RESERVED, ?CONNECT_INVALID_PACKET(#mqtt_packet_connect{
+                                client_id = <<"mqtt_client">>,
+                                proto_ver = 5,
+                                username  = <<"admin">>,
+                                password  = <<"public">>})).
+
+-define(PROTOCOL5, ?CONNECT_PACKET(#mqtt_packet_connect{
+                                proto_ver = 5,
+                                keepalive = 60,
+                                properties = #{'Message-Expiry-Interval' => 3600},
+                                client_id = <<"mqtt_client">>,
+                                will_topic = <<"will_tipic">>,
+                                will_payload = <<"will message">>,
+                                username  = <<"admin">>,
+                                password  = <<"public">>})).
+
+
+
+all() -> [{group, connect}].
+
+groups() -> [{connect, [sequence], 
+              [case1_protocol_name,
+               case2_protocol_ver%,
+             %TOTO case3_invalid_reserved
+              ]}].
+
+init_per_suite(Config) ->
+    emqx_ct_broker_helpers:run_setup_steps(),
+    Config.
+
+end_per_suite(_Config) ->
+    emqx_ct_broker_helpers:run_teardown_steps().
+
+init_per_group(_Group, Config) ->
+    Config.
+
+end_per_group(_Group, _Config) ->
+	ok.
+
+case1_protocol_name(_) ->
+    {ok, Sock} = emqx_client_sock:connect({127,0,0,1}, 1883, [binary, {packet, raw}, {active, false}], 3000),
+    MqttPacket = serialize(?CASE1_PROTOCOL_NAME),
+    emqx_client_sock:send(Sock, MqttPacket),
+    {ok, Data} = gen_tcp:recv(Sock, 0),
+    {ok, ?CONNACK_PACKET(?CONNACK_PROTO_VER), _} = raw_recv_pase(Data),
+    Disconnect  = gen_tcp:recv(Sock, 0),
+    ?assertEqual({error, closed}, Disconnect).
+ 
+case2_protocol_ver(_) ->
+    {ok, Sock} = emqx_client_sock:connect({127,0,0,1}, 1883, [binary, {packet, raw}, {active, false}], 3000),
+    Packet = serialize(?CASE2_PROTOCAL_VER),
+    emqx_client_sock:send(Sock, Packet),
+    {ok, Data} = gen_tcp:recv(Sock, 0),
+    %% case1 Unacceptable protocol version
+    {ok, ?CONNACK_PACKET(?CONNACK_PROTO_VER), _} = raw_recv_pase(Data),
+    Disconnect  = gen_tcp:recv(Sock, 0),
+    ?assertEqual({error, closed}, Disconnect).
+
+case3_invalid_reserved(_) ->
+    {ok, Sock} = emqx_client_sock:connect({127,0,0,1}, 1883, [binary, {packet, raw}, {active, false}], 3000),
+    Packet = serialize(?CASE3_PROTOCAL_INVALID_RESERVED),
+    emqx_client_sock:send(Sock, Packet),
+    {ok, Data} = gen_tcp:recv(Sock, 0),
+    %% case1 Unacceptable protocol version
+    ct:log("Data:~p~n", [raw_recv_pase(Data)]),
+    {ok, ?CONNACK_PACKET(?CONNACK_PROTO_VER), _} = raw_recv_pase(Data),
+    Disconnect  = gen_tcp:recv(Sock, 0),
+    ?assertEqual({error, closed}, Disconnect).
+
+raw_recv_pase(P) ->
+    emqx_frame:parse(P, {none, #{max_packet_size => ?MAX_PACKET_SIZE,
+                                 version         => ?MQTT_PROTO_V4} }).

test/emqx_mqtt_props_SUITE.erl

@@ -12,7 +12,7 @@
 %% See the License for the specific language governing permissions and
 %% limitations under the License.
 
--module(emqx_mqtt_properties_SUITE).
+-module(emqx_mqtt_props_SUITE).
 
 -compile(export_all).
 -compile(nowarn_export_all).
@@ -22,6 +22,7 @@
 all() -> [t_mqtt_properties_all].
 
 t_mqtt_properties_all(_) ->
-    Props = emqx_mqtt_properties:filter(?CONNECT, #{'Session-Expiry-Interval' => 1, 'Maximum-Packet-Size' => 255}),
+    Props = emqx_mqtt_props:filter(?CONNECT, #{'Session-Expiry-Interval' => 1, 'Maximum-Packet-Size' => 255}),
-    ok = emqx_mqtt_properties:validate(Props),
+    ok = emqx_mqtt_props:validate(Props),
-    #{} = emqx_mqtt_properties:filter(?CONNECT, #{'Maximum-QoS' => ?QOS_2}).
+    #{} = emqx_mqtt_props:filter(?CONNECT, #{'Maximum-QoS' => ?QOS_2}).
+

test/emqx_zone_SUITE.erl

@@ -18,15 +18,21 @@
 -compile(nowarn_export_all).
 
 -include("emqx_mqtt.hrl").
+-include_lib("eunit/include/eunit.hrl").
 
 all() -> [t_set_get_env].
 
 t_set_get_env(_) ->
-    emqx_zone:start_link(),
+    application:set_env(emqx, zones, [{china, [{language, chinese}]}]),
-    ok = emqx_zone:set_env(china, language, chinese),
+    {ok, _} = emqx_zone:start_link(),
-    timer:sleep(100),   % make sure set_env/3 is okay
+    ct:print("~p~n", [ets:tab2list(emqx_zone)]),
     chinese = emqx_zone:get_env(china, language),
     cn470 = emqx_zone:get_env(china, ism_band, cn470),
     undefined = emqx_zone:get_env(undefined, delay),
-    500 = emqx_zone:get_env(undefined, delay, 500).
+    500 = emqx_zone:get_env(undefined, delay, 500),
+    application:set_env(emqx, zones, [{zone1, [{key, val}]}]),
+    ?assertEqual(undefined, emqx_zone:get_env(zone1, key)),
+    emqx_zone:force_reload(),
+    ?assertEqual(val, emqx_zone:get_env(zone1, key)),
+    emqx_zone:stop().
 

vars

@@ -0,0 +1,9 @@
+%% vars here are for test only, not intended for release
+
+{platform_bin_dir,  "bin"}.
+{platform_data_dir, "data"}.
+{platform_etc_dir,  "etc"}.
+{platform_lib_dir,  "lib"}.
+{platform_log_dir,  "log"}.
+{platform_plugins_dir,  "plugins"}.
+