Merge branch 'dev' of github.com:emqtt/emqtt into dev

apps/emqttd/src/emqttd.erl

@@ -162,7 +162,7 @@ loaded_plugins() ->
 -spec unload_all_plugins() -> [{App :: atom(), ok | {error, any()}}].
 unload_all_plugins() ->
     PluginApps = application:get_env(emqttd, plugins, []),
-    [{App, unload_plugin(App)} || {App, _Env} <- PluginApps].
+    [{App, unload_plugin(App)} || App <- PluginApps].
 
 
 %%------------------------------------------------------------------------------

apps/emqttd/src/emqttd_app.erl

@@ -31,7 +31,7 @@
 -behaviour(application).
 
 %% Application callbacks
--export([start/2, stop/1]).
+-export([start/2, prep_stop/1, stop/1]).
 
 -define(PRINT_MSG(Msg), io:format(Msg)).
 
@@ -47,24 +47,24 @@
     State     :: term(),
     Reason    :: term().
 start(_StartType, _StartArgs) ->
-	print_banner(),
+    print_banner(),
     emqttd_mnesia:start(),
     {ok, Sup} = emqttd_sup:start_link(),
-	start_servers(Sup),
-	{ok, Listeners} = application:get_env(listeners),
+    start_servers(Sup),
+    {ok, Listeners} = application:get_env(listeners),
     emqttd:load_all_plugins(),
     emqttd:open_listeners(Listeners),
-	register(emqttd, self()),
+    register(emqttd, self()),
     print_vsn(),
-	{ok, Sup}.
+    {ok, Sup}.
 
 print_banner() ->
-	?PRINT("starting emqttd on node '~s'~n", [node()]).
+    ?PRINT("starting emqttd on node '~s'~n", [node()]).
 
 print_vsn() ->
-	{ok, Vsn} = application:get_key(vsn),
-	{ok, Desc} = application:get_key(description),
-	?PRINT("~s ~s is running now~n", [Desc, Vsn]).
+    {ok, Vsn} = application:get_key(vsn),
+    {ok, Desc} = application:get_key(description),
+    ?PRINT("~s ~s is running now~n", [Desc, Vsn]).
 
 start_servers(Sup) ->
     Servers = [{"emqttd event", emqttd_event},
@@ -128,10 +128,16 @@ worker_spec(Name, Opts) ->
         {Name, start_link, [Opts]},
             permanent, 10000, worker, [Name]}.
 
+%% close all listeners first...
+prep_stop(State) ->
+    %%TODO: esockd app should be running...
+    {ok, Listeners} = application:get_env(listeners),
+    emqttd:close_listeners(Listeners),
+    timer:sleep(2),
+    State.
+
 -spec stop(State :: term()) -> term().
 stop(_State) ->
-	{ok, Listeners} = application:get_env(listeners),
-    emqttd:close_listeners(Listeners),
-    emqttd:unload_all_plugins(),
     ok.
 
+

apps/emqttd/src/emqttd_ctl.erl

@@ -57,10 +57,10 @@ status([]) ->
     {InternalStatus, _ProvidedStatus} = init:get_status(),
     ?PRINT("Node ~p is ~p~n", [node(), InternalStatus]),
     case lists:keysearch(emqttd, 1, application:which_applications()) of
-	false ->
-		?PRINT_MSG("emqttd is not running~n");
-	{value,_Version} ->
-		?PRINT_MSG("emqttd is running~n")
+    false ->
+        ?PRINT_MSG("emqttd is not running~n");
+    {value,_Version} ->
+        ?PRINT_MSG("emqttd is running~n")
     end.
 
 %%------------------------------------------------------------------------------
@@ -72,32 +72,41 @@ cluster([]) ->
     ?PRINT("cluster nodes: ~p~n", [Nodes]);
 
 cluster([SNode]) ->
-	Node = node_name(SNode),
-	case net_adm:ping(Node) of
-	pong ->
-		application:stop(emqttd),
-        application:stop(esockd),
-        emqttd_mnesia:cluster(Node),
-        application:start(esockd),
-		application:start(emqttd),
-		?PRINT("cluster with ~p successfully.~n", [Node]);
-	pang ->
+    Node = node_name(SNode),
+    case net_adm:ping(Node) of
+    pong ->
+        case emqttd:is_running(Node) of
+            true ->
+                %%TODO: should not unload here.
+                emqttd:unload_all_plugins(),
+                application:stop(emqttd),
+                application:stop(esockd),
+                application:stop(gproc),
+                emqttd_mnesia:cluster(Node),
+                application:start(gproc),
+                application:start(esockd),
+                application:start(emqttd),
+                ?PRINT("cluster with ~p successfully.~n", [Node]);
+            false ->
+                ?PRINT("emqttd is not running on ~p~n", [Node])
+        end;
+    pang ->
         ?PRINT("failed to connect to ~p~n", [Node])
-	end.
+    end.
 
 %%------------------------------------------------------------------------------
 %% @doc Add usern
 %% @end
 %%------------------------------------------------------------------------------
 useradd([Username, Password]) ->
-	?PRINT("~p~n", [emqttd_auth_username:add_user(bin(Username), bin(Password))]).
+    ?PRINT("~p~n", [emqttd_auth_username:add_user(bin(Username), bin(Password))]).
 
 %%------------------------------------------------------------------------------
 %% @doc Delete user
 %% @end
 %%------------------------------------------------------------------------------
 userdel([Username]) ->
-	?PRINT("~p~n", [emqttd_auth_username:remove_user(bin(Username))]).
+    ?PRINT("~p~n", [emqttd_auth_username:remove_user(bin(Username))]).
 
 vm([]) ->
     [vm([Name]) || Name <- ["load", "memory", "process", "io"]];