Merge the emqx-common, emqx-router libraries

etc/emqx.conf

@@ -578,7 +578,7 @@ mqtt.bridge.max_queue_len = 10000
 mqtt.bridge.ping_down_interval = 1s
 
 ##-------------------------------------------------------------------
-## MQTT Plugins
+## Plugins
 ##-------------------------------------------------------------------
 
 ## The etc dir for plugins' config.
@@ -595,7 +595,52 @@ mqtt.plugins.loaded_file = {{ platform_data_dir }}/loaded_plugins
 mqtt.plugins.expand_plugins_dir = {{ platform_plugins_dir }}/
 
 ##--------------------------------------------------------------------
-## MQTT Listeners
+## Modules
+##--------------------------------------------------------------------
+
+##--------------------------------------------------------------------
+## Presence Module
+
+## Enable Presence Module.
+##
+## Value: on | off
+module.presence = on
+
+## Sets the QoS for presence MQTT message.
+##
+## Value: 0 | 1 | 2
+module.presence.qos = 1
+
+##--------------------------------------------------------------------
+## Subscription Module
+
+## Enable Subscription Module.
+##
+## Value: on | off
+module.subscription = off
+
+## Subscribe the Topics automatically when client connected.
+## module.subscription.1.topic = $client/%c
+## Qos of the subscription: 0 | 1 | 2
+## module.subscription.1.qos = 1
+
+## module.subscription.2.topic = $user/%u
+## module.subscription.2.qos = 1
+
+##--------------------------------------------------------------------
+## Rewrite Module
+
+## Enable Rewrite Module.
+##
+## Value: on | off
+module.rewrite = off
+
+## {rewrite, Topic, Re, Dest}
+## module.rewrite.rule.1 = x/# ^x/y/(.+)$ z/y/$1
+## module.rewrite.rule.2 = y/+/z/# ^y/(.+)/z/(.+)$ y/z/$2
+
+##--------------------------------------------------------------------
+## Listeners
 ##--------------------------------------------------------------------
 
 ##--------------------------------------------------------------------

include/emqx.hrl

@@ -156,15 +156,12 @@
 -type(mqtt_delivery() :: #mqtt_delivery{}).
 
 %%--------------------------------------------------------------------
-%% MQTT Route
+%% Route
 %%--------------------------------------------------------------------
 
--record(mqtt_route,
+-record(route, { topic :: binary(), node :: node() }).
-        { topic :: binary(),
-          node  :: node()
-        }).
 
--type(mqtt_route() :: #mqtt_route{}).
+-type(route() :: #route{}).
 
 %%--------------------------------------------------------------------
 %% MQTT Alarm

include/emqx_common.hrl

@@ -0,0 +1,13 @@
+
+-define(record_to_map(Def, Rec),
+        maps:from_list(?record_to_proplist(Def, Rec))).
+
+-define(record_to_map(Def, Rec, Fields),
+        maps:from_list(?record_to_proplist(Def, Rec, Fields))).
+
+-define(record_to_proplist(Def, Rec),
+        lists:zip(record_info(fields, Def), tl(tuple_to_list(Rec)))).
+
+-define(record_to_proplist(Def, Rec, Fields),
+        [{K, V} || {K, V} <- ?record_to_proplist(Def, Rec), lists:member(K, Fields)]).
+

include/emqx_trie.hrl

@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -20,7 +20,7 @@
         { node_id        :: trie_node_id(),
           edge_count = 0 :: non_neg_integer(),
           topic          :: binary() | undefined,
-          flags          :: [retained | static]
+          flags          :: list(atom())
         }).
 
 -record(trie_edge,

priv/emqx.schema

@@ -762,7 +762,7 @@ end}.
 end}.
 
 %%-------------------------------------------------------------------
-%% MQTT Plugins
+%% Plugins
 %%-------------------------------------------------------------------
 
 {mapping, "mqtt.plugins.etc_dir", "emqx.plugins_etc_dir", [
@@ -778,7 +778,78 @@ end}.
 ]}.
 
 %%--------------------------------------------------------------------
-%% MQTT Listeners
+%% Modules
+%%--------------------------------------------------------------------
+
+{mapping, "module.presence", "emqx.modules", [
+  {default, off},
+  {datatype, flag}
+]}.
+
+{mapping, "module.presence.qos", "emqx.modules", [
+  {default, 1},
+  {datatype, integer},
+  {validators, ["range:0-2"]}
+]}.
+
+{mapping, "module.subscription", "emqx.modules", [
+  {default, off},
+  {datatype, flag}
+]}.
+
+{mapping, "module.subscription.$id.topic", "emqx.modules", [
+  {datatype, string}
+]}.
+
+{mapping, "module.subscription.$id.qos", "emqx.modules", [
+  {default, 1},
+  {datatype, integer},
+  {validators, ["range:0-2"]}
+]}.
+
+{mapping, "module.rewrite", "emqx.modules", [
+  {default, off},
+  {datatype, flag}
+]}.
+
+{mapping, "module.rewrite.rule.$id", "emqx.modules", [
+  {datatype, string}
+]}.
+
+{translation, "emqx.modules", fun(Conf) ->
+  Subscriptions = fun() ->
+      List = cuttlefish_variable:filter_by_prefix("module.subscription", Conf),
+      QosList = [Qos || {_, Qos} <- lists:sort([{I, Qos} || {[_,"subscription", I,"qos"], Qos} <- List])],
+      TopicList = [iolist_to_binary(Topic) || {_, Topic} <-
+        lists:sort([{I, Topic} || {[_,"subscription", I, "topic"], Topic} <- List])],
+      lists:zip(TopicList, QosList)
+  end,
+  Rewrites = fun() ->
+      Rules = cuttlefish_variable:filter_by_prefix("module.rewrite.rule", Conf),
+      lists:map(fun({[_, "rewrite", "rule", I], Rule}) ->
+                    [Topic, Re, Dest] = string:tokens(Rule, " "),
+                    {rewrite, list_to_binary(Topic), list_to_binary(Re), list_to_binary(Dest)}
+                end, Rules)
+  end,
+  lists:append([
+    case cuttlefish:conf_get("module.presence", Conf) of %% Presence
+        true  -> [{emqx_mod_presence, [{qos, cuttlefish:conf_get("module.presence.qos", Conf, 1)}]}];
+        false -> []
+    end,
+    case cuttlefish:conf_get("module.subscription", Conf) of %% Subscription
+        true  -> [{emqx_mod_subscription, Subscriptions()}];
+        false -> []
+    end,
+    case cuttlefish:conf_get("module.rewrite", Conf) of %% Rewrite
+        true  -> [{emqx_mod_rewrite, Rewrites()}];
+        false -> []
+    end
+  ])
+end}.
+
+
+%%--------------------------------------------------------------------
+%% Listeners
 %%--------------------------------------------------------------------
 
 %%--------------------------------------------------------------------

src/emqx_app.erl

@@ -38,6 +38,7 @@ start(_Type, _Args) ->
     ekka:start(),
     {ok, Sup} = emqx_sup:start_link(),
     ok = register_acl_mod(),
+    emqx_modules:load(),
     start_autocluster(),
     register(emqx, self()),
     print_vsn(),
@@ -45,6 +46,7 @@ start(_Type, _Args) ->
 
 -spec(stop(State :: term()) -> term()).
 stop(_State) ->
+    emqx_modules:unload(),
     catch emqx:stop_listeners().
 
 %%--------------------------------------------------------------------

src/emqx_base62.erl

@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -16,8 +16,6 @@
 
 -module(emqx_base62).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([encode/1, decode/1]).
 
 %% @doc Encode an integer to base62 string

src/emqx_boot.erl

@@ -16,8 +16,6 @@
 
 -module(emqx_boot).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([apply_module_attributes/1, all_module_attributes/1]).
 
 %% only {F, Args}...

src/emqx_gc.erl

@@ -18,8 +18,6 @@
 
 -module(emqx_gc).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([conn_max_gc_count/0, reset_conn_gc_count/2, maybe_force_gc/2,
          maybe_force_gc/3]).
 

src/emqx_keepalive.erl

@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -14,12 +14,8 @@
 %% limitations under the License.
 %%--------------------------------------------------------------------
 
-%% @doc Client Keepalive
-
 -module(emqx_keepalive).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([start/3, check/1, cancel/1]).
 
 -record(keepalive, {statfun, statval, tsec, tmsg, tref, repeat = 0}).

src/emqx_misc.erl

@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -16,8 +16,6 @@
 
 -module(emqx_misc).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([merge_opts/2, start_timer/2, start_timer/3, cancel_timer/1,
          proc_stats/0, proc_stats/1]).
 

src/emqx_mod_presence.erl

@@ -0,0 +1,73 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
+%%
+%% 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_mod_presence).
+
+-behaviour(emqx_gen_mod).
+
+-include("emqx.hrl").
+
+-export([load/1, unload/1]).
+
+-export([on_client_connected/3, on_client_disconnected/3]).
+
+load(Env) ->
+    emqx:hook('client.connected',    fun ?MODULE:on_client_connected/3, [Env]),
+    emqx:hook('client.disconnected', fun ?MODULE:on_client_disconnected/3, [Env]).
+
+on_client_connected(ConnAck, Client = #mqtt_client{client_id  = ClientId,
+                                                   username   = Username,
+                                                   peername   = {IpAddr, _},
+                                                   clean_sess = CleanSess,
+                                                   proto_ver  = ProtoVer}, Env) ->
+    Payload = mochijson2:encode([{clientid, ClientId},
+                                 {username, Username},
+                                 {ipaddress, iolist_to_binary(emqx_net:ntoa(IpAddr))},
+                                 {clean_sess, CleanSess},
+                                 {protocol, ProtoVer},
+                                 {connack, ConnAck},
+                                 {ts, emqx_time:now_secs()}]),
+    Msg = message(qos(Env), topic(connected, ClientId), Payload),
+    emqx:publish(emqx_message:set_flag(sys, Msg)),
+    {ok, Client}.
+
+on_client_disconnected(Reason, #mqtt_client{client_id = ClientId,
+                                            username = Username}, Env) ->
+    Payload = mochijson2:encode([{clientid, ClientId},
+                                 {username, Username},
+                                 {reason, reason(Reason)},
+                                 {ts, emqx_time:now_secs()}]),
+    Msg = message(qos(Env), topic(disconnected, ClientId), Payload),
+    emqx:publish(emqx_message:set_flag(sys, Msg)), ok.
+
+unload(_Env) ->
+    emqx:unhook('client.connected',    fun ?MODULE:on_client_connected/3),
+    emqx:unhook('client.disconnected', fun ?MODULE:on_client_disconnected/3).
+
+message(Qos, Topic, Payload) ->
+    emqx_message:make(presence, Qos, Topic, iolist_to_binary(Payload)).
+
+topic(connected, ClientId) ->
+    emqx_topic:systop(list_to_binary(["clients/", ClientId, "/connected"]));
+topic(disconnected, ClientId) ->
+    emqx_topic:systop(list_to_binary(["clients/", ClientId, "/disconnected"])).
+
+qos(Env) -> proplists:get_value(qos, Env, 0).
+
+reason(Reason) when is_atom(Reason) -> Reason;
+reason({Error, _}) when is_atom(Error) -> Error;
+reason(_) -> internal_error.
+

src/emqx_mod_rewrite.erl

@@ -0,0 +1,91 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
+%%
+%% 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_mod_rewrite).
+
+-include_lib("emqx.hrl").
+
+-export([load/1, unload/1]).
+
+-export([rewrite_subscribe/4, rewrite_unsubscribe/4, rewrite_publish/2]).
+
+%%--------------------------------------------------------------------
+%% API
+%%--------------------------------------------------------------------
+
+load(Rules0) ->
+    Rules = compile(Rules0),
+    emqx:hook('client.subscribe',  fun ?MODULE:rewrite_subscribe/4, [Rules]),
+    emqx:hook('client.unsubscribe',fun ?MODULE:rewrite_unsubscribe/4, [Rules]),
+    emqx:hook('message.publish',   fun ?MODULE:rewrite_publish/2, [Rules]).
+
+rewrite_subscribe(_ClientId, _Username, TopicTable, Rules) ->
+    lager:info("Rewrite subscribe: ~p", [TopicTable]),
+    {ok, [{match_rule(Topic, Rules), Opts} || {Topic, Opts} <- TopicTable]}.
+
+rewrite_unsubscribe(_ClientId, _Username, TopicTable, Rules) ->
+    lager:info("Rewrite unsubscribe: ~p", [TopicTable]),
+    {ok, [{match_rule(Topic, Rules), Opts} || {Topic, Opts} <- TopicTable]}.
+
+rewrite_publish(Message=#mqtt_message{topic = Topic}, Rules) ->
+    %%TODO: this will not work if the client is always online.
+    RewriteTopic =
+    case get({rewrite, Topic}) of
+        undefined ->
+            DestTopic = match_rule(Topic, Rules),
+            put({rewrite, Topic}, DestTopic), DestTopic;
+        DestTopic ->
+            DestTopic
+        end,
+    {ok, Message#mqtt_message{topic = RewriteTopic}}.
+
+unload(_) ->
+    emqx:unhook('client.subscribe',  fun ?MODULE:rewrite_subscribe/4),
+    emqx:unhook('client.unsubscribe',fun ?MODULE:rewrite_unsubscribe/4),
+    emqx:unhook('message.publish',   fun ?MODULE:rewrite_publish/2).
+
+%%--------------------------------------------------------------------
+%% Internal functions
+%%--------------------------------------------------------------------
+
+match_rule(Topic, []) ->
+    Topic;
+
+match_rule(Topic, [{rewrite, Filter, MP, Dest} | Rules]) ->
+    case emqx_topic:match(Topic, Filter) of
+        true  -> match_regx(Topic, MP, Dest);
+        false -> match_rule(Topic, Rules)
+    end.
+
+match_regx(Topic, MP, Dest) ->
+    case re:run(Topic, MP, [{capture, all_but_first, list}]) of
+        {match, Captured} ->
+            Vars = lists:zip(["\\$" ++ integer_to_list(I)
+                                || I <- lists:seq(1, length(Captured))], Captured),
+            iolist_to_binary(lists:foldl(
+                    fun({Var, Val}, Acc) ->
+                        re:replace(Acc, Var, Val, [global])
+                    end, Dest, Vars));
+        nomatch ->
+            Topic
+    end.
+
+compile(Rules) ->
+    lists:map(fun({rewrite, Topic, Re, Dest}) ->
+                  {ok, MP} = re:compile(Re),
+                  {rewrite, Topic, MP, Dest}
+              end, Rules).
+

src/emqx_mod_subscription.erl

@@ -0,0 +1,61 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
+%%
+%% 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_mod_subscription).
+
+-behaviour(emqx_gen_mod).
+
+-include_lib("emqx.hrl").
+
+-include_lib("emqx_mqtt.hrl").
+
+-export([load/1, on_client_connected/3, unload/1]).
+
+-define(TAB, ?MODULE).
+
+%%--------------------------------------------------------------------
+%% Load/Unload Hook
+%%--------------------------------------------------------------------
+
+load(Topics) ->
+    emqx:hook('client.connected', fun ?MODULE:on_client_connected/3, [Topics]).
+
+on_client_connected(?CONNACK_ACCEPT, Client = #mqtt_client{client_id  = ClientId,
+                                                           client_pid = ClientPid,
+                                                           username   = Username}, Topics) ->
+
+    Replace = fun(Topic) -> rep(<<"%u">>, Username, rep(<<"%c">>, ClientId, Topic)) end,
+    TopicTable = [{Replace(Topic), Qos} || {Topic, Qos} <- Topics],
+    ClientPid ! {subscribe, TopicTable},
+    {ok, Client};
+
+on_client_connected(_ConnAck, _Client, _State) ->
+    ok.
+
+unload(_) ->
+    emqx:unhook('client.connected', fun ?MODULE:on_client_connected/3).
+
+%%--------------------------------------------------------------------
+%% Internal Functions
+%%--------------------------------------------------------------------
+
+rep(<<"%c">>, ClientId, Topic) ->
+    emqx_topic:feed_var(<<"%c">>, ClientId, Topic);
+rep(<<"%u">>, undefined, Topic) ->
+    Topic;
+rep(<<"%u">>, Username, Topic) ->
+    emqx_topic:feed_var(<<"%u">>, Username, Topic).
+

src/emqx_modules.erl

@@ -0,0 +1,33 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
+%%
+%% 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_modules).
+
+-export([load/0, unload/0]).
+
+load() ->
+    lists:foreach(
+      fun({Mod, Env}) ->
+        ok = Mod:load(Env),
+        io:format("Load ~s module successfully.~n", [Mod])
+      end, emqx:env(modules, [])).
+
+unload() ->
+    lists:foreach(
+      fun({Mod, Env}) ->
+          Mod:unload(Env) end,
+      emqx:env(modules, [])).
+

src/emqx_net.erl

@@ -16,8 +16,6 @@
 
 -module(emqx_net).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include_lib("kernel/include/inet.hrl").
 
 -export([tcp_name/3, tcp_host/1, getopts/2, setopts/2, getaddr/2,

src/emqx_pmon.erl

@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -16,8 +16,6 @@
 
 -module(emqx_pmon).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([new/0, monitor/2, demonitor/2, erase/2]).
 
 -type(pmon() :: {?MODULE, map()}).

src/emqx_router.erl

@@ -28,11 +28,14 @@
 -boot_mnesia({mnesia, [boot]}).
 -copy_mnesia({mnesia, [copy]}).
 
--export([start_link/0, topics/0, local_topics/0]).
+-export([start_link/1]).
 
 %% For eunit tests
 -export([start/0, stop/0]).
 
+%% Topics
+-export([topics/0, local_topics/0]).
+
 %% Route APIs
 -export([add_route/1, get_routes/1, del_route/1, has_route/1]).
 
@@ -49,7 +52,7 @@
 
 -export([dump/0]).
 
--record(state, {stats_timer}).
+-record(state, {stats_fun, stats_timer}).
 
 -define(ROUTER, ?MODULE).
 
@@ -60,21 +63,21 @@
 %%--------------------------------------------------------------------
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(mqtt_route, [
+    ok = ekka_mnesia:create_table(route, [
                 {type, bag},
                 {ram_copies, [node()]},
-                {record_name, mqtt_route},
+                {record_name, route},
-                {attributes, record_info(fields, mqtt_route)}]);
+                {attributes, record_info(fields, route)}]);
 
 mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(mqtt_route, ram_copies).
+    ok = ekka_mnesia:copy_table(route, ram_copies).
 
 %%--------------------------------------------------------------------
 %% Start the Router
 %%--------------------------------------------------------------------
 
-start_link() ->
+start_link(StatsFun) ->
-    gen_server:start_link({local, ?ROUTER}, ?MODULE, [], []).
+    gen_server:start_link({local, ?ROUTER}, ?MODULE, [StatsFun], []).
 
 %%--------------------------------------------------------------------
 %% Topics
@@ -82,39 +85,39 @@ start_link() ->
 
 -spec(topics() -> list(binary())).
 topics() ->
-    mnesia:dirty_all_keys(mqtt_route).
+    mnesia:dirty_all_keys(route).
 
 -spec(local_topics() -> list(binary())).
 local_topics() ->
-    ets:select(mqtt_local_route, [{{'$1', '_'}, [], ['$1']}]).
+    ets:select(local_route, [{{'$1', '_'}, [], ['$1']}]).
 
 %%--------------------------------------------------------------------
 %% Match API
 %%--------------------------------------------------------------------
 
 %% @doc Match Routes.
--spec(match(Topic:: binary()) -> [mqtt_route()]).
+-spec(match(Topic:: binary()) -> [route()]).
 match(Topic) when is_binary(Topic) ->
     %% Optimize: ets???
     Matched = mnesia:ets(fun emqx_trie:match/1, [Topic]),
     %% Optimize: route table will be replicated to all nodes.
-    lists:append([ets:lookup(mqtt_route, To) || To <- [Topic | Matched]]).
+    lists:append([ets:lookup(route, To) || To <- [Topic | Matched]]).
 
 %% @doc Print Routes.
 -spec(print(Topic :: binary()) -> [ok]).
 print(Topic) ->
     [io:format("~s -> ~s~n", [To, Node]) ||
-        #mqtt_route{topic = To, node = Node} <- match(Topic)].
+        #route{topic = To, node = Node} <- match(Topic)].
 
 %%--------------------------------------------------------------------
 %% Route Management API
 %%--------------------------------------------------------------------
 
 %% @doc Add Route.
--spec(add_route(binary() | mqtt_route()) -> ok | {error, Reason :: term()}).
+-spec(add_route(binary() | route()) -> ok | {error, Reason :: term()}).
 add_route(Topic) when is_binary(Topic) ->
-    add_route(#mqtt_route{topic = Topic, node = node()});
+    add_route(#route{topic = Topic, node = node()});
-add_route(Route = #mqtt_route{topic = Topic}) ->
+add_route(Route = #route{topic = Topic}) ->
     case emqx_topic:wildcard(Topic) of
         true  -> case mnesia:is_transaction() of
                      true  -> add_trie_route(Route);
@@ -126,23 +129,23 @@ add_route(Route = #mqtt_route{topic = Topic}) ->
 add_direct_route(Route) ->
     mnesia:async_dirty(fun mnesia:write/1, [Route]).
 
-add_trie_route(Route = #mqtt_route{topic = Topic}) ->
+add_trie_route(Route = #route{topic = Topic}) ->
-    case mnesia:wread({mqtt_route, Topic}) of
+    case mnesia:wread({route, Topic}) of
         [] -> emqx_trie:insert(Topic);
         _  -> ok
     end,
     mnesia:write(Route).
 
 %% @doc Lookup Routes
--spec(get_routes(binary()) -> [mqtt_route()]).
+-spec(get_routes(binary()) -> [route()]).
 get_routes(Topic) ->
-    ets:lookup(mqtt_route, Topic).
+    ets:lookup(route, Topic).
 
 %% @doc Delete Route
--spec(del_route(binary() | mqtt_route()) -> ok | {error, Reason :: term()}).
+-spec(del_route(binary() | route()) -> ok | {error, Reason :: term()}).
 del_route(Topic) when is_binary(Topic) ->
-    del_route(#mqtt_route{topic = Topic, node = node()});
+    del_route(#route{topic = Topic, node = node()});
-del_route(Route = #mqtt_route{topic = Topic}) ->
+del_route(Route = #route{topic = Topic}) ->
     case emqx_topic:wildcard(Topic) of
         true  -> case mnesia:is_transaction() of
                      true  -> del_trie_route(Route);
@@ -154,8 +157,8 @@ del_route(Route = #mqtt_route{topic = Topic}) ->
 del_direct_route(Route) ->
     mnesia:async_dirty(fun mnesia:delete_object/1, [Route]).
 
-del_trie_route(Route = #mqtt_route{topic = Topic}) ->
+del_trie_route(Route = #route{topic = Topic}) ->
-    case mnesia:wread({mqtt_route, Topic}) of
+    case mnesia:wread({route, Topic}) of
         [Route] -> %% Remove route and trie
                    mnesia:delete_object(Route),
                    emqx_trie:delete(Topic);
@@ -167,7 +170,7 @@ del_trie_route(Route = #mqtt_route{topic = Topic}) ->
 %% @doc Has route?
 -spec(has_route(binary()) -> boolean()).
 has_route(Topic) when is_binary(Topic) ->
-    ets:member(mqtt_route, Topic).
+    ets:member(route, Topic).
 
 %% @private
 -spec(trans(function(), list(any())) -> ok | {error, term()}).
@@ -183,7 +186,7 @@ trans(Fun, Args) ->
 
 -spec(get_local_routes() -> list({binary(), node()})).
 get_local_routes() ->
-    ets:tab2list(mqtt_local_route).
+    ets:tab2list(local_route).
 
 -spec(add_local_route(binary()) -> ok).
 add_local_route(Topic) ->
@@ -195,15 +198,15 @@ del_local_route(Topic) ->
     
 -spec(match_local(binary()) -> [mqtt_route()]).
 match_local(Name) ->
-    case ets:info(mqtt_local_route, size) of
+    case ets:info(local_route, size) of
         0 -> [];
         _ -> ets:foldl(
                fun({Filter, Node}, Matched) ->
                    case emqx_topic:match(Name, Filter) of
-                       true  -> [#mqtt_route{topic = {local, Filter}, node = Node} | Matched];
+                       true  -> [#route{topic = {local, Filter}, node = Node} | Matched];
                        false -> Matched
                    end
-               end, [], mqtt_local_route)
+               end, [], local_route)
     end.
 
 -spec(clean_local_routes() -> ok).
@@ -211,7 +214,7 @@ clean_local_routes() ->
     gen_server:call(?ROUTER, clean_local_routes).
 
 dump() ->
-    [{route, ets:tab2list(mqtt_route)}, {local_route, ets:tab2list(mqtt_local_route)}].
+    [{route, ets:tab2list(route)}, {local_route, ets:tab2list(local_route)}].
 
 %% For unit test.
 start() ->
@@ -224,23 +227,23 @@ stop() ->
 %% gen_server Callbacks
 %%--------------------------------------------------------------------
 
-init([]) ->
+init([StatsFun]) ->
     ekka:monitor(membership),
-    ets:new(mqtt_local_route, [set, named_table, protected]),
+    ets:new(local_route, [set, named_table, protected]),
     {ok, TRef} = timer:send_interval(timer:seconds(1), stats),
-    {ok, #state{stats_timer = TRef}}.
+    {ok, #state{stats_fun = StatsFun, stats_timer = TRef}}.
 
 handle_call({add_local_route, Topic}, _From, State) ->
     %% why node()...?
-    ets:insert(mqtt_local_route, {Topic, node()}),
+    ets:insert(local_route, {Topic, node()}),
     {reply, ok, State};
 
 handle_call({del_local_route, Topic}, _From, State) ->
-    ets:delete(mqtt_local_route, Topic),
+    ets:delete(local_route, Topic),
     {reply, ok, State};
 
 handle_call(clean_local_routes, _From, State) ->
-    ets:delete_all_objects(mqtt_local_route),
+    ets:delete_all_objects(local_route),
     {reply, ok, State};
 
 handle_call(stop, _From, State) ->
@@ -253,19 +256,15 @@ handle_cast(_Msg, State) ->
     {noreply, State}.
 
 handle_info({membership, {mnesia, down, Node}}, State) ->
-    global:trans({?LOCK, self()},
+    global:trans({?LOCK, self()}, fun() -> clean_routes_(Node) end),
-                 fun() ->
+    handle_info(stats, State);
-                     clean_routes_(Node),
-                     update_stats_()
-                 end),
-    {noreply, State, hibernate};
 
 handle_info({membership, _Event}, State) ->
     %% ignore
     {noreply, State};
 
-handle_info(stats, State) ->
+handle_info(stats, State = #state{stats_fun = StatsFun}) ->
-    update_stats_(),
+    StatsFun(mnesia:table_info(route, size)),
     {noreply, State, hibernate};
 
 handle_info(_Info, State) ->
@@ -282,15 +281,12 @@ code_change(_OldVsn, State, _Extra) ->
 %% Internal Functions
 %%--------------------------------------------------------------------
 
-%% Clean Routes on Node
+%% Clean routes on the down node.
 clean_routes_(Node) ->
-    Pattern = #mqtt_route{_ = '_', node = Node},
+    Pattern = #route{_ = '_', node = Node},
     Clean = fun() ->
-                [mnesia:delete_object(mqtt_route, R, write) ||
+                [mnesia:delete_object(route, R, write) ||
-                    R <- mnesia:match_object(mqtt_route, Pattern, write)]
+                    R <- mnesia:match_object(route, Pattern, write)]
             end,
     mnesia:transaction(Clean).
 
-update_stats_() ->
-    emqx_stats:setstats('routes/count', 'routes/max', mnesia:table_info(mqtt_route, size)).
-

src/emqx_router_sup.erl

@@ -0,0 +1,38 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
+%%
+%% 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_router_sup).
+
+-behaviour(supervisor).
+
+-export([start_link/0]).
+
+-export([init/1]).
+
+start_link() ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+init([]) ->
+    StatsFun = emqx_stats:statsfun('routes/count', 'routes/max'),
+    SupFlags = #{strategy => one_for_all, intensity => 1, period => 5},
+    Router = #{id => emqx_router,
+               start => {emqx_router, start_link, [StatsFun]},
+               restart => permanent,
+               shutdown => 30000,
+               type => worker,
+               modules => [emqx_router]},
+    {ok, {SupFlags, [Router]}}.
+

src/emqx_time.erl

@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
 %%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
@@ -16,8 +16,6 @@
 
 -module(emqx_time).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([seed/0, now_secs/0, now_secs/1, now_ms/0, now_ms/1, ts_from_ms/1]).
 
 seed() ->

src/emqx_topic.erl

@@ -16,12 +16,8 @@
 
 -module(emqx_topic).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqx_mqtt.hrl").
 
--include("emqx_internal.hrl").
-
 -import(lists, [reverse/1]).
 
 -export([match/2, validate/1, triples/1, words/1, wildcard/1]).
@@ -206,8 +202,14 @@ parse(Topic, Options) ->
     {Topic, Options}.
 
 if_not_contain(Key, Options, Fun) when Key == local; Key == fastlane ->
-    ?IF(lists:member(Key, Options), error(invalid_topic), Fun());
+    case lists:member(Key, Options) of
+       true  -> error(invalid_topic);
+       false -> Fun()
+    end;
 
 if_not_contain(share, Options, Fun) ->
-    ?IF(lists:keyfind(share, 1, Options), error(invalid_topic), Fun()).
+    case lists:keyfind(share, 1, Options) of
+        true  -> error(invalid_topic);
+        false -> Fun()
+    end.
 

src/emqx_trie.erl

@@ -41,21 +41,21 @@
 -spec(mnesia(boot | copy) -> ok).
 mnesia(boot) ->
     %% Trie Table
-    ok = ekka_mnesia:create_table(mqtt_trie, [
+    ok = ekka_mnesia:create_table(trie, [
                 {ram_copies, [node()]},
                 {record_name, trie},
                 {attributes, record_info(fields, trie)}]),
     %% Trie Node Table
-    ok = ekka_mnesia:create_table(mqtt_trie_node, [
+    ok = ekka_mnesia:create_table(trie_node, [
                 {ram_copies, [node()]},
                 {record_name, trie_node},
                 {attributes, record_info(fields, trie_node)}]);
 
 mnesia(copy) ->
     %% Copy Trie Table
-    ok = ekka_mnesia:copy_table(mqtt_trie),
+    ok = ekka_mnesia:copy_table(trie),
     %% Copy Trie Node Table
-    ok = ekka_mnesia:copy_table(mqtt_trie_node).
+    ok = ekka_mnesia:copy_table(trie_node).
 
 %%--------------------------------------------------------------------
 %% Trie API
@@ -64,7 +64,7 @@ mnesia(copy) ->
 %% @doc Insert topic to trie
 -spec(insert(Topic :: binary()) -> ok).
 insert(Topic) when is_binary(Topic) ->
-    case mnesia:read(mqtt_trie_node, Topic) of
+    case mnesia:read(trie_node, Topic) of
         [#trie_node{topic = Topic}] ->
             ok;
         [TrieNode = #trie_node{topic = undefined}] ->
@@ -85,14 +85,14 @@ match(Topic) when is_binary(Topic) ->
 %% @doc Lookup a Trie Node
 -spec(lookup(NodeId :: binary()) -> [#trie_node{}]).
 lookup(NodeId) ->
-    mnesia:read(mqtt_trie_node, NodeId).
+    mnesia:read(trie_node, NodeId).
 
 %% @doc Delete topic from trie
 -spec(delete(Topic :: binary()) -> ok).
 delete(Topic) when is_binary(Topic) ->
-    case mnesia:read(mqtt_trie_node, Topic) of
+    case mnesia:read(trie_node, Topic) of
         [#trie_node{edge_count = 0}] ->
-            mnesia:delete({mqtt_trie_node, Topic}),
+            mnesia:delete({trie_node, Topic}),
             delete_path(lists:reverse(emqx_topic:triples(Topic)));
         [TrieNode] ->
             write_trie_node(TrieNode#trie_node{topic = undefined});
@@ -108,9 +108,9 @@ delete(Topic) when is_binary(Topic) ->
 %% @doc Add path to trie tree.
 add_path({Node, Word, Child}) ->
     Edge = #trie_edge{node_id = Node, word = Word},
-    case mnesia:read(mqtt_trie_node, Node) of
+    case mnesia:read(trie_node, Node) of
         [TrieNode = #trie_node{edge_count = Count}] ->
-            case mnesia:wread({mqtt_trie, Edge}) of
+            case mnesia:wread({trie, Edge}) of
                 [] ->
                     write_trie_node(TrieNode#trie_node{edge_count = Count+1}),
                     write_trie(#trie{edge = Edge, node_id = Child});
@@ -131,11 +131,11 @@ match_node(NodeId, Words) ->
     match_node(NodeId, Words, []).
 
 match_node(NodeId, [], ResAcc) ->
-    mnesia:read(mqtt_trie_node, NodeId) ++ 'match_#'(NodeId, ResAcc);
+    mnesia:read(trie_node, NodeId) ++ 'match_#'(NodeId, ResAcc);
 
 match_node(NodeId, [W|Words], ResAcc) ->
     lists:foldl(fun(WArg, Acc) ->
-        case mnesia:read(mqtt_trie, #trie_edge{node_id = NodeId, word = WArg}) of
+        case mnesia:read(trie, #trie_edge{node_id = NodeId, word = WArg}) of
             [#trie{node_id = ChildId}] -> match_node(ChildId, Words, Acc);
             [] -> Acc
         end
@@ -144,9 +144,9 @@ match_node(NodeId, [W|Words], ResAcc) ->
 %% @private
 %% @doc Match node with '#'.
 'match_#'(NodeId, ResAcc) ->
-    case mnesia:read(mqtt_trie, #trie_edge{node_id = NodeId, word = '#'}) of
+    case mnesia:read(trie, #trie_edge{node_id = NodeId, word = '#'}) of
         [#trie{node_id = ChildId}] ->
-            mnesia:read(mqtt_trie_node, ChildId) ++ ResAcc;
+            mnesia:read(trie_node, ChildId) ++ ResAcc;
         [] ->
             ResAcc
     end.
@@ -156,10 +156,10 @@ match_node(NodeId, [W|Words], ResAcc) ->
 delete_path([]) ->
     ok;
 delete_path([{NodeId, Word, _} | RestPath]) ->
-    mnesia:delete({mqtt_trie, #trie_edge{node_id = NodeId, word = Word}}),
+    mnesia:delete({trie, #trie_edge{node_id = NodeId, word = Word}}),
-    case mnesia:read(mqtt_trie_node, NodeId) of
+    case mnesia:read(trie_node, NodeId) of
         [#trie_node{edge_count = 1, topic = undefined}] ->
-            mnesia:delete({mqtt_trie_node, NodeId}),
+            mnesia:delete({trie_node, NodeId}),
             delete_path(RestPath);
         [TrieNode = #trie_node{edge_count = 1, topic = _}] ->
             write_trie_node(TrieNode#trie_node{edge_count = 0});
@@ -171,9 +171,9 @@ delete_path([{NodeId, Word, _} | RestPath]) ->
 
 %% @private
 write_trie(Trie) ->
-    mnesia:write(mqtt_trie, Trie, write).
+    mnesia:write(trie, Trie, write).
 
 %% @private
 write_trie_node(TrieNode) ->
-    mnesia:write(mqtt_trie_node, TrieNode, write).
+    mnesia:write(trie_node, TrieNode, write).
 

test/emqx_base62_SUITE.erl

@@ -0,0 +1,35 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
+%%
+%% 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_base62_SUITE).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-define(BASE62, emqx_base62).
+
+-compile(export_all).
+
+all() -> [t_base62_encode].
+
+t_base62_encode(_) ->
+    10 = ?BASE62:decode(?BASE62:encode(10)),
+    100 = ?BASE62:decode(?BASE62:encode(100)),
+    9999 = ?BASE62:decode(?BASE62:encode(9999)),
+    65535 = ?BASE62:decode(?BASE62:encode(65535)),
+    <<X:128/unsigned-big-integer>> = emqx_guid:gen(),
+    <<Y:128/unsigned-big-integer>> = emqx_guid:gen(),
+    X = ?BASE62:decode(?BASE62:encode(X)),
+    Y = ?BASE62:decode(?BASE62:encode(Y)).

test/emqx_guid_SUITE.erl

@@ -0,0 +1,41 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
+%%
+%% 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_guid_SUITE).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-compile(export_all).
+
+all() -> [t_guid_gen, t_guid_hexstr, t_guid_base62].
+
+t_guid_gen(_) ->
+    Guid1 = emqx_guid:gen(),
+    Guid2 = emqx_guid:gen(),
+    <<_:128>> = Guid1,
+    true = (Guid2 >= Guid1),
+    {Ts1, _, 0} = emqx_guid:new(),
+    Ts2 = emqx_guid:timestamp(emqx_guid:gen()),
+    true = Ts2 > Ts1.
+
+t_guid_hexstr(_) ->
+    Guid = emqx_guid:gen(),
+    ?assertEqual(Guid, emqx_guid:from_hexstr(emqx_guid:to_hexstr(Guid))).
+
+t_guid_base62(_) ->
+    Guid = emqx_guid:gen(),
+    ?assertEqual(Guid, emqx_guid:from_base62(emqx_guid:to_base62(Guid))).
+

test/emqx_inflight_SUITE.erl

@@ -16,8 +16,6 @@
 
 -module(emqx_inflight_SUITE).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include_lib("eunit/include/eunit.hrl").
 
 %% CT

test/emqx_keepalive_SUITE.erl

@@ -0,0 +1,43 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
+%%
+%% 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_keepalive_SUITE).
+
+-compile(export_all).
+
+all() -> [{group, keepalive}].
+
+groups() -> [{keepalive, [], [t_keepalive]}].
+
+%%--------------------------------------------------------------------
+%% Keepalive
+%%--------------------------------------------------------------------
+
+t_keepalive(_) ->
+    {ok, KA} = emqx_keepalive:start(fun() -> {ok, 1} end, 1, {keepalive, timeout}),
+    [resumed, timeout] = lists:reverse(keepalive_recv(KA, [])).
+
+keepalive_recv(KA, Acc) ->
+    receive
+        {keepalive, timeout} ->
+            case emqx_keepalive:check(KA) of
+                {ok, KA1} -> keepalive_recv(KA1, [resumed | Acc]);
+                {error, timeout} -> [timeout | Acc]
+            end
+        after 4000 ->
+                Acc
+    end.
+

test/emqx_misc_SUITE.erl

@@ -0,0 +1,46 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
+%%
+%% 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_misc_SUITE).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-compile(export_all).
+
+-define(SOCKOPTS, [binary,
+                   {packet,    raw},
+                   {reuseaddr, true},
+                   {backlog,   512},
+                   {nodelay,   true}]).
+
+all() -> [t_merge_opts].
+
+t_merge_opts(_) ->
+    Opts = emqx_misc:merge_opts(?SOCKOPTS, [raw,
+                                            binary,
+                                            {backlog, 1024},
+                                            {nodelay, false},
+                                            {max_clients, 1024},
+                                            {acceptors, 16}]),
+    ?assertEqual(1024, proplists:get_value(backlog, Opts)),
+    ?assertEqual(1024, proplists:get_value(max_clients, Opts)),
+    [binary, raw,
+     {acceptors, 16},
+     {backlog, 1024},
+     {max_clients, 1024},
+     {nodelay, false},
+     {packet, raw},
+     {reuseaddr, true}] = lists:sort(Opts).

test/emqx_pqueue_SUITE.erl

@@ -0,0 +1,69 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
+%%
+%% 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_pqueue_SUITE).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-compile(export_all).
+
+-define(PQ, emqx_pqueue).
+
+all() -> [t_priority_queue_plen, t_priority_queue_out2].
+
+t_priority_queue_plen(_) ->
+    Q = ?PQ:new(),
+    0 = ?PQ:plen(0, Q),
+    Q0 = ?PQ:in(z, Q),
+    1 = ?PQ:plen(0, Q0),
+    Q1 = ?PQ:in(x, 1, Q0),
+    1 = ?PQ:plen(1, Q1),
+    Q2 = ?PQ:in(y, 2, Q1),
+    1 = ?PQ:plen(2, Q2),
+    Q3 = ?PQ:in(z, 2, Q2),
+    2 = ?PQ:plen(2, Q3),
+    {_, Q4} = ?PQ:out(1, Q3),
+    0 = ?PQ:plen(1, Q4),
+    {_, Q5} = ?PQ:out(Q4),
+    1 = ?PQ:plen(2, Q5),
+    {_, Q6} = ?PQ:out(Q5),
+    0 = ?PQ:plen(2, Q6),
+    1 = ?PQ:len(Q6),
+    {_, Q7} = ?PQ:out(Q6),
+    0 = ?PQ:len(Q7).
+
+t_priority_queue_out2(_) ->
+    Els = [a, {b, 1}, {c, 1}, {d, 2}, {e, 2}, {f, 2}],
+    Q  = ?PQ:new(),
+    Q0 = lists:foldl(
+            fun({El, P}, Acc) ->
+                    ?PQ:in(El, P, Acc);
+                (El, Acc) ->
+                    ?PQ:in(El, Acc)
+            end, Q, Els),
+    {Val, Q1} = ?PQ:out(Q0),
+    {value, d} = Val,
+    {Val1, Q2} = ?PQ:out(2, Q1),
+    {value, e} = Val1,
+    {Val2, Q3} = ?PQ:out(1, Q2),
+    {value, b} = Val2,
+    {Val3, Q4} = ?PQ:out(Q3),
+    {value, f} = Val3,
+    {Val4, Q5} = ?PQ:out(Q4),
+    {value, c} = Val4,
+    {Val5, Q6} = ?PQ:out(Q5),
+    {value, a} = Val5,
+    {empty, _Q7} = ?PQ:out(Q6).

test/emqx_router_SUITE.erl

@@ -16,12 +16,12 @@
 
 -module(emqx_router_SUITE).
 
--compile(export_all).
-
 -include("emqx.hrl").
 
 -include_lib("eunit/include/eunit.hrl").
 
+-compile(export_all).
+
 -define(R, emqx_router).
 
 all() ->
@@ -35,7 +35,7 @@ groups() ->
        t_match_route,
        t_print,
        t_has_route,
-       router_unused]},
+       t_unused]},
      {local_route, [sequence],
       [t_get_local_topics,
        t_add_del_local_route,
@@ -44,7 +44,7 @@ groups() ->
 init_per_suite(Config) ->
     ekka:start(),
     ekka_mnesia:ensure_started(),
-    {ok, _R} = emqx_router:start(),
+    {ok, _} = emqx_router_sup:start_link(),
     Config.
 
 end_per_suite(_Config) ->
@@ -81,10 +81,10 @@ t_match_route(_) ->
     ?R:add_route(<<"a/+/c">>),
     ?R:add_route(<<"a/b/#">>),
     ?R:add_route(<<"#">>),
-    ?assertEqual([#mqtt_route{topic = <<"#">>, node = Node},
+    ?assertEqual([#route{topic = <<"#">>, node = Node},
-                  #mqtt_route{topic = <<"a/+/c">>, node = Node},
+                  #route{topic = <<"a/+/c">>, node = Node},
-                  #mqtt_route{topic = <<"a/b/#">>, node = Node},
+                  #route{topic = <<"a/b/#">>, node = Node},
-                  #mqtt_route{topic = <<"a/b/c">>, node = Node}],
+                  #route{topic = <<"a/b/c">>, node = Node}],
                  lists:sort(?R:match(<<"a/b/c">>))).
 
 t_has_route(_) ->
@@ -119,12 +119,12 @@ t_match_local_route(_) ->
     ?R:add_local_route(<<"a/+/c">>),
     ?R:add_local_route(<<"a/b/#">>),
     ?R:add_local_route(<<"#">>),
-    Matched = [Topic || #mqtt_route{topic = {local, Topic}} <- ?R:match_local(<<"a/b/c">>)],
+    Matched = [Topic || #route{topic = {local, Topic}} <- ?R:match_local(<<"a/b/c">>)],
     ?assertEqual([<<"#">>, <<"a/+/c">>, <<"a/b/#">>, <<"a/b/c">>], lists:sort(Matched)).
 
 clear_tables() ->
     ?R:clean_local_routes(),
-    lists:foreach(fun mnesia:clear_table/1, [mqtt_route, mqtt_trie, mqtt_trie_node]).
+    lists:foreach(fun mnesia:clear_table/1, [route, trie, trie_node]).
 
 router_add_del(_) ->
     %% Add
@@ -132,9 +132,9 @@ router_add_del(_) ->
     ?R:add_route(<<"a/b/c">>),
     ?R:add_route(<<"+/#">>),
     Routes = [R1, R2 | _] = [
-            #mqtt_route{topic = <<"#">>,     node = node()},
+            #route{topic = <<"#">>,     node = node()},
-            #mqtt_route{topic = <<"+/#">>,   node = node()},
+            #route{topic = <<"+/#">>,   node = node()},
-            #mqtt_route{topic = <<"a/b/c">>, node = node()}],
+            #route{topic = <<"a/b/c">>, node = node()}],
     Routes = lists:sort(?R:match(<<"a/b/c">>)),
 
     %% Batch Add
@@ -147,7 +147,7 @@ router_add_del(_) ->
     {atomic, []} = mnesia:transaction(fun emqx_trie:lookup/1, [<<"a/b/c">>]),
 
     %% Batch Del
-    R3 = #mqtt_route{topic = <<"#">>, node = 'a@127.0.0.1'},
+    R3 = #route{topic = <<"#">>, node = 'a@127.0.0.1'},
     ?R:add_route(R3),
     ?R:del_route(R1),
     ?R:del_route(R2),
@@ -155,16 +155,17 @@ router_add_del(_) ->
     [] = lists:sort(?R:match(<<"a/b/c">>)).
 
 t_print(_) ->
-    Routes = [#mqtt_route{topic = <<"a/b/c">>, node = node()},
+    Routes = [#route{topic = <<"a/b/c">>, node = node()},
-              #mqtt_route{topic = <<"#">>,     node = node()},
+              #route{topic = <<"#">>,     node = node()},
-              #mqtt_route{topic = <<"+/#">>,   node = node()}],
+              #route{topic = <<"+/#">>,   node = node()}],
     lists:foreach(fun(R) -> ?R:add_route(R) end, Routes),
     ?R:print(<<"a/b/c">>),
     ?R:del_route(<<"+/#">>),
     ?R:del_route(<<"a/b/c">>),
     ?R:del_route(<<"#">>).
 
-router_unused(_) ->
+t_unused(_) ->
-    gen_server:call(emqx_router, bad_call),
+    gen_server:call(?R, bad_call),
-    gen_server:cast(emqx_router, bad_msg),
+    gen_server:cast(?R, bad_msg),
-    emqx_router ! bad_info.
+    ?R ! bad_info.
+

test/emqx_time_SUITE.erl

@@ -0,0 +1,28 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2018 EMQ Enterprise, Inc.
+%%
+%% 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_time_SUITE).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-compile(export_all).
+
+all() -> [t_time_now_to].
+
+t_time_now_to(_) ->
+    emqx_time:seed(),
+    emqx_time:now_secs(),
+    emqx_time:now_ms().

test/emqx_trie_SUITE.erl

@@ -133,5 +133,5 @@ t_delete3(_) ->
             end).
 
 clear_tables() ->
-    lists:foreach(fun mnesia:clear_table/1, [mqtt_trie, mqtt_trie_node]).
+    lists:foreach(fun mnesia:clear_table/1, [trie, trie_node]).