0.14

TODO

@@ -11,3 +11,4 @@ TODO 5. dashboard
 
 TODO 6. emqttd_ctl
 
+TODO 7. transaction on route, and topic?

src/emqttd.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd main module.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd).
 

src/emqttd_access_control.erl

@@ -22,7 +22,6 @@
 %%% @doc Authentication and ACL Control Server
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_access_control).
 

src/emqttd_access_rule.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd ACL Rule
 %%% 
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_access_rule).
 

src/emqttd_acl_internal.erl

@@ -22,7 +22,6 @@
 %%% @doc Internal ACL that load rules from etc/acl.config
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_acl_internal).
 

src/emqttd_acl_mod.erl

@@ -22,7 +22,6 @@
 %%% @doc ACL module behaviour
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_acl_mod).
 

src/emqttd_alarm.erl

@@ -22,7 +22,6 @@
 %%% @doc Copy alarm_handler
 %%% 
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_alarm).
 

src/emqttd_app.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd application.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_app).
 

src/emqttd_auth_anonymous.erl

@@ -19,10 +19,9 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc Anonymous authentication module
+%%% @doc Anonymous Authentication Module
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_auth_anonymous).
 

src/emqttd_auth_clientid.erl

@@ -22,7 +22,6 @@
 %%% @doc ClientId Authentication Module
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_auth_clientid).
 

src/emqttd_auth_ldap.erl

@@ -22,7 +22,6 @@
 %%% @doc LDAP Authentication Module
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_auth_ldap).
 

src/emqttd_auth_mod.erl

@@ -19,10 +19,9 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc emqttd authentication behaviour
+%%% @doc emqttd Authentication Behaviour
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_auth_mod).
 

src/emqttd_auth_username.erl

@@ -22,7 +22,6 @@
 %%% @doc Authentication with username and password
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_auth_username).
 

src/emqttd_bridge.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd bridge
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_bridge).
 

src/emqttd_bridge_sup.erl

@@ -22,7 +22,6 @@
 %%% @doc Bridge Supervisor
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_bridge_sup).
 

src/emqttd_broker.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd broker
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_broker).
 

src/emqttd_cli.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd cli
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_cli).
 

src/emqttd_client.erl

@@ -22,7 +22,6 @@
 %%% @doc MQTT Client Connection
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_client).
 

src/emqttd_cm.erl

@@ -22,7 +22,6 @@
 %%% @doc MQTT Client Manager
 %%%  
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_cm).
 

src/emqttd_cm_sup.erl

@@ -22,7 +22,6 @@
 %%% @doc Client Manager Supervisor.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_cm_sup).
 

src/emqttd_ctl.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd control
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_ctl).
 

src/emqttd_dist.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd distribution functions
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_dist).
 

src/emqttd_gen_mod.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd gen_mod behaviour
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_gen_mod).
 

src/emqttd_guid.erl

@@ -36,7 +36,6 @@
 %%% @end
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_guid).
 

src/emqttd_http.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd http publish API and websocket client.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_http).
 

src/emqttd_message.erl

@@ -22,7 +22,6 @@
 %%% @doc MQTT Message Functions
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_message).
 

src/emqttd_metrics.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd metrics. responsible for collecting broker metrics
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_metrics).
 

src/emqttd_mnesia.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd mnesia
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mnesia).
 
@@ -121,7 +120,7 @@ copy_table(Table) ->
 %%------------------------------------------------------------------------------
 wait_for_tables() -> 
     %%TODO: is not right?
-    %%lager:info("local_tables: ~p", [mnesia:system_info(local_tables)]),
+    io:format("mnesia wait_for_tables: ~p~n", [mnesia:system_info(local_tables)]),
     mnesia:wait_for_tables(mnesia:system_info(local_tables), infinity).
 
 %%------------------------------------------------------------------------------

src/emqttd_mod_autosub.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd auto subscribe module.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mod_autosub).
 

src/emqttd_mod_presence.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd presence management module
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mod_presence).
 

src/emqttd_mod_rewrite.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd rewrite module
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mod_rewrite).
 

src/emqttd_mod_sup.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd module supervisor.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mod_sup).
 

src/emqttd_mqueue.erl

@@ -48,7 +48,6 @@
 %%% @end
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mqueue).
 

src/emqttd_net.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd net utility functions. some functions copied from rabbitmq.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_net).
 

src/emqttd_opts.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd options handler.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_opts).
 

src/emqttd_packet.erl

@@ -22,7 +22,6 @@
 %%% @doc MQTT Packet Functions
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_packet).
 

src/emqttd_parser.erl

@@ -22,7 +22,6 @@
 %%% @doc MQTT Packet Parser
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_parser).
 

src/emqttd_plugins.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd plugins.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_plugins).
 

src/emqttd_pool_sup.erl

@@ -22,7 +22,6 @@
 %%% @doc Common Pool Supervisor
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_pool_sup).
 

src/emqttd_pooler.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd pooler.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_pooler).
 

src/emqttd_protocol.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd protocol.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_protocol).
 

src/emqttd_pubsub.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd pubsub
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_pubsub).
 
@@ -66,31 +65,46 @@
 %%%=============================================================================
 %%% Mnesia callbacks
 %%%=============================================================================
-
 mnesia(boot) ->
-    %% Topic Table
-    ok = emqttd_mnesia:create_table(topic, [
-                {type, bag},
-                {ram_copies, [node()]},
-                {record_name, mqtt_topic},
-                {attributes, record_info(fields, mqtt_topic)}]),
-    RamOrDisc = case env(subscription) of
-        disc -> disc_copies;
-        _    -> ram_copies
-    end,
-    %% Subscription Table
-    ok = emqttd_mnesia:create_table(subscription, [
-                {type, bag},
-                {RamOrDisc, [node()]},
-                {record_name, mqtt_subscription},
-                {attributes, record_info(fields, mqtt_subscription)}]);
+    ok = create_table(topic, ram_copies),
+    case env(subscription) of
+        disc  -> ok = create_table(subscription, disc_copies);
+        ram   -> ok = create_table(subscription, ram_copies);
+        false -> ok
+    end;
 
 mnesia(copy) ->
     ok = emqttd_mnesia:copy_table(topic),
-    ok = emqttd_mnesia:copy_table(subscription).
+    case env(subscription) of
+        false -> ok;
+        _     -> ok = emqttd_mnesia:copy_table(subscription)
+    end.
+
+%% Topic Table
+create_table(topic, RamOrDisc) ->
+    emqttd_mnesia:create_table(topic, [
+            {type, bag},
+            {RamOrDisc, [node()]},
+            {record_name, mqtt_topic},
+            {attributes, record_info(fields, mqtt_topic)}]);
+
+%% Subscription Table
+create_table(subscription, RamOrDisc) ->
+    emqttd_mnesia:create_table(subscription, [
+            {type, bag},
+            {RamOrDisc, [node()]},
+            {record_name, mqtt_subscription},
+            {attributes, record_info(fields, mqtt_subscription)}]).
 
 env(Key) ->
-    proplists:get_value(Key, emqttd_broker:env(pubsub)).
+    case get({pubsub, Key}) of
+        undefined ->
+            Val = proplists:get_value(Key, emqttd_broker:env(pubsub)),
+            put({pubsub, Key}, Val),
+            Val;
+        Val ->
+            Val
+    end.
 
 %%%=============================================================================
 %%% API
@@ -309,7 +323,7 @@ remove_subscriptions(SubId, Topics) ->
 %%%=============================================================================
 
 trace(publish, From, _Msg) when is_atom(From) ->
-    %% Dont' trace broker publish
+    %% Dont' trace '$SYS' publish
     ignore;
 
 trace(publish, From, #mqtt_message{topic = Topic, payload = Payload}) ->

src/emqttd_pubsub_helper.erl

@@ -22,7 +22,6 @@
 %%% @doc PubSub Helper
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_pubsub_helper).
 

src/emqttd_pubsub_sup.erl

@@ -22,7 +22,6 @@
 %%% @doc PubSub Supervisor
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_pubsub_sup).
 
@@ -39,8 +38,7 @@
 -export([init/1]).
 
 start_link() ->
-    Opts = emqttd_broker:env(pubsub),
-    supervisor:start_link({local, ?MODULE}, ?MODULE, [Opts]).
+    supervisor:start_link({local, ?MODULE}, ?MODULE, [emqttd_broker:env(pubsub)]).
 
 init([Opts]) ->
     %% PubSub Helper

src/emqttd_retainer.erl

@@ -26,7 +26,6 @@
 %%% @end
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%% 
 %%%-----------------------------------------------------------------------------
 -module(emqttd_retainer).
 
@@ -66,7 +65,6 @@ mnesia(boot) ->
                 {ram_copies, [node()]},
                 {record_name, mqtt_retained},
                 {attributes, record_info(fields, mqtt_retained)}]);
-
 mnesia(copy) ->
     ok = emqttd_mnesia:copy_table(retained).
 

src/emqttd_router.erl

@@ -32,7 +32,6 @@
 %%% @end
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_router).
 
@@ -40,8 +39,12 @@
 
 -include("emqttd_protocol.hrl").
 
--export([init/1, lookup/1, route/2, add_routes/2,
-         delete_routes/1, delete_routes/2]).
+-export([init/1, route/2, lookup_routes/1,
+         add_routes/2, delete_routes/1, delete_routes/2]).
+
+-ifdef(TEST).
+-compile(export_all).
+-endif.
 
 %%------------------------------------------------------------------------------
 %% @doc Create route tables.
@@ -71,7 +74,7 @@ ensure_tab(Tab, Opts) ->
 %%------------------------------------------------------------------------------
 -spec add_routes(list({binary(), mqtt_qos()}), pid()) -> ok.
 add_routes(TopicTable, Pid) when is_pid(Pid) ->
-    case lookup(Pid) of
+    case lookup_routes(Pid) of
         [] ->
             erlang:monitor(process, Pid),
             insert_routes(TopicTable, Pid);
@@ -85,8 +88,8 @@ add_routes(TopicTable, Pid) when is_pid(Pid) ->
 %% @doc Lookup Routes
 %% @end
 %%------------------------------------------------------------------------------
--spec lookup(pid()) -> list({binary(), mqtt_qos()}).
-lookup(Pid) when is_pid(Pid) ->
+-spec lookup_routes(pid()) -> list({binary(), mqtt_qos()}).
+lookup_routes(Pid) when is_pid(Pid) ->
     [{Topic, Qos} || {_, Topic, Qos} <- ets:lookup(reverse_route, Pid)].
 
 %%------------------------------------------------------------------------------
@@ -100,7 +103,7 @@ delete_routes(Topics, Pid) ->
 
 -spec delete_routes(pid()) -> ok.
 delete_routes(Pid) when is_pid(Pid) ->
-    Routes = [{Topic, Pid} || {Topic, _Qos} <- lookup(Pid)],
+    Routes = [{Topic, Pid} || {Topic, _Qos} <- lookup_routes(Pid)],
     ets:delete(reverse_route, Pid),
     lists:foreach(fun delete_route_only/1, Routes).
 
@@ -112,7 +115,7 @@ delete_routes(Pid) when is_pid(Pid) ->
 route(Queue = <<"$Q/", _Q>>, Msg) ->
     case ets:lookup(route, Queue) of
         [] ->
-            setstats(dropped, true);
+            emqttd_metrics:inc('messages/dropped');
         Routes ->
             Idx = crypto:rand_uniform(1, length(Routes) + 1),
             {_, SubPid, SubQos} = lists:nth(Idx, Routes),
@@ -120,12 +123,15 @@ route(Queue = <<"$Q/", _Q>>, Msg) ->
     end;
 
 route(Topic, Msg) ->
-    Routes = ets:lookup(route, Topic),
-    setstats(dropped, Routes =:= []),
+    case ets:lookup(route, Topic) of
+        [] ->
+            emqttd_metrics:inc('messages/dropped');
+        Routes ->
             lists:foreach(
                 fun({_Topic, SubPid, SubQos}) ->
                     SubPid ! {dispatch, tune_qos(SubQos, Msg)}
-        end, Routes).
+                end, Routes)
+    end.
 
 tune_qos(SubQos, Msg = #mqtt_message{qos = PubQos}) when PubQos > SubQos ->
     Msg#mqtt_message{qos = SubQos};
@@ -185,8 +191,3 @@ delete_route({Topic, Pid}) ->
 delete_route_only({Topic, Pid}) ->
     ets:match_delete(route, {Topic, Pid, '_'}).
 
-setstats(dropped, false) ->
-    ignore;
-setstats(dropped, true) ->
-    emqttd_metrics:inc('messages/dropped').
-

src/emqttd_serialiser.erl

@@ -22,7 +22,6 @@
 %%% @doc MQTT Packet Serialiser
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_serialiser).
 

src/emqttd_session.erl

@@ -43,7 +43,6 @@
 %%% @end
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_session).
 

src/emqttd_session_sup.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd session supervisor.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_session_sup).
 

src/emqttd_sm.erl

@@ -22,7 +22,6 @@
 %%% @doc Session Manager
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_sm).
 
@@ -66,14 +65,14 @@
 %%%=============================================================================
 
 mnesia(boot) ->
-    %% Global session...
+    %% Global Session Table
     ok = emqttd_mnesia:create_table(session, [
                 {type, ordered_set},
                 {ram_copies, [node()]},
                 {record_name, mqtt_session},
                 {attributes, record_info(fields, mqtt_session)},
+                %% TODO: index_read is slow...
                 {index, [sess_pid]}]);
-
 mnesia(copy) ->
     ok = emqttd_mnesia:copy_table(session).
 

src/emqttd_sm_helper.erl

@@ -22,7 +22,6 @@
 %%% @doc Session Helper.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_sm_helper).
 

src/emqttd_sm_sup.erl

@@ -22,7 +22,6 @@
 %%% @doc Session Manager Supervisor.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_sm_sup).
 

src/emqttd_sup.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd top supervisor.
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_sup).
 

src/emqttd_sysmon.erl

@@ -22,7 +22,6 @@
 %%% @doc emqttd system monitor
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_sysmon).
 

src/emqttd_topic.erl

@@ -22,7 +22,6 @@
 %%% @doc MQTT Topic Functions
 %%%
 %%% @author Feng Lee <feng@emqtt.io>
-%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_topic).