0.14

include/emqttd.hrl

@@ -60,26 +60,15 @@
 -type mqtt_topic() :: #mqtt_topic{}.
 
 %%------------------------------------------------------------------------------
-%% MQTT Subscriber
+%% MQTT Subscription
 %%------------------------------------------------------------------------------
--record(mqtt_subscriber, {
+-record(mqtt_subscription, {
+    clientid :: binary() | atom(),
     topic    :: binary(),
-    subpid   :: pid(),
     qos = 0  :: 0 | 1 | 2
 }).
 
--type mqtt_subscriber() :: #mqtt_subscriber{}.
+-type mqtt_subscription() :: #mqtt_subscription{}.
-
-%%------------------------------------------------------------------------------
-%% P2P Queue Subscriber
-%%------------------------------------------------------------------------------
--record(mqtt_queue, {
-    name     :: binary(),
-    qpid     :: pid(),
-    qos = 0  :: 0 | 1 | 2
-}).
-
--type mqtt_queue() :: #mqtt_queue{}.
 
 %%------------------------------------------------------------------------------
 %% MQTT Client

include/emqttd_internal.hrl

@@ -19,12 +19,18 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Internal Header File
-%%% MQTT Internal Header.
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 
+-define(GPROC_POOL(JoinOrLeave, Pool, I),
+        (begin
+            case JoinOrLeave of
+                join  -> gproc_pool:connect_worker(Pool, {Pool, Id});
+                leave -> gproc_pool:disconnect_worker(Pool, {Pool, I})
+            end
+        end)).
+
 -define(record_to_proplist(Def, Rec),
         lists:zip(record_info(fields, Def),
                   tl(tuple_to_list(Rec)))).

rel/files/emqttd.config.development

@@ -141,11 +141,18 @@
             %% Max Payload Size of retained message
             {max_playload_size, 65536}
         ]},
-        %% PubSub
+
+        %% PubSub and Router
         {pubsub, [
-            %% default should be scheduler numbers
+            %% Default should be scheduler numbers
-            %% {pool_size, 8}
+            %% {pool_size, 8},
+
+            %% Route aging time(second)
+            {shard, true}, 
+
+            {aging, 10}
         ]},
+
         %% Bridge
         {bridge, [
             %%TODO: bridge queue size

rel/files/emqttd.config.production

@@ -138,6 +138,15 @@
             %% default should be scheduler numbers
             %% {pool_size, 8}
         ]},
+
+        %% Router
+        {router, [
+            %% Default should be scheduler numbers
+            %% {pool_size, 8},
+            %% Route aging time(second)
+            {aging, 5}
+        ]},
+
         %% Bridge
         {bridge, [
             %%TODO: bridge queue size

rel/files/vm.args

@@ -34,6 +34,9 @@
 ## Valid range is 1-2097151. Default is 1024.
 ## +zdbbl 8192
 
+## CPU Schedulers
+## +sbt db
+
 ##-------------------------------------------------------------------------
 ## Env
 ##-------------------------------------------------------------------------

src/emqttd.erl

@@ -19,19 +19,17 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd main module.
-%%% emqttd main module.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([start/0, env/1, env/2,
          open_listeners/1, close_listeners/1,
          load_all_mods/0, is_mod_enabled/1,
-         is_running/1]).
+         is_running/1, ensure_pool/3]).
 
 -define(MQTT_SOCKOPTS, [
 	binary,
@@ -130,3 +128,13 @@ is_running(Node) ->
         Pid when is_pid(Pid) -> true
     end.
 
+%%------------------------------------------------------------------------------
+%% @doc Ensure gproc pool exist.
+%% @end
+%%------------------------------------------------------------------------------
+ensure_pool(Pool, Type, Opts) ->
+    try gproc_pool:new(Pool, Type, Opts)
+    catch
+        error:exists -> ok
+    end.
+

src/emqttd_access_control.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Authentication and ACL Control Server
-%%% emqttd authentication and ACL server.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_access_control).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -behaviour(gen_server).

src/emqttd_access_rule.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd ACL Rule
-%%% emqttd ACL rule.
+%%% 
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_access_rule).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -type who() :: all | binary() |

src/emqttd_acl_internal.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Internal ACL that load rules from etc/acl.config
-%%% Internal ACL that load rules from etc/acl.config
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_acl_internal).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -export([all_rules/0]).

src/emqttd_acl_mod.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc ACL module behaviour
-%%% ACL module behaviour.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_acl_mod).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 %%%=============================================================================

src/emqttd_alarm.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Copy alarm_handler
-%%% copy alarm_handler.
+%%% 
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_alarm).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -behaviour(gen_event).

src/emqttd_app.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd application.
-%%% emqttd application.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_app).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd_cli.hrl").
 
 -behaviour(application).
@@ -72,11 +70,12 @@ start_listeners() ->
 start_servers(Sup) ->
     Servers = [{"emqttd ctl", emqttd_ctl},
                {"emqttd trace", emqttd_trace},
+               {"emqttd router", {supervisor, emqttd_router_sup}},
                {"emqttd pubsub", {supervisor, emqttd_pubsub_sup}},
                {"emqttd stats", emqttd_stats},
                {"emqttd metrics", emqttd_metrics},
-               {"emqttd retained", emqttd_retained},
+               {"emqttd retainer", emqttd_retainer},
-               {"emqttd pooler", {supervisor, emqttd_pooler_sup}},
+               {"emqttd pooler", {supervisor, emqttd_pooler}},
                {"emqttd client manager", {supervisor, emqttd_cm_sup}},
                {"emqttd session manager", {supervisor, emqttd_sm_sup}},
                {"emqttd session supervisor", {supervisor, emqttd_session_sup}},

src/emqttd_auth_anonymous.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Anonymous authentication module
-%%% Anonymous authentication module.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_auth_anonymous).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -behaviour(emqttd_auth_mod).
 
 -export([init/1, check/3, description/0]).

src/emqttd_auth_clientid.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc ClientId Authentication Module
-%%% ClientId Authentication Module.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_auth_clientid).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -export([add_clientid/1, add_clientid/2,

src/emqttd_auth_ldap.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc LDAP Authentication Module
-%%% LDAP Authentication Module
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_auth_ldap).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -import(proplists, [get_value/2, get_value/3]).

src/emqttd_auth_mod.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd authentication behaviour
-%%% emqttd authentication behaviour.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_auth_mod).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 %%%=============================================================================

src/emqttd_auth_username.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Authentication with username and password
-%%% Authentication with username and password.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_auth_username).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_cli.hrl").

src/emqttd_bridge.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd bridge
-%%% emqttd bridge.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_bridge).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").

src/emqttd_bridge_sup.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Bridge Supervisor
-%%% emqttd bridge supervisor.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_bridge_sup).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -behavior(supervisor).
 
 -export([start_link/0,

src/emqttd_broker.erl

@@ -19,14 +19,14 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd broker
-%%% emqttd broker.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_broker).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(gen_server).
 
 -include("emqttd.hrl").
 
@@ -48,18 +48,16 @@
 %% Tick API
 -export([start_tick/1, stop_tick/1]).
 
--behaviour(gen_server).
-
--define(SERVER, ?MODULE).
-
 %% gen_server Function Exports
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
--define(BROKER_TAB, mqtt_broker).
-
 -record(state, {started_at, sys_interval, heartbeat, tick_tref}).
 
+-define(SERVER, ?MODULE).
+
+-define(BROKER_TAB, mqtt_broker).
+
 %% $SYS Topics of Broker
 -define(SYSTOP_BROKERS, [
     version,      % Broker version

src/emqttd_cli.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd cli
-%%% emqttd cli.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_cli).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_cli.hrl").

src/emqttd_client.erl

@@ -19,14 +19,14 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc MQTT Client Connection
-%%% MQTT Client Connection.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_client).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(gen_server).
 
 -include("emqttd.hrl").
 
@@ -34,8 +34,6 @@
 
 -include("emqttd_internal.hrl").
 
--behaviour(gen_server).
-
 %% API Function Exports
 -export([start_link/2, session/1, info/1, kick/1]).
 

src/emqttd_cm.erl

@@ -19,19 +19,19 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc MQTT Client Manager
-%%% MQTT Client Manager
+%%%  
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_cm).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
+-include("emqttd_internal.hrl").
+
 %% API Exports 
--export([start_link/2, pool/0]).
+-export([start_link/3]).
 
 -export([lookup/1, lookup_proc/1, register/1, unregister/1]).
 
@@ -44,28 +44,27 @@
 %% gen_server2 priorities
 -export([prioritise_call/4, prioritise_cast/3, prioritise_info/3]).
 
--record(state, {id, statsfun, monitors}).
+-record(state, {pool, id, statsfun, monitors}).
 
--define(CM_POOL, ?MODULE).
+-define(POOL, ?MODULE).
 
 %%%=============================================================================
 %%% API
 %%%=============================================================================
 
 %%------------------------------------------------------------------------------
-%% @doc Start client manager
+%% @doc Start Client Manager
 %% @end
 %%------------------------------------------------------------------------------
--spec start_link(Id, StatsFun) -> {ok, pid()} | ignore | {error, any()} when
+-spec start_link(Pool, Id, StatsFun) -> {ok, pid()} | ignore | {error, any()} when
+        Pool :: atom(),
         Id   :: pos_integer(),
         StatsFun :: fun().
-start_link(Id, StatsFun) ->
+start_link(Pool, Id, StatsFun) ->
-    gen_server2:start_link(?MODULE, [Id, StatsFun], []).
+    gen_server2:start_link(?MODULE, [Pool, Id, StatsFun], []).
-
-pool() -> ?CM_POOL.
 
 %%------------------------------------------------------------------------------
-%% @doc Lookup client by clientId
+%% @doc Lookup Client by ClientId
 %% @end
 %%------------------------------------------------------------------------------
 -spec lookup(ClientId :: binary()) -> mqtt_client() | undefined.
@@ -81,19 +80,18 @@ lookup(ClientId) when is_binary(ClientId) ->
 %%------------------------------------------------------------------------------
 -spec lookup_proc(ClientId :: binary()) -> pid() | undefined.
 lookup_proc(ClientId) when is_binary(ClientId) ->
-    try ets:lookup_element(mqtt_client, ClientId, #mqtt_client.client_pid) of
+    try ets:lookup_element(mqtt_client, ClientId, #mqtt_client.client_pid)
-        Pid -> Pid
     catch
         error:badarg -> undefined
     end.
 
 %%------------------------------------------------------------------------------
-%% @doc Register clientId with pid.
+%% @doc Register ClientId with Pid.
 %% @end
 %%------------------------------------------------------------------------------
 -spec register(Client :: mqtt_client()) -> ok.
 register(Client = #mqtt_client{client_id = ClientId}) ->
-    CmPid = gproc_pool:pick_worker(?CM_POOL, ClientId),
+    CmPid = gproc_pool:pick_worker(?POOL, ClientId),
     gen_server2:cast(CmPid, {register, Client}).
 
 %%------------------------------------------------------------------------------
@@ -102,16 +100,18 @@ register(Client = #mqtt_client{client_id = ClientId}) ->
 %%------------------------------------------------------------------------------
 -spec unregister(ClientId :: binary()) -> ok.
 unregister(ClientId) when is_binary(ClientId) ->
-    CmPid = gproc_pool:pick_worker(?CM_POOL, ClientId),
+    CmPid = gproc_pool:pick_worker(?POOL, ClientId),
     gen_server2:cast(CmPid, {unregister, ClientId, self()}).
 
 %%%=============================================================================
 %%% gen_server callbacks
 %%%=============================================================================
 
-init([Id, StatsFun]) ->
+init([Pool, Id, StatsFun]) ->
-    gproc_pool:connect_worker(?CM_POOL, {?MODULE, Id}),
+    ?GPROC_POOL(join, Pool, Id),
-    {ok, #state{id = Id, statsfun = StatsFun, monitors = dict:new()}}.
+    {ok, #state{pool = Pool, id = Id,
+                statsfun = StatsFun,
+                monitors = dict:new()}}.
 
 prioritise_call(_Req, _From, _Len, _State) ->
     1.
@@ -172,9 +172,8 @@ handle_info(Info, State) ->
     lager:error("Unexpected Info: ~p", [Info]),
     {noreply, State}.
 
-terminate(_Reason, #state{id = Id}) ->
+terminate(_Reason, #state{pool = Pool, id = Id}) ->
-    gproc_pool:disconnect_worker(?CM_POOL, {?MODULE, Id}),
+    ?GPROC_POOL(leave, Pool, Id), ok.
-    ok.
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.

src/emqttd_cm_sup.erl

@@ -19,41 +19,46 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Client Manager Supervisor.
-%%% emqttd client manager supervisor.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_cm_sup).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(supervisor).
 
 -include("emqttd.hrl").
 
--behaviour(supervisor).
-
 %% API
 -export([start_link/0]).
 
 %% Supervisor callbacks
 -export([init/1]).
 
+-define(CM, emqttd_cm).
+
+-define(TAB, mqtt_client).
+
 start_link() ->
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
 init([]) ->
-    ets:new(mqtt_client, [ordered_set, named_table, public,
+    %% Create client table
-                          {keypos, 2}, {write_concurrency, true}]),
+    create_client_tab(),
-    Schedulers = erlang:system_info(schedulers),
-    gproc_pool:new(emqttd_cm:pool(), hash, [{size, Schedulers}]),
-    StatsFun = emqttd_stats:statsfun('clients/count', 'clients/max'),
-    Children = lists:map(
-                 fun(I) ->
-                    Name = {emqttd_cm, I},
-                    gproc_pool:add_worker(emqttd_cm:pool(), Name, I),
-                    {Name, {emqttd_cm, start_link, [I, StatsFun]},
-                                permanent, 10000, worker, [emqttd_cm]}
-                 end, lists:seq(1, Schedulers)),
-    {ok, {{one_for_all, 10, 100}, Children}}.
 
+    %% CM Pool Sup
+    MFA = {?CM, start_link, [emqttd_stats:statsfun('clients/count', 'clients/max')]},
+    PoolSup = emqttd_pool_sup:spec(pool_sup, [?CM, hash, erlang:system_info(schedulers), MFA]),
+
+    {ok, {{one_for_all, 10, 3600}, [PoolSup]}}.
+
+create_client_tab() ->
+    case ets:info(?TAB, name) of
+        undefined ->
+            ets:new(?TAB, [ordered_set, named_table, public,
+                           {keypos, 2}, {write_concurrency, true}]);
+        _ ->
+            ok
+    end.
 

src/emqttd_ctl.erl

@@ -19,21 +19,19 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd control
-%%% emqttd control.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_ctl).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(gen_server).
 
 -include("emqttd.hrl").
 
 -include("emqttd_cli.hrl").
 
--behaviour(gen_server).
-
 -define(SERVER, ?MODULE).
 
 %% API Function Exports

src/emqttd_dist.erl

@@ -19,10 +19,10 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd distribution functions
-%%% emqttd distribution functions.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_dist).
 

src/emqttd_gen_mod.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd gen_mod behaviour
-%%% emqttd gen_mod behaviour
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_gen_mod).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -ifdef(use_specs).

src/emqttd_guid.erl

@@ -34,11 +34,12 @@
 %%% 4. Sequence:  2 bytes sequence in one process
 %%%
 %%% @end
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
+%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_guid).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([gen/0, new/0, timestamp/1]).
 
 -define(MAX_SEQ, 16#FFFF).

src/emqttd_http.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd http publish API and websocket client.
-%%% emqttd http publish API and websocket client.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_http).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").

src/emqttd_keepalive.erl

@@ -21,12 +21,11 @@
 %%%-----------------------------------------------------------------------------
 %%% @doc client keepalive
 %%%
-%%% @end
+%%% @author Feng Lee <feng@emqtt.io>
+%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_keepalive).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([start/3, check/1, cancel/1]).
 
 -record(keepalive, {statfun, statval,

src/emqttd_message.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc MQTT Message Functions
-%%% MQTT Message Functions
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_message).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").

src/emqttd_metrics.erl

@@ -19,21 +19,19 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd metrics. responsible for collecting broker metrics
-%%% emqttd metrics. responsible for collecting broker metrics.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_metrics).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(gen_server).
 
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").
 
--behaviour(gen_server).
-
 -define(SERVER, ?MODULE).
 
 %% API Function Exports

src/emqttd_mnesia.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd mnesia
-%%% emqttd mnesia.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mnesia).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -export([start/0, cluster/1]).

src/emqttd_mod_autosub.erl

@@ -19,21 +19,19 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd auto subscribe module.
-%%% emqttd auto subscribe module.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mod_autosub).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(emqttd_gen_mod).
 
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").
 
--behaviour(emqttd_gen_mod).
-
 -export([load/1, client_connected/3, unload/1]).
 
 -record(state, {topics}).

src/emqttd_mod_presence.erl

@@ -19,19 +19,17 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd presence management module
-%%% emqttd presence management module.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mod_presence).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(emqttd_gen_mod).
 
 -include("emqttd.hrl").
 
--behaviour(emqttd_gen_mod).
-
 -export([load/1, unload/1]).
 
 -export([client_connected/3, client_disconnected/3]).

src/emqttd_mod_rewrite.erl

@@ -19,19 +19,17 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd rewrite module
-%%% emqttd rewrite module.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mod_rewrite).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(emqttd_gen_mod).
 
 -include("emqttd.hrl").
 
--behaviour(emqttd_gen_mod).
-
 -export([load/1, reload/1, unload/1]).
 
 -export([rewrite/3, rewrite/4]).

src/emqttd_mod_sup.erl

@@ -19,19 +19,17 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd module supervisor.
-%%% emqttd module supervisor.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mod_sup).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(supervisor).
 
 -include("emqttd.hrl").
 
--behaviour(supervisor).
-
 %% API
 -export([start_link/0, start_child/1, start_child/2]).
 

src/emqttd_mqueue.erl

@@ -46,11 +46,12 @@
 %%%    otherwise dropped the oldest pending one.
 %%%
 %%% @end
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
+%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_mqueue).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").

src/emqttd_net.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd net utility functions. some functions copied from rabbitmq.
-%%% emqttd net utility functions. some functions copied from rabbitmq.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_net).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include_lib("kernel/include/inet.hrl").
 
 -export([tcp_name/3, tcp_host/1, getopts/2, setopts/2,

src/emqttd_opts.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd options handler.
-%%% emqttd options handler.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_opts).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([merge/2, g/2, g/3]).
 
 %%------------------------------------------------------------------------------

src/emqttd_packet.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc MQTT Packet Functions
-%%% MQTT Packet Functions
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_packet).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").

src/emqttd_parser.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc MQTT Packet Parser
-%%% MQTT Packet Parser.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_parser).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").

src/emqttd_plugins.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd plugins.
-%%% emqttd plugins.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_plugins).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -export([load/0, unload/0]).

src/emqttd_pooler.erl

@@ -19,32 +19,42 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd pooler.
-%%% emqttd pooler.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_pooler).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -behaviour(gen_server).
 
+-include("emqttd_internal.hrl").
+
+%% Start the pool supervisor
+-export([start_link/0]).
+
 %% API Exports 
--export([start_link/1, submit/1, async_submit/1]).
+-export([start_link/2, submit/1, async_submit/1]).
 
 %% gen_server Function Exports
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
--record(state, {id}).
+-record(state, {pool, id}).
+
+%%------------------------------------------------------------------------------
+%% @doc Start Pooler Supervisor.
+%% @end
+%%------------------------------------------------------------------------------
+start_link() ->
+    emqttd_pool_sup:start_link(pooler, random, {?MODULE, start_link, []}).
 
 %%%=============================================================================
 %%% API
 %%%=============================================================================
--spec start_link(Id :: pos_integer()) -> {ok, pid()} | ignore | {error, any()}.
+-spec start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, any()}.
-start_link(Id) ->
+start_link(Pool, Id) ->
-    gen_server:start_link({local, name(Id)}, ?MODULE, [Id], []).
+    gen_server:start_link({local, name(Id)}, ?MODULE, [Pool, Id], []).
 
 name(Id) ->
     list_to_atom(lists:concat([?MODULE, "_", integer_to_list(Id)])).
@@ -54,22 +64,24 @@ name(Id) ->
 %% @end
 %%------------------------------------------------------------------------------
 submit(Fun) ->
-   gen_server:call(gproc_pool:pick_worker(pooler), {submit, Fun}, infinity).
+    Worker = gproc_pool:pick_worker(pooler),
+    gen_server:call(Worker, {submit, Fun}, infinity).
 
 %%------------------------------------------------------------------------------
 %% @doc Submit work to pooler asynchronously
 %% @end
 %%------------------------------------------------------------------------------
 async_submit(Fun) ->
-    gen_server:cast(gproc_pool:pick_worker(pooler), {async_submit, Fun}).
+    Worker = gproc_pool:pick_worker(pooler),
+    gen_server:cast(Worker, {async_submit, Fun}).
 
 %%%=============================================================================
 %%% gen_server callbacks
 %%%=============================================================================
 
-init([Id]) ->
+init([Pool, Id]) ->
-    gproc_pool:connect_worker(pooler, {pooler, Id}),
+    ?GPROC_POOL(join, Pool, Id),
-    {ok, #state{id = Id}}.
+    {ok, #state{pool = Pool, id = Id}}.
 
 handle_call({submit, Fun}, _From, State) ->
     {reply, run(Fun), State};
@@ -90,8 +102,8 @@ handle_cast(_Msg, State) ->
 handle_info(_Info, State) ->
     {noreply, State}.
 
-terminate(_Reason, #state{id = I}) ->
+terminate(_Reason, #state{pool = Pool, id = Id}) ->
-    gproc_pool:disconnect_worker(pooler, {pooler, I}), ok.
+    ?GPROC_POOL(leave, Pool, Id), ok.
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
@@ -105,4 +117,3 @@ run({M, F, A}) ->
 run(Fun) when is_function(Fun) ->
     Fun().
 
-

src/emqttd_pooler_sup.erl

@@ -1,58 +0,0 @@
-%%%-----------------------------------------------------------------------------
-%%% Copyright (c) 2012-2015 eMQTT.IO, All Rights Reserved.
-%%%
-%%% Permission is hereby granted, free of charge, to any person obtaining a copy
-%%% of this software and associated documentation files (the "Software"), to deal
-%%% in the Software without restriction, including without limitation the rights
-%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-%%% copies of the Software, and to permit persons to whom the Software is
-%%% furnished to do so, subject to the following conditions:
-%%%
-%%% The above copyright notice and this permission notice shall be included in all
-%%% copies or substantial portions of the Software.
-%%%
-%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-%%% SOFTWARE.
-%%%-----------------------------------------------------------------------------
-%%% @doc
-%%% emqttd pooler supervisor.
-%%%
-%%% @end
-%%%-----------------------------------------------------------------------------
--module(emqttd_pooler_sup).
-
--author("Feng Lee <feng@emqtt.io>").
-
--include("emqttd.hrl").
-
--behaviour(supervisor).
-
-%% API
--export([start_link/0, start_link/1]).
-
-%% Supervisor callbacks
--export([init/1]).
-
-start_link() ->
-    start_link(erlang:system_info(schedulers)).
-
-start_link(PoolSize) ->
-    supervisor:start_link({local, ?MODULE}, ?MODULE, [PoolSize]).
-
-init([PoolSize]) ->
-    gproc_pool:new(pooler, random, [{size, PoolSize}]),
-    Children = lists:map(
-                 fun(I) ->
-                         gproc_pool:add_worker(pooler, {pooler, I}, I),
-                         {{emqttd_pooler, I},
-                            {emqttd_pooler, start_link, [I]},
-                                permanent, 5000, worker, [emqttd_pooler]}
-                 end, lists:seq(1, PoolSize)),
-    {ok, {{one_for_all, 10, 100}, Children}}.
-
-

src/emqttd_protocol.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd protocol.
-%%% emqttd protocol.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_protocol).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").

src/emqttd_pubsub.erl

@@ -19,19 +19,21 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd pubsub
-%%% emqttd pubsub.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_pubsub).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(gen_server2).
 
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").
 
+-include("emqttd_internal.hrl").
+
 %% Mnesia Callbacks
 -export([mnesia/1]).
 
@@ -39,59 +41,42 @@
 -copy_mnesia({mnesia, [copy]}).
 
 %% API Exports 
--export([start_link/2]).
+-export([start_link/3]).
 
--export([create/1,
+-export([create/1, subscribe/1, subscribe/2, unsubscribe/1, publish/1]).
-         subscribe/1, subscribe/2,
-         unsubscribe/1,
-         publish/1]).
 
 %% Local node
--export([dispatch/2, match/1]).
+-export([match/1]).
-
--behaviour(gen_server2).
 
 %% gen_server Function Exports
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
-%% gen_server2 priorities
+-record(state, {pool, id}).
--export([prioritise_call/4, prioritise_cast/3, prioritise_info/3]).
 
--define(POOL, pubsub).
+-define(ROUTER, emqttd_router).
-
--record(state, {id, submap :: map()}).
 
 %%%=============================================================================
 %%% Mnesia callbacks
 %%%=============================================================================
 
 mnesia(boot) ->
-    %% p2p queue table
-    ok = emqttd_mnesia:create_table(queue, [
-                {type, set},
-                {ram_copies, [node()]},
-                {record_name, mqtt_queue},
-                {attributes, record_info(fields, mqtt_queue)}]),
     %% topic table
     ok = emqttd_mnesia:create_table(topic, [
                 {type, bag},
                 {ram_copies, [node()]},
                 {record_name, mqtt_topic},
                 {attributes, record_info(fields, mqtt_topic)}]),
-    %% local subscriber table, not shared with other nodes 
+    %% subscription table
-    ok = emqttd_mnesia:create_table(subscriber, [
+    ok = emqttd_mnesia:create_table(subscription, [
                 {type, bag},
                 {ram_copies, [node()]},
-                {record_name, mqtt_subscriber},
+                {record_name, mqtt_subscription},
-                {attributes, record_info(fields, mqtt_subscriber)},
+                {attributes, record_info(fields, mqtt_subscription)}]);
-                {index, [subpid]},
-                {local_content, true}]);
 
 mnesia(copy) ->
-    ok = emqttd_mnesia:copy_table(queue),
     ok = emqttd_mnesia:copy_table(topic),
-    ok = emqttd_mnesia:copy_table(subscriber).
+    ok = emqttd_mnesia:copy_table(subscription).
 
 %%%=============================================================================
 %%% API
@@ -101,11 +86,12 @@ mnesia(copy) ->
 %% @doc Start one pubsub server
 %% @end
 %%------------------------------------------------------------------------------
--spec start_link(Id, Opts) -> {ok, pid()} | ignore | {error, any()} when
+-spec start_link(Pool, Id, Opts) -> {ok, pid()} | ignore | {error, any()} when
+    Pool :: atom(),
     Id   :: pos_integer(),
-    Opts :: list().
+    Opts :: list(tuple()).
-start_link(Id, Opts) ->
+start_link(Pool, Id, Opts) ->
-    gen_server2:start_link({local, name(Id)}, ?MODULE, [Id, Opts], []).
+    gen_server2:start_link({local, name(Id)}, ?MODULE, [Pool, Id, Opts], []).
 
 name(Id) ->
     list_to_atom("emqttd_pubsub_" ++ integer_to_list(Id)).
@@ -115,38 +101,38 @@ name(Id) ->
 %% @end
 %%------------------------------------------------------------------------------
 -spec create(Topic :: binary()) -> ok | {error, Error :: any()}.
-create(<<"$Q/", _Queue/binary>>) ->
-    %% protecte from queue
-    {error, cannot_create_queue};
-
 create(Topic) when is_binary(Topic) ->
-    TopicR = #mqtt_topic{topic = Topic, node = node()},
+    case mnesia:transaction(fun add_topic/1, [#mqtt_topic{topic = Topic, node = node()}]) of
-    case mnesia:transaction(fun add_topic/1, [TopicR]) of
+        {atomic, ok}     -> setstats(topics), ok;
-        {atomic, ok} ->
+        {aborted, Error} -> {error, Error}
-            setstats(topics), ok;
-        {aborted, Error} ->
-            {error, Error}
     end.
 
 %%------------------------------------------------------------------------------
-%% @doc Subscribe topic
+%% @doc Subscribe Topic
 %% @end
 %%------------------------------------------------------------------------------
+-spec subscribe(Topic, Qos) -> {ok, Qos} when
+     Topic :: binary(),
+     Qos   :: mqtt_qos() | mqtt_qos_name().
+subscribe(Topic, Qos) ->
+    %%TODO:...
+    subscribe([{Topic, Qos}]).
+
 -spec subscribe({Topic, Qos} | list({Topic, Qos})) ->
     {ok, Qos | list(Qos)} | {error, any()} when
     Topic   :: binary(),
     Qos     :: mqtt_qos() | mqtt_qos_name().
 subscribe({Topic, Qos}) when is_binary(Topic) andalso (?IS_QOS(Qos) orelse is_atom(Qos)) ->
-    call({subscribe, self(), Topic, ?QOS_I(Qos)});
+    %%TODO:...
+    subscribe([{Topic, Qos}]);
 
-subscribe(Topics = [{_Topic, _Qos} | _]) ->
+subscribe(TopicTable0 = [{_Topic, _Qos} | _]) ->
-    call({subscribe, self(), [{Topic, ?QOS_I(Qos)} || {Topic, Qos} <- Topics]}).
+    Self = self(),
-
+    TopicTable = [{Topic, ?QOS_I(Qos)} || {Topic, Qos} <- TopicTable0],
--spec subscribe(Topic, Qos) -> {ok, Qos} when
+    ?ROUTER:add_routes(TopicTable, Self),
-     Topic :: binary(),
+    PubSub = gproc_pool:pick_worker(pubsub, Self),
-     Qos   :: mqtt_qos() | mqtt_qos_name().
+    SubReq = {subscribe, Self, TopicTable},
-subscribe(Topic, Qos) ->
+    gen_server2:call(PubSub, SubReq, infinity).
-    subscribe({Topic, Qos}).
 
 %%------------------------------------------------------------------------------
 %% @doc Unsubscribe Topic or Topics
@@ -154,18 +140,13 @@ subscribe(Topic, Qos) ->
 %%------------------------------------------------------------------------------
 -spec unsubscribe(binary() | list(binary())) -> ok.
 unsubscribe(Topic) when is_binary(Topic) ->
-    cast({unsubscribe, self(), Topic});
+    unsubscribe([Topic]);
 
 unsubscribe(Topics = [Topic|_]) when is_binary(Topic) ->
-    cast({unsubscribe, self(), Topics}).
+    Self = self(),
-
+    ?ROUTER:delete_routes(Topics, Self),
-call(Req) ->
+    PubSub = gproc_pool:pick_worker(pubsub, Self),
-    Pid = gproc_pool:pick_worker(?POOL, self()),
+    gen_server2:cast(PubSub, {unsubscribe, Self, Topics}).
-    gen_server2:call(Pid, Req, infinity).
-
-cast(Msg) ->
-    Pid = gproc_pool:pick_worker(?POOL, self()),
-    gen_server2:cast(Pid, Msg).
 
 %%------------------------------------------------------------------------------
 %% @doc Publish to cluster nodes
@@ -178,7 +159,7 @@ publish(Msg = #mqtt_message{from = From}) ->
                = emqttd_broker:foldl_hooks('message.publish', [], Msg),
 
     %% Retain message first. Don't create retained topic.
-    case emqttd_retained:retain(Msg1) of
+    case emqttd_retainer:retain(Msg1) of
         ok ->
             %% TODO: why unset 'retain' flag?
             publish(Topic, emqttd_message:unset_flag(Msg1));
@@ -186,42 +167,12 @@ publish(Msg = #mqtt_message{from = From}) ->
             publish(Topic, Msg1)
      end.
 
-publish(Queue = <<"$Q/", _/binary>>, Msg = #mqtt_message{qos = Qos}) ->
-    lists:foreach(
-        fun(#mqtt_queue{qpid = QPid, qos = SubQos}) -> 
-            Msg1 = if
-                Qos > SubQos -> Msg#mqtt_message{qos = SubQos};
-                true -> Msg
-            end,
-            QPid ! {dispatch, Msg1}
-        end, mnesia:dirty_read(queue, Queue));
-    
 publish(Topic, Msg) when is_binary(Topic) ->
 	lists:foreach(fun(#mqtt_topic{topic=Name, node=Node}) ->
-        case Node =:= node() of
+                rpc:cast(Node, ?ROUTER, route, [Name, Msg])
-            true -> dispatch(Name, Msg);
-            false -> rpc:cast(Node, ?MODULE, dispatch, [Name, Msg])
-        end
         end, match(Topic)).
 
-%%------------------------------------------------------------------------------
+%%TODO: Benchmark and refactor...
-%% @doc Dispatch message locally. should only be called by publish.
-%% @end
-%%------------------------------------------------------------------------------
--spec dispatch(Topic :: binary(), Msg :: mqtt_message()) -> non_neg_integer().
-dispatch(Topic, Msg = #mqtt_message{qos = Qos}) when is_binary(Topic) ->
-    Subscribers = mnesia:dirty_read(subscriber, Topic),
-    setstats(dropped, Subscribers =:= []),
-    lists:foreach(
-        fun(#mqtt_subscriber{subpid=SubPid, qos = SubQos}) ->
-                Msg1 = if
-                    Qos > SubQos -> Msg#mqtt_message{qos = SubQos};
-                    true -> Msg
-                end,
-                SubPid ! {dispatch, Msg1}
-            end, Subscribers), 
-    length(Subscribers).
-
 -spec match(Topic :: binary()) -> [mqtt_topic()].
 match(Topic) when is_binary(Topic) ->
 	MatchedTopics = mnesia:async_dirty(fun emqttd_trie:match/1, [Topic]),
@@ -231,75 +182,16 @@ match(Topic) when is_binary(Topic) ->
 %%% gen_server callbacks
 %%%=============================================================================
 
-init([Id, _Opts]) ->
+init([Pool, Id, _Opts]) ->
-    %%process_flag(priority, high),
+    ?GPROC_POOL(join, Pool, Id),
-    %%process_flag(min_heap_size, 1024*1024),
+    {ok, #state{pool = Pool, id = Id}}.
-    gproc_pool:connect_worker(pubsub, {?MODULE, Id}),
-    {ok, #state{id = Id, submap = maps:new()}}.
 
-prioritise_call(Msg, _From, _Len, _State) ->
+%%TODO: clientId???
-    case Msg of
+handle_call({subscribe, _SubPid, TopicTable}, _From, State) ->
-        {subscriber, _, _} -> 1;
+    Records = [#mqtt_topic{topic = Topic, node = node()} || {Topic, _Qos} <- TopicTable],
-        _                  -> 0
+    case mnesia:transaction(fun() -> [add_topic(Record) || Record <- Records] end) of
-    end.
-
-prioritise_cast(Msg, _Len, _State) ->
-    case Msg of
-        {unsubscribe, _, _} -> 2;
-        _                   -> 0
-    end.
-
-prioritise_info(Msg, _Len, _State) ->
-    case Msg of
-        {'DOWN', _, _, _, _} -> 3;
-        _                    -> 0
-    end.
-
-handle_call({subscribe, SubPid, Topics}, _From, State) ->
-    TopicSubs = lists:map(fun({<<"$Q/", _/binary>> = Queue, Qos}) ->
-                            #mqtt_queue{name = Queue, qpid = SubPid, qos = Qos};
-                             ({Topic, Qos}) ->
-                            {#mqtt_topic{topic = Topic, node = node()},
-                             #mqtt_subscriber{topic = Topic, subpid = SubPid, qos = Qos}}
-                end, Topics),
-    F = fun() ->
-            lists:map(fun(QueueR) when is_record(QueueR, mqtt_queue) ->
-                        add_queue(QueueR);
-                         (TopicSub) ->
-                        add_subscriber(TopicSub)
-                end, TopicSubs)
-        end,
-    case mnesia:transaction(F) of
         {atomic, _Result} ->
-            setstats(all),
+            {reply, {ok, [Qos || {_Topic, Qos} <- TopicTable]}, setstats(State)};
-            NewState = monitor_subscriber(SubPid, State),
-            %%TODO: grant all qos
-            {reply, {ok, [Qos || {_Topic, Qos} <- Topics]}, NewState};
-        {aborted, Error} ->
-            {reply, {error, Error}, State}
-    end;
-
-handle_call({subscribe, SubPid, <<"$Q/", _/binary>> = Queue, Qos}, _From, State) ->
-    case mnesia:dirty_read(queue, Queue) of
-        [OldQueueR] -> lager:error("Queue is overwrited by ~p: ~p", [SubPid, OldQueueR]);
-        [] -> ok
-    end,
-    QueueR = #mqtt_queue{name = Queue, qpid = SubPid, qos = Qos},
-    case mnesia:transaction(fun add_queue/1, [QueueR]) of
-        {atomic, ok} ->
-            setstats(queues),
-            {reply, {ok, Qos}, monitor_subscriber(SubPid, State)};
-        {aborted, Error} -> 
-            {reply, {error, Error}, State}
-    end;
-
-handle_call({subscribe, SubPid, Topic, Qos}, _From, State) ->
-    TopicR = #mqtt_topic{topic = Topic, node = node()},
-    Subscriber = #mqtt_subscriber{topic = Topic, subpid = SubPid, qos = Qos},
-    case mnesia:transaction(fun add_subscriber/1, [{TopicR, Subscriber}]) of
-        {atomic, ok} -> 
-            setstats(all),
-            {reply, {ok, Qos}, monitor_subscriber(SubPid, State)};
         {aborted, Error} ->
             {reply, {error, Error}, State}
     end;
@@ -308,95 +200,20 @@ handle_call(Req, _From, State) ->
     lager:error("Bad Request: ~p", [Req]),
 	{reply, {error, badreq}, State}.
 
+%%TODO: clientId???
 handle_cast({unsubscribe, SubPid, Topics}, State) when is_list(Topics) ->
-
-    TopicSubs = lists:map(fun(<<"$Q/", _/binary>> = Queue) ->
-                                #mqtt_queue{name = Queue, qpid = SubPid};
-                             (Topic) ->
-                                {#mqtt_topic{topic = Topic, node = node()},
-                                 #mqtt_subscriber{topic = Topic, subpid = SubPid, _ = '_'}}
-                end, Topics),
-    F = fun() ->
-            lists:foreach(
-                fun(QueueR) when is_record(QueueR, mqtt_queue) ->
-                        remove_queue(QueueR);
-                    (TopicSub) ->
-                        remove_subscriber(TopicSub)
-                end, TopicSubs) 
-        end,
-    case mnesia:transaction(F) of
-        {atomic, _} -> ok;
-        {aborted, Error} -> lager:error("unsubscribe ~p error: ~p", [Topics, Error])
-    end,
-    setstats(all),
-    {noreply, State};
-
-handle_cast({unsubscribe, SubPid, <<"$Q/", _/binary>> = Queue}, State) ->
-    QueueR = #mqtt_queue{name = Queue, qpid = SubPid},
-    case mnesia:transaction(fun remove_queue/1, [QueueR]) of
-        {atomic, _} -> 
-            setstats(queues);
-        {aborted, Error} -> 
-            lager:error("unsubscribe queue ~s error: ~p", [Queue, Error])
-    end,
-    {noreply, State};
-
-handle_cast({unsubscribe, SubPid, Topic}, State) ->
-    TopicR = #mqtt_topic{topic = Topic, node = node()},
-    Subscriber = #mqtt_subscriber{topic = Topic, subpid = SubPid, _ = '_'},
-    case mnesia:transaction(fun remove_subscriber/1, [{TopicR, Subscriber}]) of
-        {atomic, _} -> ok;
-        {aborted, Error} -> lager:error("unsubscribe ~s error: ~p", [Topic, Error])
-    end,
-    setstats(all),
     {noreply, State};
 
 handle_cast(Msg, State) ->
     lager:error("Bad Msg: ~p", [Msg]),
 	{noreply, State}.
 
-handle_info({'DOWN', _Mon, _Type, DownPid, _Info}, State = #state{submap = SubMap}) ->
-    case maps:is_key(DownPid, SubMap) of
-        true ->
-            Node = node(),
-            F = fun() -> 
-                    %% remove queue...
-                    Queues = mnesia:match_object(queue, #mqtt_queue{qpid = DownPid, _ = '_'}, write),
-                    lists:foreach(fun(QueueR) ->
-                                mnesia:delete_object(queue, QueueR, write) 
-                        end, Queues),
-
-                    %% remove subscribers...
-                    Subscribers = mnesia:index_read(subscriber, DownPid, #mqtt_subscriber.subpid),
-                    lists:foreach(fun(Sub = #mqtt_subscriber{topic = Topic}) ->
-                                mnesia:delete_object(subscriber, Sub, write),
-                                try_remove_topic(#mqtt_topic{topic = Topic, node = Node})
-                        end, Subscribers)
-            end,
-            case catch mnesia:transaction(F) of
-                {atomic, _} -> ok;
-                {aborted, Reason} ->
-                    lager:error("Failed to delete 'DOWN' subscriber ~p: ~p", [DownPid, Reason])
-            end,
-            setstats(all),
-            {noreply, State#state{submap = maps:remove(DownPid, SubMap)}};
-        false ->
-            lager:error("Unexpected 'DOWN' from ~p", [DownPid]),
-            {noreply, State}
-    end;
-
 handle_info(Info, State) ->
     lager:error("Unexpected Info: ~p", [Info]),
 	{noreply, State}.
 
-terminate(_Reason, _State) ->
+terminate(_Reason, #state{pool = Pool, id = Id}) ->
-    TopicR = #mqtt_topic{_ = '_', node = node()},
+    ?GPROC_POOL(leave, Pool, Id), setstats(all).
-    F = fun() ->
-            [mnesia:delete_object(topic, R, write) || R <- mnesia:match_object(topic, TopicR, write)]
-            %%TODO: remove trie??
-        end,
-    mnesia:transaction(F),
-    setstats(all).
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
@@ -405,9 +222,6 @@ code_change(_OldVsn, State, _Extra) ->
 %%% Internal functions
 %%%=============================================================================
 
-add_queue(QueueR) ->
-    mnesia:write(queue, QueueR, write).
-
 add_topic(TopicR = #mqtt_topic{topic = Topic}) ->
     case mnesia:wread({topic, Topic}) of
         [] ->
@@ -420,51 +234,6 @@ add_topic(TopicR = #mqtt_topic{topic = Topic}) ->
             end
     end.
 
-%% Fix issue #53 - Remove Overlapping Subscriptions
-add_subscriber({TopicR, Subscriber = #mqtt_subscriber{topic = Topic, subpid = SubPid, qos = Qos}})
-                when is_record(TopicR, mqtt_topic) ->
-    case add_topic(TopicR) of
-        ok ->
-            OverlapSubs = [Sub || Sub = #mqtt_subscriber{topic = SubTopic, qos = SubQos}
-                                    <- mnesia:index_read(subscriber, SubPid, #mqtt_subscriber.subpid),
-                                        SubTopic =:= Topic, SubQos =/= Qos],
-
-            %% remove overlapping subscribers
-            if
-                length(OverlapSubs) =:= 0 -> ok;
-                true ->
-                    lager:warning("Remove overlapping subscribers: ~p", [OverlapSubs]),
-                    [mnesia:delete_object(subscriber, OverlapSub, write) || OverlapSub <- OverlapSubs]
-            end,
-
-            %% insert subscriber
-            mnesia:write(subscriber, Subscriber, write);
-        Error -> 
-            Error
-    end.
-
-monitor_subscriber(SubPid, State = #state{submap = SubMap}) ->
-    NewSubMap = case maps:is_key(SubPid, SubMap) of
-        false ->
-            maps:put(SubPid, erlang:monitor(process, SubPid), SubMap);
-        true ->
-            SubMap
-    end,
-    State#state{submap = NewSubMap}.
-
-remove_queue(#mqtt_queue{name = Name, qpid = Pid}) ->
-    case mnesia:wread({queue, Name}) of
-        [R = #mqtt_queue{qpid = Pid}] -> 
-            mnesia:delete(queue, R, write);
-        _ ->
-            ok
-    end.
-
-remove_subscriber({TopicR, Subscriber}) when is_record(TopicR, mqtt_topic) ->
-    [mnesia:delete_object(subscriber, Sub, write) || 
-        Sub <- mnesia:match_object(subscriber, Subscriber, write)],
-    try_remove_topic(TopicR).
-
 try_remove_topic(TopicR = #mqtt_topic{topic = Topic}) ->
     case mnesia:read({subscriber, Topic}) of
         [] ->
@@ -481,24 +250,9 @@ try_remove_topic(TopicR = #mqtt_topic{topic = Topic}) ->
 %%% Stats functions
 %%%=============================================================================
 
-setstats(all) ->
+setstats(State) ->
-    [setstats(Stat) || Stat <- [queues, topics, subscribers]];
-
-setstats(queues) ->
-    emqttd_stats:setstats('queues/count', 'queues/max',
-                           mnesia:table_info(queue, size));
-
-setstats(topics) ->
     emqttd_stats:setstats('topics/count', 'topics/max',
-                          mnesia:table_info(topic, size));
+                          mnesia:table_info(topic, size)), State.
-setstats(subscribers) ->
-    emqttd_stats:setstats('subscribers/count', 'subscribers/max',
-                           mnesia:table_info(subscriber, size)).
-
-setstats(dropped, false) ->
-    ignore;
-setstats(dropped, true) ->
-    emqttd_metrics:inc('messages/dropped').
 
 %%%=============================================================================
 %%% Trace functions

src/emqttd_pubsub_sup.erl

@@ -19,18 +19,18 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc PubSub Supervisor
-%%% emqttd pubsub supervisor.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_pubsub_sup).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(supervisor).
 
 -include("emqttd.hrl").
 
--behaviour(supervisor).
+-define(HELPER, emqttd_pubsub_helper).
 
 %% API
 -export([start_link/0]).
@@ -39,19 +39,42 @@
 -export([init/1]).
 
 start_link() ->
-    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
-
-init([]) ->
     Opts = emqttd_broker:env(pubsub),
-    Schedulers = erlang:system_info(schedulers),
+    supervisor:start_link({local, ?MODULE}, ?MODULE, [Opts]).
-    PoolSize = proplists:get_value(pool_size, Opts, Schedulers),
+
-    gproc_pool:new(pubsub, hash, [{size, PoolSize}]),
+init([Opts]) ->
-    Children = lists:map(
+    %% Route Table
-                 fun(I) ->
+    create_route_tabs(Opts),
-                    Name = {emqttd_pubsub, I},
+
-                    gproc_pool:add_worker(pubsub, Name, I),
+    %% PubSub Pool Sup
-                    {Name, {emqttd_pubsub, start_link, [I, Opts]},
+    MFA = {emqttd_pubsub, start_link, [Opts]},
-                        permanent, 10000, worker, [emqttd_pubsub]}
+    PoolSup = emqttd_pool_sup:spec(pool_sup, [
-                 end, lists:seq(1, PoolSize)),
+                pubsub, hash, pool_size(Opts), MFA]),
-    {ok, {{one_for_all, 10, 100}, Children}}.
+
+    %% PubSub Helper
+    Helper = {helper, {?HELPER, start_link, [Opts]},
+                permanent, infinity, worker, [?HELPER]},
+    {ok, {{one_for_all, 10, 60}, [Helper, PoolSup]}}.
+
+pool_size(Opts) ->
+    Schedulers = erlang:system_info(schedulers),
+    proplists:get_value(pool_size, Opts, Schedulers).
+
+create_route_tabs(_Opts) ->
+    TabOpts = [bag, public, named_table,
+               {write_concurrency, true}],
+    %% Route Table: Topic -> {Pid, QoS}
+    %% Route Shard: {Topic, Shard} -> {Pid, QoS}
+    ensure_tab(route, TabOpts),
+
+    %% Reverse Route Table: Pid -> {Topic, QoS}
+    ensure_tab(reverse_route, TabOpts).
+
+ensure_tab(Tab, Opts) ->
+    case ets:info(Tab, name) of
+        undefined ->
+            ets:new(Tab, Opts);
+        _ ->
+            ok
+    end.
 

src/emqttd_retainer.erl

@@ -19,16 +19,18 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc MQTT retained message storage.
-%%% MQTT retained message storage.
 %%%
 %%% TODO: should match topic tree
 %%%
 %%% @end
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
+%%% 
 %%%-----------------------------------------------------------------------------
--module(emqttd_retained).
+-module(emqttd_retainer).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(gen_server).
 
 -include("emqttd.hrl").
 
@@ -46,8 +48,6 @@
 %% API Function Exports
 -export([start_link/0, expire/1]).
 
--behaviour(gen_server).
-
 %% gen_server Function Exports
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).

src/emqttd_serialiser.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc MQTT Packet Serialiser
-%%% MQTT Packet Serialiser.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_serialiser).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").

src/emqttd_session.erl

@@ -19,9 +19,7 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Session for persistent MQTT client.
-%%%
-%%% Session for persistent MQTT client.
 %%%
 %%% Session State in the broker consists of:
 %%%
@@ -43,11 +41,12 @@
 %%% State of Message:  newcome, inflight, pending
 %%%
 %%% @end
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
+%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_session).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").
@@ -344,7 +343,7 @@ handle_cast({subscribe, TopicTable0, AckFun}, Session = #session{client_id = Cli
                                     %% <MQTT V3.1.1>: 3.8.4
                                     %% Where the Topic Filter is not identical to any existing Subscription’s filter,
                                     %% a new Subscription is created and all matching retained messages are sent.
-                                    emqttd_retained:dispatch(Topic, self()),
+                                    emqttd_retainer:dispatch(Topic, self()),
                                     [{Topic, Qos} | Acc]
                             end
                         end, Subscriptions, TopicTable),

src/emqttd_session_sup.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd session supervisor.
-%%% emqttd session supervisor.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_session_sup).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -behavior(supervisor).
 
 -export([start_link/0, start_session/3]).

src/emqttd_sm.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Session Manager
-%%% emqttd session manager.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_sm).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 %% Mnesia Callbacks
@@ -54,7 +52,7 @@
 
 -record(state, {id}).
 
--define(SM_POOL, ?MODULE).
+-define(POOL, ?MODULE).
 
 -define(TIMEOUT, 60000).
 
@@ -92,19 +90,13 @@ start_link(Id) ->
 name(Id) ->
     list_to_atom("emqttd_sm_" ++ integer_to_list(Id)).
 
-%%------------------------------------------------------------------------------
-%% @doc Pool name.
-%% @end
-%%------------------------------------------------------------------------------
-pool() -> ?SM_POOL.
-
 %%------------------------------------------------------------------------------
 %% @doc Start a session
 %% @end
 %%------------------------------------------------------------------------------
 -spec start_session(CleanSess :: boolean(), binary()) -> {ok, pid(), boolean()} | {error, any()}.
 start_session(CleanSess, ClientId) ->
-    SM = gproc_pool:pick_worker(?SM_POOL, ClientId),
+    SM = gproc_pool:pick_worker(?POOL, ClientId),
     call(SM, {start_session, {CleanSess, ClientId, self()}}).
 
 %%------------------------------------------------------------------------------
@@ -150,7 +142,7 @@ call(SM, Req) ->
 %%%=============================================================================
 
 init([Id]) ->
-    gproc_pool:connect_worker(?SM_POOL, {?MODULE, Id}),
+    gproc_pool:connect_worker(?POOL, {?MODULE, Id}),
     {ok, #state{id = Id}}.
 
 prioritise_call(_Msg, _From, _Len, _State) ->
@@ -193,6 +185,7 @@ handle_cast(Msg, State) ->
     lager:error("Unexpected Msg: ~p", [Msg]),
     {noreply, State}.
 
+%%TODO: fix this issue that index_read is really slow...
 handle_info({'DOWN', _MRef, process, DownPid, _Reason}, State) ->
     mnesia:transaction(fun() ->
                 [mnesia:delete_object(session, Sess, write) || Sess 
@@ -205,7 +198,7 @@ handle_info(Info, State) ->
     {noreply, State}.
 
 terminate(_Reason, #state{id = Id}) ->
-    gproc_pool:disconnect_worker(?SM_POOL, {?MODULE, Id}), ok.
+    gproc_pool:disconnect_worker(?POOL, {?MODULE, Id}), ok.
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.

src/emqttd_sm_helper.erl

@@ -19,23 +19,21 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Session Helper.
-%%% emqttd session helper.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_sm_helper).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(gen_server).
 
 -include("emqttd.hrl").
 
 -include_lib("stdlib/include/ms_transform.hrl").
 
 %% API Function Exports
--export([start_link/0]).
+-export([start_link/1]).
-
--behaviour(gen_server).
 
 %% gen_server Function Exports
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
@@ -47,11 +45,11 @@
 %% @doc Start a session helper
 %% @end
 %%------------------------------------------------------------------------------
--spec start_link() -> {ok, pid()} | ignore | {error, any()}.
+-spec start_link(fun()) -> {ok, pid()} | ignore | {error, any()}.
-start_link() ->
+start_link(StatsFun) ->
-    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [StatsFun], []).
 
-init([]) ->
+init([StatsFun]) ->
     mnesia:subscribe(system),
     {ok, TRef} = timer:send_interval(timer:seconds(1), tick),
     StatsFun = emqttd_stats:statsfun('sessions/count', 'sessions/max'),

src/emqttd_sm_sup.erl

@@ -19,25 +19,27 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Session Manager Supervisor.
-%%% emqttd session manager supervisor.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_sm_sup).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(supervisor).
 
 -include("emqttd.hrl").
 
--define(CHILD(Mod), {Mod, {Mod, start_link, []},
+-define(SM, emqttd_sm).
-                        permanent, 5000, worker, [Mod]}).
+
+-define(HELPER, emqttd_sm_helper).
+
+-define(TABS, [mqtt_transient_session,
+               mqtt_persistent_session]).
 
 %% API
 -export([start_link/0]).
 
--behaviour(supervisor).
-
 %% Supervisor callbacks
 -export([init/1]).
 
@@ -45,20 +47,23 @@ start_link() ->
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
 init([]) ->
-    init_session_ets(),
+    %% Create session tables
-    Schedulers = erlang:system_info(schedulers),
+    create_session_tabs(),
-    gproc_pool:new(emqttd_sm:pool(), hash, [{size, Schedulers}]),
-    Managers = lists:map(
-                 fun(I) ->
-                   Name = {emqttd_sm, I},
-                   gproc_pool:add_worker(emqttd_sm:pool(), Name, I),
-                   {Name, {emqttd_sm, start_link, [I]},
-                               permanent, 10000, worker, [emqttd_sm]}
-                 end, lists:seq(1, Schedulers)),
-    {ok, {{one_for_all, 10, 100}, [?CHILD(emqttd_sm_helper) | Managers]}}.
 
-init_session_ets() ->
+    %% Helper
-    Tables = [mqtt_transient_session, mqtt_persistent_session],
+    StatsFun = emqttd_stats:statsfun('sessions/count', 'sessions/max'),
-    Attrs  = [ordered_set, named_table, public, {write_concurrency, true}],
+    Helper = {?HELPER, {?HELPER, start_link, [StatsFun]},
-    lists:foreach(fun(Tab) -> ets:new(Tab, Attrs) end, Tables).
+                permanent, 5000, worker, [?HELPER]},
+
+    %% SM Pool Sup
+    MFA = {?SM, start_link, []},
+    PoolSup = emqttd_pool_sup:spec(pool_sup, [
+                ?SM, hash, erlang:system_info(schedulers), MFA]),
+
+    {ok, {{one_for_all, 10, 3600}, [Helper, PoolSup]}}.
+    
+create_session_tabs() ->
+    Opts = [ordered_set, named_table, public,
+               {write_concurrency, true}],
+    [ets:new(Tab, Opts) || Tab <- ?TABS].
 

src/emqttd_stats.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd statistics
-%%% emqttd statistics.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_stats).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -behaviour(gen_server).

src/emqttd_sup.erl

@@ -19,19 +19,17 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd top supervisor.
-%%% emqttd supervisor.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_sup).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(supervisor).
 
 -include("emqttd.hrl").
 
--behaviour(supervisor).
-
 %% API
 -export([start_link/0, start_child/1, start_child/2]).
 

src/emqttd_sysmon.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd system monitor
-%%% emqttd system monitor.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_sysmon).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -behavior(gen_server).
 
 -export([start_link/1]).

src/emqttd_topic.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc MQTT Topic Functions
-%%% MQTT Topic Functions
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_topic).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -import(lists, [reverse/1]).
  
 -export([match/2, validate/1, triples/1, words/1, wildcard/1]).

src/emqttd_trace.erl

@@ -19,22 +19,20 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc Trace MQTT packets/messages by clientid or topic.
-%%% Trace MQTT packets/messages by clientid or topic.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_trace).
 
--author("Feng Lee <feng@emqtt.io>").
+-behaviour(gen_server).
 
 %% API Function Exports
 -export([start_link/0]).
 
 -export([start_trace/2, stop_trace/1, all_traces/0]).
 
--behaviour(gen_server).
-
 %% gen_server Function Exports
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).

src/emqttd_trie.erl

@@ -25,11 +25,12 @@
 %%% [Trie](http://en.wikipedia.org/wiki/Trie)
 %%%
 %%% @end
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
+%%%
 %%%-----------------------------------------------------------------------------
 -module(emqttd_trie).
 
--author("Feng Lee <feng@emqtt.io>").
-
 %% Mnesia Callbacks
 -export([mnesia/1]).
 

src/emqttd_util.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd utility functions
-%%% emqttd utility functions. 
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_util).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -export([apply_module_attributes/1,
          all_module_attributes/1,
          cancel_timer/1,

src/emqttd_vm.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd erlang vm.
-%%% emqttd erlang vm.
+%%%
+%%% @author huangdan
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_vm).
 
--author('huangdan').
-
 -export([schedulers/0]).
 
 -export([microsecs/0]).

src/emqttd_ws_client.erl

@@ -19,15 +19,13 @@
 %%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 %%% SOFTWARE.
 %%%-----------------------------------------------------------------------------
-%%% @doc
+%%% @doc emqttd websocket client
-%%% emqttd websocket client.
+%%%
+%%% @author Feng Lee <feng@emqtt.io>
 %%%
-%%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_ws_client).
 
--author("Feng Lee <feng@emqtt.io>").
-
 -include("emqttd.hrl").
 
 -include("emqttd_protocol.hrl").