Merge pull request #787 from emqtt/emq20

Stop plugins before the broker stopped, clean routes when a node down

src/emqttd.erl

@@ -14,7 +14,7 @@
 %% limitations under the License.
 %%--------------------------------------------------------------------
 
-%% Facade Module for The EMQTT Broker
+%% Facade Module for The EMQ Broker
 
 -module(emqttd).
 
@@ -22,7 +22,7 @@
 
 -include("emqttd_protocol.hrl").
 
--export([start/0, env/1, env/2, is_running/1]).
+-export([start/0, env/1, env/2, is_running/1, stop/0]).
 
 %% PubSub API
 -export([subscribe/1, subscribe/2, subscribe/3, publish/1,
@@ -57,6 +57,10 @@
 -spec(start() -> ok | {error, any()}).
 start() -> application:start(?APP).
 
+%% @doc Stop emqttd application.
+-spec(stop() -> ok | {error, any()}).
+stop() -> application:stop(?APP).
+
 %% @doc Environment
 -spec(env(Key:: atom()) -> {ok, any()} | undefined).
 env(Key) -> application:get_env(?APP, Key).

src/emqttd_router.erl

@@ -284,11 +284,11 @@ code_change(_OldVsn, State, _Extra) ->
 clean_routes_(Node) ->
     Pattern = #mqtt_route{_ = '_', node = Node},
     Clean = fun() ->
-                [mnesia:delete_object(route, R, write) ||
-                    R <- mnesia:match_object(route, Pattern, write)]
+                [mnesia:delete_object(mqtt_route, R, write) ||
+                    R <- mnesia:match_object(mqtt_route, Pattern, write)]
             end,
     mnesia:transaction(Clean).
 
 update_stats_() ->
-    emqttd_stats:setstats('routes/count', 'routes/max', mnesia:table_info(route, size)).
+    emqttd_stats:setstats('routes/count', 'routes/max', mnesia:table_info(mqtt_route, size)).
 

src/emqttd_session.erl

@@ -332,10 +332,14 @@ handle_cast({unsubscribe, TopicTable}, Session = #session{client_id     = Client
         end, Subscriptions, TopicTable),
     hibernate(Session#session{subscriptions = Subscriptions1});
 
-handle_cast({destroy, ClientId}, Session = #session{client_id = ClientId}) ->
+handle_cast({destroy, ClientId}, Session = #session{client_id = ClientId, client_pid = undefined}) ->
     ?LOG(warning, "destroyed", [], Session),
     shutdown(destroy, Session);
 
+handle_cast({destroy, ClientId}, Session = #session{client_id = ClientId, client_pid = OldClientPid}) ->
+    ?LOG(warning, "kickout ~p", [OldClientPid], Session),
+    shutdown(conflict, Session);
+
 handle_cast({resume, ClientId, ClientPid}, Session = #session{client_id      = ClientId,
                                                               client_pid     = OldClientPid,
                                                               clean_sess     = CleanSess,