From 153977609e6c73b5600a24e0f5c75a44164e96b7 Mon Sep 17 00:00:00 2001
From: Zaiming Shi <zmstone@gmail.com>
Date: Sat, 13 Feb 2021 21:07:31 +0100
Subject: [PATCH 1/8] feat(listeners): Add identifier to listeners

Listeners are internally identifiered by the listen-on tuple
which is not UI friendly when we have to find a listener by this
'signature'.

The listeners are actually named in configs, but the names are
discarded in the parsing functions.

This commit is to keep the name and provide an API to find
listener by name (identifier).
---
 .../emqx_management/src/emqx_mgmt.erl         |  1 +
 .../emqx_management/src/emqx_mgmt_cli.erl     | 43 +++++++----
 priv/emqx.schema                              | 24 ++++--
 src/emqx_listeners.erl                        | 77 +++++++++++++++----
 4 files changed, 111 insertions(+), 34 deletions(-)

diff --git a/lib-opensource/emqx_management/src/emqx_mgmt.erl b/lib-opensource/emqx_management/src/emqx_mgmt.erl
index 4174f3224..63a2b22a3 100644
--- a/lib-opensource/emqx_management/src/emqx_mgmt.erl
+++ b/lib-opensource/emqx_management/src/emqx_mgmt.erl
@@ -535,6 +535,7 @@ list_listeners(Node) when Node =:= node() ->
     Tcp = lists:map(fun({{Protocol, ListenOn}, _Pid}) ->
         #{protocol        => Protocol,
           listen_on       => ListenOn,
+          identifier      => emqx_listeners:find_id_by_listen_on(ListenOn),
           acceptors       => esockd:get_acceptors({Protocol, ListenOn}),
           max_conns       => esockd:get_max_connections({Protocol, ListenOn}),
           current_conns   => esockd:get_current_connections({Protocol, ListenOn}),
diff --git a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
index 8a9f57f67..6d2315acc 100644
--- a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
+++ b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
@@ -510,14 +510,14 @@ trace_off(Who, Name) ->
 
 listeners([]) ->
     foreach(fun({{Protocol, ListenOn}, _Pid}) ->
-                Info = [{acceptors,      esockd:get_acceptors({Protocol, ListenOn})},
+                Info = [{identifier,     {string, emqx_listeners:find_id_by_listen_on(ListenOn)}},
+                        {acceptors,      esockd:get_acceptors({Protocol, ListenOn})},
                         {max_conns,      esockd:get_max_connections({Protocol, ListenOn})},
                         {current_conn,   esockd:get_current_connections({Protocol, ListenOn})},
-                        {shutdown_count, esockd:get_shutdown_count({Protocol, ListenOn})}],
+                        {shutdown_count, esockd:get_shutdown_count({Protocol, ListenOn})}
+                       ],
                     emqx_ctl:print("listener on ~s:~s~n", [Protocol, esockd:to_string(ListenOn)]),
-                foreach(fun({Key, Val}) ->
-                            emqx_ctl:print("  ~-16s: ~w~n", [Key, Val])
-                        end, Info)
+                foreach(fun indent_print/1, Info)
             end, esockd:listeners()),
     foreach(fun({Protocol, Opts}) ->
                 Info = [{acceptors,      maps:get(num_acceptors, proplists:get_value(transport_options, Opts, #{}), 0)},
@@ -525,9 +525,7 @@ listeners([]) ->
                         {current_conn,   proplists:get_value(all_connections, Opts)},
                         {shutdown_count, []}],
                     emqx_ctl:print("listener on ~s:~p~n", [Protocol, proplists:get_value(port, Opts)]),
-                foreach(fun({Key, Val}) ->
-                            emqx_ctl:print("  ~-16s: ~w~n", [Key, Val])
-                        end, Info)
+                foreach(fun indent_print/1, Info)
             end, ranch:info());
 
 listeners(["stop",  Name = "http" ++ _N, ListenOn]) ->
@@ -538,22 +536,32 @@ listeners(["stop",  Name = "http" ++ _N, ListenOn]) ->
             emqx_ctl:print("Failed to stop ~s listener on ~s, error:~p~n", [Name, ListenOn, Error])
     end;
 
-listeners(["stop", Proto, ListenOn]) ->
+listeners(["stop", "mqtt:" ++ _ = Identifier]) ->
+    stop_listener(emqx_listeners:find_by_id(Identifier), Identifier);
+
+listeners(["stop", _Proto, ListenOn]) ->
+    %% this clause is kept to be backward compatible
     ListenOn1 = case string:tokens(ListenOn, ":") of
         [Port]     -> list_to_integer(Port);
         [IP, Port] -> {IP, list_to_integer(Port)}
     end,
-    case emqx_listeners:stop_listener({list_to_atom(Proto), ListenOn1, []}) of
-        ok ->
-            emqx_ctl:print("Stop ~s listener on ~s successfully.~n", [Proto, ListenOn]);
-        {error, Error} ->
-            emqx_ctl:print("Failed to stop ~s listener on ~s, error:~p~n", [Proto, ListenOn, Error])
-    end;
+    stop_listener(emqx_listeners:find_by_listen_on(ListenOn1), ListenOn1);
 
 listeners(_) ->
     emqx_ctl:usage([{"listeners",                        "List listeners"},
                     {"listeners stop    <Proto> <Port>", "Stop a listener"}]).
 
+stop_listener(false, Input) ->
+    emqx_ctl:print("No such listener ~p~n", [Input]);
+stop_listener(#{listen_on := ListenOn} = Listener, _Input) ->
+    ID = emqx_listeners:identifier(Listener),
+    case emqx_listeners:stop_listener(Listener) of
+        ok ->
+            emqx_ctl:print("Stop ~s listener on ~s successfully.~n", [ID, ListenOn]);
+        {error, Error} ->
+            emqx_ctl:print("Failed to stop ~s listener on ~s, error:~p~n", [ID, ListenOn, Error])
+    end.
+
 %%--------------------------------------------------------------------
 %% @doc data Command
 
@@ -707,3 +715,8 @@ format(_, Val) ->
     Val.
 
 bin(S) -> iolist_to_binary(S).
+
+indent_print({Key, {string, Val}}) ->
+    emqx_ctl:print("  ~-16s: ~s~n", [Key, Val]);
+indent_print({Key, Val}) ->
+    emqx_ctl:print("  ~-16s: ~w~n", [Key, Val]).
diff --git a/priv/emqx.schema b/priv/emqx.schema
index bdf8a053f..26713ce2a 100644
--- a/priv/emqx.schema
+++ b/priv/emqx.schema
@@ -1996,8 +1996,15 @@ end}.
                                   Other -> Other
                               end
                       end,
-                      [{Atom(Type), ListenOnN, [{deflate_options, DeflateOpts(Prefix)},
-                                              {tcp_options, TcpOpts(Prefix)} | LisOpts(Prefix)]}]
+                      [#{ proto => Atom(Type)
+                        , name => Name
+                        , listen_on => ListenOnN
+                        , opts => [ {deflate_options, DeflateOpts(Prefix)}
+                                  , {tcp_options, TcpOpts(Prefix)}
+                                  | LisOpts(Prefix)
+                                  ]
+                        }
+                      ]
                    end,
     SslListeners = fun(Type, Name) ->
                        Prefix = string:join(["listener", Type, Name], "."),
@@ -2005,9 +2012,16 @@ end}.
                            undefined ->
                                [];
                            ListenOn ->
-                               [{Atom(Type), ListenOn, [{deflate_options, DeflateOpts(Prefix)},
-                                                        {tcp_options, TcpOpts(Prefix)},
-                                                        {ssl_options, SslOpts(Prefix)} | LisOpts(Prefix)]}]
+                               [#{ proto => Atom(Type)
+                                 , name => Name
+                                 , listen_on => ListenOn
+                                 , opts => [ {deflate_options, DeflateOpts(Prefix)}
+                                           , {tcp_options, TcpOpts(Prefix)}
+                                           , {ssl_options, SslOpts(Prefix)}
+                                           | LisOpts(Prefix)
+                                           ]
+                                 }
+                               ]
                        end
                    end,
 
diff --git a/src/emqx_listeners.erl b/src/emqx_listeners.erl
index 3b642d55f..f7912a7fd 100644
--- a/src/emqx_listeners.erl
+++ b/src/emqx_listeners.erl
@@ -33,11 +33,43 @@
         , restart_listener/3
         ]).
 
--type(listener() :: {esockd:proto(), esockd:listen_on(), [esockd:option()]}).
+-export([ find_id_by_listen_on/1
+        , find_by_listen_on/1
+        , find_by_id/1
+        , identifier/1
+        ]).
 
-%%--------------------------------------------------------------------
-%% APIs
-%%--------------------------------------------------------------------
+-type(listener() :: #{ name := binary()
+                     , proto := esockd:proto()
+                     , listen_on := esockd:listen_on()
+                     , opts := [esockd:option()]
+                     }).
+
+%% @doc Find listener identifier by listen-on.
+%% Return empty string (binary) if listener is not found in config.
+-spec(find_id_by_listen_on(esockd:listen_on()) -> binary()).
+find_id_by_listen_on(ListenOn) ->
+    case find_by_listen_on(ListenOn) of
+        false -> <<>>;
+        L -> identifier(L)
+    end.
+
+%% @doc Find listener by listen-on.
+%% Return 'false' if not found.
+-spec(find_by_listen_on(esockd:listen_on()) -> listener() | false).
+find_by_listen_on(ListenOn) ->
+    find_by_listen_on(ListenOn, emqx:get_env(listeners, [])).
+
+%% @doc Find listener by identifier.
+%% Return 'false' if not found.
+-spec(find_by_id(string() | binary()) -> listener() | false).
+find_by_id(Id) ->
+    find_by_id(iolist_to_binary(Id), emqx:get_env(listeners, [])).
+
+%% @doc Return the ID of the given listener.
+-spec identifier(listener()) -> binary().
+identifier(#{proto := Proto, name := Name}) ->
+    identifier(Proto, Name).
 
 %% @doc Start all listeners.
 -spec(start() -> ok).
@@ -45,13 +77,14 @@ start() ->
     lists:foreach(fun start_listener/1, emqx:get_env(listeners, [])).
 
 -spec(start_listener(listener()) -> ok).
-start_listener({Proto, ListenOn, Options}) ->
+start_listener(#{proto := Proto, name := Name, listen_on := ListenOn, opts := Options}) ->
+    ID = identifier(Proto, Name),
     case start_listener(Proto, ListenOn, Options) of
-        {ok, _} -> io:format("Start mqtt:~s listener on ~s successfully.~n",
-                             [Proto, format(ListenOn)]);
+        {ok, _} -> io:format("Start ~s listener on ~s successfully.~n",
+                             [ID, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to start mqtt:~s listener on ~s - ~0p~n!",
-                      [Proto, format(ListenOn), Reason]),
+            io:format(standard_error, "Failed to start mqtt listener ~s on ~s - ~0p~n!",
+                      [ID, format(ListenOn), Reason]),
             error(Reason)
     end.
 
@@ -115,7 +148,7 @@ restart() ->
     lists:foreach(fun restart_listener/1, emqx:get_env(listeners, [])).
 
 -spec(restart_listener(listener()) -> any()).
-restart_listener({Proto, ListenOn, Options}) ->
+restart_listener(#{proto := Proto, listen_on := ListenOn, opts := Options}) ->
     restart_listener(Proto, ListenOn, Options).
 
 -spec(restart_listener(esockd:proto(), esockd:listen_on(), [esockd:option()]) -> any()).
@@ -138,14 +171,14 @@ stop() ->
     lists:foreach(fun stop_listener/1, emqx:get_env(listeners, [])).
 
 -spec(stop_listener(listener()) -> ok | {error, term()}).
-stop_listener({Proto, ListenOn, Opts}) ->
+stop_listener(#{proto := Proto, name := Name, listen_on := ListenOn, opts := Opts}) ->
+    ID = identifier(Proto, Name),
     StopRet = stop_listener(Proto, ListenOn, Opts),
     case StopRet of
-        ok -> io:format("Stop mqtt:~s listener on ~s successfully.~n",
-                        [Proto, format(ListenOn)]);
+        ok -> io:format("Stop ~s listener on ~s successfully.~n", [ID, format(ListenOn)]);
         {error, Reason} ->
             io:format(standard_error, "Failed to stop mqtt:~s listener on ~s - ~p~n.",
-                      [Proto, format(ListenOn), Reason])
+                      [ID, format(ListenOn), Reason])
     end,
     StopRet.
 
@@ -181,3 +214,19 @@ ws_name(Name, {_Addr, Port}) ->
     ws_name(Name, Port);
 ws_name(Name, Port) ->
     list_to_atom(lists:concat([Name, ":", Port])).
+
+identifier(Proto, Name) when is_atom(Proto) ->
+    identifier(atom_to_list(Proto), Name);
+identifier(Proto, Name) ->
+    iolist_to_binary(["mqtt", ":", Proto, ":", Name]).
+
+find_by_listen_on(ListenOn, []) -> error({unknown_listener, ListenOn});
+find_by_listen_on(ListenOn, [#{listen_on := ListenOn} = L | _]) -> L;
+find_by_listen_on(ListenOn, [_ | Rest]) -> find_by_listen_on(ListenOn, Rest).
+
+find_by_id(_Id, []) -> false;
+find_by_id(Id, [L | Rest]) ->
+    case identifier(L) =:= Id of
+        true -> L;
+        false -> find_by_id(Id, Rest)
+    end.

From 2d79c870c1711430b9ff6d20bff46b3e4747bcee Mon Sep 17 00:00:00 2001
From: Zaiming Shi <zmstone@gmail.com>
Date: Sun, 14 Feb 2021 20:16:42 +0100
Subject: [PATCH 2/8] fix(mgmt_cli): Do not doulbe print listener stop results

---
 .../emqx_management/src/emqx_mgmt_cli.erl       | 17 ++++++++---------
 1 file changed, 8 insertions(+), 9 deletions(-)

diff --git a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
index 6d2315acc..ea5345772 100644
--- a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
+++ b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
@@ -549,18 +549,17 @@ listeners(["stop", _Proto, ListenOn]) ->
 
 listeners(_) ->
     emqx_ctl:usage([{"listeners",                        "List listeners"},
-                    {"listeners stop    <Proto> <Port>", "Stop a listener"}]).
+                    {"listeners stop    <Identifier>",   "Stop a listener"},
+                    {"listeners stop    <Proto> <Port>", "Stop a listener"}
+                   ]).
 
 stop_listener(false, Input) ->
     emqx_ctl:print("No such listener ~p~n", [Input]);
-stop_listener(#{listen_on := ListenOn} = Listener, _Input) ->
-    ID = emqx_listeners:identifier(Listener),
-    case emqx_listeners:stop_listener(Listener) of
-        ok ->
-            emqx_ctl:print("Stop ~s listener on ~s successfully.~n", [ID, ListenOn]);
-        {error, Error} ->
-            emqx_ctl:print("Failed to stop ~s listener on ~s, error:~p~n", [ID, ListenOn, Error])
-    end.
+stop_listener(#{} = Listener, _Input) ->
+    %% Discard reason here, reasons are io:format logged to group leader
+    %% in case of emqx_ctl RPC call, it's logged to the remore node.
+    _ = emqx_listeners:stop_listener(Listener),
+    ok.
 
 %%--------------------------------------------------------------------
 %% @doc data Command

From dfa9bbc0c282c0b894ec21ad36d940522f942f2d Mon Sep 17 00:00:00 2001
From: Zaiming Shi <zmstone@gmail.com>
Date: Sun, 14 Feb 2021 20:40:01 +0100
Subject: [PATCH 3/8] refactor(listeners): use emqx_ctl to print cli messages

It's better to keep cli print behaviour consistent.

It also makes tests easier as they can meck emqx_ctl:print to
validate the output.
---
 .../emqx_management/src/emqx_mgmt_cli.erl      | 17 +++++++++++------
 src/emqx_listeners.erl                         | 18 +++++++-----------
 2 files changed, 18 insertions(+), 17 deletions(-)

diff --git a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
index ea5345772..2ec0dce49 100644
--- a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
+++ b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
@@ -554,12 +554,17 @@ listeners(_) ->
                    ]).
 
 stop_listener(false, Input) ->
-    emqx_ctl:print("No such listener ~p~n", [Input]);
-stop_listener(#{} = Listener, _Input) ->
-    %% Discard reason here, reasons are io:format logged to group leader
-    %% in case of emqx_ctl RPC call, it's logged to the remore node.
-    _ = emqx_listeners:stop_listener(Listener),
-    ok.
+    ok = emqx_ctl:print("No such listener ~p~n", [Input]);
+stop_listener(#{listen_on := ListenOn} = Listener, _Input) ->
+    ID = emqx_listeners:identifier(Listener),
+    ListenOnStr = emqx_listeners:format_listen_on(ListenOn),
+    case emqx_listeners:stop_listener(Listener) of
+        ok ->
+            ok = emqx_ctl:print("Stop ~s listener on ~s successfully.~n", [ID, ListenOnStr]);
+        {error, Reason} ->
+            ok = emqx_ctl:print("Failed to stop ~s listener on ~s - ~p~n.",
+                                [ID, ListenOnStr, Reason])
+    end.
 
 %%--------------------------------------------------------------------
 %% @doc data Command
diff --git a/src/emqx_listeners.erl b/src/emqx_listeners.erl
index f7912a7fd..5057922f5 100644
--- a/src/emqx_listeners.erl
+++ b/src/emqx_listeners.erl
@@ -28,7 +28,6 @@
 -export([ start_listener/1
         , start_listener/3
         , stop_listener/1
-        , stop_listener/3
         , restart_listener/1
         , restart_listener/3
         ]).
@@ -37,6 +36,7 @@
         , find_by_listen_on/1
         , find_by_id/1
         , identifier/1
+        , format_listen_on/1
         ]).
 
 -type(listener() :: #{ name := binary()
@@ -76,6 +76,10 @@ identifier(#{proto := Proto, name := Name}) ->
 start() ->
     lists:foreach(fun start_listener/1, emqx:get_env(listeners, [])).
 
+%% @doc Format address:port for logging.
+-spec(format_listen_on(esockd:listen_on()) -> binary()).
+format_listen_on(ListenOn) -> format(ListenOn).
+
 -spec(start_listener(listener()) -> ok).
 start_listener(#{proto := Proto, name := Name, listen_on := ListenOn, opts := Options}) ->
     ID = identifier(Proto, Name),
@@ -171,16 +175,8 @@ stop() ->
     lists:foreach(fun stop_listener/1, emqx:get_env(listeners, [])).
 
 -spec(stop_listener(listener()) -> ok | {error, term()}).
-stop_listener(#{proto := Proto, name := Name, listen_on := ListenOn, opts := Opts}) ->
-    ID = identifier(Proto, Name),
-    StopRet = stop_listener(Proto, ListenOn, Opts),
-    case StopRet of
-        ok -> io:format("Stop ~s listener on ~s successfully.~n", [ID, format(ListenOn)]);
-        {error, Reason} ->
-            io:format(standard_error, "Failed to stop mqtt:~s listener on ~s - ~p~n.",
-                      [ID, format(ListenOn), Reason])
-    end,
-    StopRet.
+stop_listener(#{proto := Proto, listen_on := ListenOn, opts := Opts}) ->
+    stop_listener(Proto, ListenOn, Opts).
 
 -spec(stop_listener(esockd:proto(), esockd:listen_on(), [esockd:option()])
       -> ok | {error, term()}).

From c5a02c729ad9f03cb0170fb1b8f004d2ef422330 Mon Sep 17 00:00:00 2001
From: Zaiming Shi <zmstone@gmail.com>
Date: Sun, 14 Feb 2021 21:30:50 +0100
Subject: [PATCH 4/8] feat(listeners): Add an api to ensure all listeners are
 started

---
 .../emqx_management/src/emqx_mgmt_cli.erl     |  8 +++----
 .../emqx_management/test/emqx_mgmt_SUITE.erl  | 20 +++++++++++++----
 src/emqx_listeners.erl                        | 22 +++++++++++++++++++
 3 files changed, 42 insertions(+), 8 deletions(-)

diff --git a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
index 2ec0dce49..6a0a1bd5a 100644
--- a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
+++ b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
@@ -554,16 +554,16 @@ listeners(_) ->
                    ]).
 
 stop_listener(false, Input) ->
-    ok = emqx_ctl:print("No such listener ~p~n", [Input]);
+    emqx_ctl:print("No such listener ~p~n", [Input]);
 stop_listener(#{listen_on := ListenOn} = Listener, _Input) ->
     ID = emqx_listeners:identifier(Listener),
     ListenOnStr = emqx_listeners:format_listen_on(ListenOn),
     case emqx_listeners:stop_listener(Listener) of
         ok ->
-            ok = emqx_ctl:print("Stop ~s listener on ~s successfully.~n", [ID, ListenOnStr]);
+            emqx_ctl:print("Stop ~s listener on ~s successfully.~n", [ID, ListenOnStr]);
         {error, Reason} ->
-            ok = emqx_ctl:print("Failed to stop ~s listener on ~s - ~p~n.",
-                                [ID, ListenOnStr, Reason])
+            emqx_ctl:print("Failed to stop ~s listener on ~s - ~p~n.",
+                           [ID, ListenOnStr, Reason])
     end.
 
 %%--------------------------------------------------------------------
diff --git a/lib-opensource/emqx_management/test/emqx_mgmt_SUITE.erl b/lib-opensource/emqx_management/test/emqx_mgmt_SUITE.erl
index e41b9ee6a..d8cb91dab 100644
--- a/lib-opensource/emqx_management/test/emqx_mgmt_SUITE.erl
+++ b/lib-opensource/emqx_management/test/emqx_mgmt_SUITE.erl
@@ -49,7 +49,8 @@ groups() ->
         t_broker_cmd,
         t_router_cmd,
         t_subscriptions_cmd,
-        t_listeners_cmd
+        t_listeners_cmd_old,
+        t_listeners_cmd_new
        ]}].
 
 apps() ->
@@ -275,12 +276,23 @@ t_subscriptions_cmd(_) ->
     ?assertEqual(emqx_mgmt_cli:subscriptions(["del", "client", "b/b/c"]), "ok~n"),
     unmock_print().
 
-t_listeners_cmd(_) ->
+t_listeners_cmd_old(_) ->
+    ok = emqx_listeners:ensure_all_started(),
     mock_print(),
     ?assertEqual(emqx_mgmt_cli:listeners([]), ok),
     ?assertEqual(
-       emqx_mgmt_cli:listeners(["stop", "wss", "8084"]),
-       "Stop wss listener on 8084 successfully.\n"
+       "Stop mqtt:wss:external listener on 0.0.0.0:8084 successfully.\n",
+       emqx_mgmt_cli:listeners(["stop", "wss", "8084"])
+      ),
+    unmock_print().
+
+t_listeners_cmd_new(_) ->
+    ok = emqx_listeners:ensure_all_started(),
+    mock_print(),
+    ?assertEqual(emqx_mgmt_cli:listeners([]), ok),
+    ?assertEqual(
+       "Stop mqtt:wss:external listener on 0.0.0.0:8084 successfully.\n",
+       emqx_mgmt_cli:listeners(["stop", "mqtt:wss:external"])
       ),
     unmock_print().
 
diff --git a/src/emqx_listeners.erl b/src/emqx_listeners.erl
index 5057922f5..c1ccae165 100644
--- a/src/emqx_listeners.erl
+++ b/src/emqx_listeners.erl
@@ -21,6 +21,7 @@
 
 %% APIs
 -export([ start/0
+        , ensure_all_started/0
         , restart/0
         , stop/0
         ]).
@@ -76,6 +77,27 @@ identifier(#{proto := Proto, name := Name}) ->
 start() ->
     lists:foreach(fun start_listener/1, emqx:get_env(listeners, [])).
 
+%% @doc Ensure all configured listeners are started.
+%% Raise exception if any of them failed to start.
+-spec(ensure_all_started() -> ok).
+ensure_all_started() ->
+    ensure_all_started(emqx:get_env(listeners, []), []).
+
+ensure_all_started([], []) -> ok;
+ensure_all_started([], Failed) -> error(Failed);
+ensure_all_started([L | Rest], Results) ->
+    #{proto := Proto, listen_on := ListenOn, opts := Options} = L,
+    NewResults =
+        case start_listener(Proto, ListenOn, Options) of
+            {ok, _Pid} ->
+                Results;
+            {error, {already_started, _Pid}} ->
+                Results;
+            {error, Reason} ->
+                [{identifier(L), Reason} | Results]
+        end,
+    ensure_all_started(Rest, NewResults).
+
 %% @doc Format address:port for logging.
 -spec(format_listen_on(esockd:listen_on()) -> binary()).
 format_listen_on(ListenOn) -> format(ListenOn).

From 2fc758c49a6cab3e81096bb5039fb4b4cafdffc1 Mon Sep 17 00:00:00 2001
From: Zaiming Shi <zmstone@gmail.com>
Date: Sun, 14 Feb 2021 23:10:20 +0100
Subject: [PATCH 5/8] fix(test): Pin emqx-ct-helper 1.3.6

---
 rebar.config.erl | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/rebar.config.erl b/rebar.config.erl
index 08b3e2be9..123c39af5 100644
--- a/rebar.config.erl
+++ b/rebar.config.erl
@@ -47,8 +47,8 @@ test_plugins() ->
 
 test_deps() ->
     [ {bbmustache, "1.10.0"}
-     , {emqx_ct_helpers, {git, "https://github.com/emqx/emqx-ct-helpers", {tag, "1.3.5"}}}
-     , meck
+    , {emqx_ct_helpers, {git, "https://github.com/emqx/emqx-ct-helpers", {tag, "1.3.6"}}}
+    , meck
     ].
 
 default_compile_opts() ->

From a93d62ace692ebf2ab293c5beb19b10c7eb5e9a7 Mon Sep 17 00:00:00 2001
From: Zaiming Shi <zmstone@gmail.com>
Date: Mon, 15 Feb 2021 10:34:47 +0100
Subject: [PATCH 6/8] refactor(cli): Print listener ID as table head

---
 .../emqx_management/src/emqx_mgmt_cli.erl      | 18 ++++++++++++++----
 src/emqx_listeners.erl                         |  6 +++---
 2 files changed, 17 insertions(+), 7 deletions(-)

diff --git a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
index 6a0a1bd5a..e3f98b4e4 100644
--- a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
+++ b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
@@ -510,21 +510,23 @@ trace_off(Who, Name) ->
 
 listeners([]) ->
     foreach(fun({{Protocol, ListenOn}, _Pid}) ->
-                Info = [{identifier,     {string, emqx_listeners:find_id_by_listen_on(ListenOn)}},
+                Info = [{listen_on,      {string, emqx_listeners:format_listen_on(ListenOn)}},
                         {acceptors,      esockd:get_acceptors({Protocol, ListenOn})},
                         {max_conns,      esockd:get_max_connections({Protocol, ListenOn})},
                         {current_conn,   esockd:get_current_connections({Protocol, ListenOn})},
                         {shutdown_count, esockd:get_shutdown_count({Protocol, ListenOn})}
                        ],
-                    emqx_ctl:print("listener on ~s:~s~n", [Protocol, esockd:to_string(ListenOn)]),
+                    emqx_ctl:print("~s~n", [listener_identifier(Protocol, ListenOn)]),
                 foreach(fun indent_print/1, Info)
             end, esockd:listeners()),
     foreach(fun({Protocol, Opts}) ->
-                Info = [{acceptors,      maps:get(num_acceptors, proplists:get_value(transport_options, Opts, #{}), 0)},
+                Port = proplists:get_value(port, Opts),
+                Info = [{listen_on,      {string, emqx_listeners:format_listen_on(Port)}},
+                        {acceptors,      maps:get(num_acceptors, proplists:get_value(transport_options, Opts, #{}), 0)},
                         {max_conns,      proplists:get_value(max_connections, Opts)},
                         {current_conn,   proplists:get_value(all_connections, Opts)},
                         {shutdown_count, []}],
-                    emqx_ctl:print("listener on ~s:~p~n", [Protocol, proplists:get_value(port, Opts)]),
+                    emqx_ctl:print("~s~n", [listener_identifier(Protocol, Port)]),
                 foreach(fun indent_print/1, Info)
             end, ranch:info());
 
@@ -724,3 +726,11 @@ indent_print({Key, {string, Val}}) ->
     emqx_ctl:print("  ~-16s: ~s~n", [Key, Val]);
 indent_print({Key, Val}) ->
     emqx_ctl:print("  ~-16s: ~w~n", [Key, Val]).
+
+listener_identifier(Protocol, ListenOn) ->
+    case emqx_listeners:find_id_by_listen_on(ListenOn) of
+        false ->
+            "http" ++ _ = atom_to_list(Protocol); %% assert
+        ID ->
+            ID
+    end.
diff --git a/src/emqx_listeners.erl b/src/emqx_listeners.erl
index c1ccae165..fad205cb1 100644
--- a/src/emqx_listeners.erl
+++ b/src/emqx_listeners.erl
@@ -48,10 +48,10 @@
 
 %% @doc Find listener identifier by listen-on.
 %% Return empty string (binary) if listener is not found in config.
--spec(find_id_by_listen_on(esockd:listen_on()) -> binary()).
+-spec(find_id_by_listen_on(esockd:listen_on()) -> binary() | false).
 find_id_by_listen_on(ListenOn) ->
     case find_by_listen_on(ListenOn) of
-        false -> <<>>;
+        false -> false;
         L -> identifier(L)
     end.
 
@@ -238,7 +238,7 @@ identifier(Proto, Name) when is_atom(Proto) ->
 identifier(Proto, Name) ->
     iolist_to_binary(["mqtt", ":", Proto, ":", Name]).
 
-find_by_listen_on(ListenOn, []) -> error({unknown_listener, ListenOn});
+find_by_listen_on(_ListenOn, []) -> false;
 find_by_listen_on(ListenOn, [#{listen_on := ListenOn} = L | _]) -> L;
 find_by_listen_on(ListenOn, [_ | Rest]) -> find_by_listen_on(ListenOn, Rest).
 

From 65f3a2b685dc3287c67d74f9cf9149d1033eca67 Mon Sep 17 00:00:00 2001
From: Zaiming Shi <zmstone@gmail.com>
Date: Mon, 15 Feb 2021 10:42:17 +0100
Subject: [PATCH 7/8] fix(cli): Stop http listener without port number

---
 lib-opensource/emqx_management/src/emqx_mgmt_cli.erl | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
index e3f98b4e4..047830e5b 100644
--- a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
+++ b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
@@ -530,12 +530,13 @@ listeners([]) ->
                 foreach(fun indent_print/1, Info)
             end, ranch:info());
 
-listeners(["stop",  Name = "http" ++ _N, ListenOn]) ->
+listeners(["stop",  Name = "http" ++ _N | _MaybePort]) ->
+    %% _MaybePort is to be backward compatible, to stop http listener, there is no need for the port number
     case minirest:stop_http(list_to_atom(Name)) of
         ok ->
-            emqx_ctl:print("Stop ~s listener on ~s successfully.~n", [Name, ListenOn]);
+            emqx_ctl:print("Stop ~s listener successfully.~n", [Name]);
         {error, Error} ->
-            emqx_ctl:print("Failed to stop ~s listener on ~s, error:~p~n", [Name, ListenOn, Error])
+            emqx_ctl:print("Failed to stop ~s listener, error:~p~n", [Name, Error])
     end;
 
 listeners(["stop", "mqtt:" ++ _ = Identifier]) ->

From 64cfaf4385d832362121fcaf5c23d2fab1c1688b Mon Sep 17 00:00:00 2001
From: Zaiming Shi <zmstone@gmail.com>
Date: Thu, 18 Feb 2021 13:31:02 +0100
Subject: [PATCH 8/8] improve(print): io:format error message without
 indentation

Avoid squeezing lines to the right.
---
 apps/emqx_coap/src/emqx_coap_server.erl              |  4 ++--
 apps/emqx_exproto/src/emqx_exproto.erl               |  8 ++++----
 apps/emqx_lwm2m/src/emqx_lwm2m_coap_server.erl       |  4 ++--
 apps/emqx_sn/src/emqx_sn_app.erl                     |  4 ++--
 apps/emqx_stomp/src/emqx_stomp.erl                   |  4 ++--
 bin/nodetool                                         |  2 +-
 lib-opensource/emqx_management/src/emqx_mgmt_cli.erl | 10 +++++-----
 src/emqx_listeners.erl                               |  2 +-
 8 files changed, 19 insertions(+), 19 deletions(-)

diff --git a/apps/emqx_coap/src/emqx_coap_server.erl b/apps/emqx_coap/src/emqx_coap_server.erl
index 0d571fac3..4774bd310 100644
--- a/apps/emqx_coap/src/emqx_coap_server.erl
+++ b/apps/emqx_coap/src/emqx_coap_server.erl
@@ -55,7 +55,7 @@ start_listener({Proto, ListenOn, Opts}) ->
             io:format("Start coap:~s listener on ~s successfully.~n",
                       [Proto, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to start coap:~s listener on ~s - ~0p~n!",
+            io:format(standard_error, "Failed to start coap:~s listener on ~s: ~0p~n",
                       [Proto, format(ListenOn), Reason]),
             error(Reason)
     end.
@@ -71,7 +71,7 @@ stop_listener({Proto, ListenOn, _Opts}) ->
         ok -> io:format("Stop coap:~s listener on ~s successfully.~n",
                         [Proto, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to stop coap:~s listener on ~s - ~p~n.",
+            io:format(standard_error, "Failed to stop coap:~s listener on ~s: ~0p~n.",
                       [Proto, format(ListenOn), Reason])
     end,
     Ret.
diff --git a/apps/emqx_exproto/src/emqx_exproto.erl b/apps/emqx_exproto/src/emqx_exproto.erl
index 7d986ecdd..07f56ef1b 100644
--- a/apps/emqx_exproto/src/emqx_exproto.erl
+++ b/apps/emqx_exproto/src/emqx_exproto.erl
@@ -69,7 +69,7 @@ start_connection_handler_instance({_Proto, _LisType, _ListenOn, Opts}) ->
         {ok, _ClientChannelPid} ->
             {_Proto, _LisType, _ListenOn, [{handler, Name} | LisOpts]};
         {error, Reason} ->
-            io:format(standard_error, "Failed to start ~s's connection handler - ~0p~n!",
+            io:format(standard_error, "Failed to start ~s's connection handler: ~0p~n",
                       [Name, Reason]),
             error(Reason)
     end.
@@ -85,7 +85,7 @@ start_server({Name, Port, SSLOptions}) ->
             io:format("Start ~s gRPC server on ~w successfully.~n",
                       [Name, Port]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to start ~s gRPC server on ~w - ~0p~n!",
+            io:format(standard_error, "Failed to start ~s gRPC server on ~w: ~0p~n",
                       [Name, Port, Reason]),
             error({failed_start_server, Reason})
     end.
@@ -101,7 +101,7 @@ start_listener({Proto, LisType, ListenOn, Opts}) ->
             io:format("Start ~s listener on ~s successfully.~n",
                       [Name, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to start ~s listener on ~s - ~0p~n!",
+            io:format(standard_error, "Failed to start ~s listener on ~s: ~0p~n",
                       [Name, format(ListenOn), Reason]),
             error(Reason)
     end.
@@ -132,7 +132,7 @@ stop_listener({Proto, LisType, ListenOn, Opts}) ->
             io:format("Stop ~s listener on ~s successfully.~n",
                       [Name, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to stop ~s listener on ~s - ~p~n.",
+            io:format(standard_error, "Failed to stop ~s listener on ~s: ~0p~n",
                       [Name, format(ListenOn), Reason])
     end,
     StopRet.
diff --git a/apps/emqx_lwm2m/src/emqx_lwm2m_coap_server.erl b/apps/emqx_lwm2m/src/emqx_lwm2m_coap_server.erl
index d366e44e1..47ea6a2ba 100644
--- a/apps/emqx_lwm2m/src/emqx_lwm2m_coap_server.erl
+++ b/apps/emqx_lwm2m/src/emqx_lwm2m_coap_server.erl
@@ -47,7 +47,7 @@ start_listener({Proto, ListenOn, Opts}) ->
             io:format("Start lwm2m:~s listener on ~s successfully.~n",
                       [Proto, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to start lwm2m:~s listener on ~s - ~0p~n!",
+            io:format(standard_error, "Failed to start lwm2m:~s listener on ~s: ~0p~n",
                       [Proto, format(ListenOn), Reason]),
             error(Reason)
     end.
@@ -63,7 +63,7 @@ stop_listener({Proto, ListenOn, _Opts}) ->
         ok -> io:format("Stop lwm2m:~s listener on ~s successfully.~n",
                         [Proto, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to stop lwm2m:~s listener on ~s - ~p~n.",
+            io:format(standard_error, "Failed to stop lwm2m:~s listener on ~s: ~0p~n",
                       [Proto, format(ListenOn), Reason])
     end,
     Ret.
diff --git a/apps/emqx_sn/src/emqx_sn_app.erl b/apps/emqx_sn/src/emqx_sn_app.erl
index 1f59edc9d..8a2f15865 100644
--- a/apps/emqx_sn/src/emqx_sn_app.erl
+++ b/apps/emqx_sn/src/emqx_sn_app.erl
@@ -71,7 +71,7 @@ start_listener({Proto, ListenOn, Options}) ->
         {ok, _} -> io:format("Start mqttsn:~s listener on ~s successfully.~n",
                              [Proto, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to start mqttsn:~s listener on ~s - ~0p~n!",
+            io:format(standard_error, "Failed to start mqttsn:~s listener on ~s: ~0p~n",
                       [Proto, format(ListenOn), Reason]),
             error(Reason)
     end.
@@ -101,7 +101,7 @@ stop_listener({Proto, ListenOn, Opts}) ->
         ok -> io:format("Stop mqttsn:~s listener on ~s successfully.~n",
                         [Proto, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to stop mqttsn:~s listener on ~s - ~p~n.",
+            io:format(standard_error, "Failed to stop mqttsn:~s listener on ~s: ~0p~n",
                       [Proto, format(ListenOn), Reason])
     end,
     StopRet.
diff --git a/apps/emqx_stomp/src/emqx_stomp.erl b/apps/emqx_stomp/src/emqx_stomp.erl
index d7a6db9ee..3d6396d43 100644
--- a/apps/emqx_stomp/src/emqx_stomp.erl
+++ b/apps/emqx_stomp/src/emqx_stomp.erl
@@ -73,7 +73,7 @@ start_listener({Proto, ListenOn, Options}) ->
         {ok, _} -> io:format("Start stomp:~s listener on ~s successfully.~n",
                              [Proto, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to start stomp:~s listener on ~s - ~0p~n!",
+            io:format(standard_error, "Failed to start stomp:~s listener on ~s: ~0p~n",
                       [Proto, format(ListenOn), Reason]),
             error(Reason)
     end.
@@ -102,7 +102,7 @@ stop_listener({Proto, ListenOn, Opts}) ->
         ok -> io:format("Stop stomp:~s listener on ~s successfully.~n",
                         [Proto, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to stop stomp:~s listener on ~s - ~p~n.",
+            io:format(standard_error, "Failed to stop stomp:~s listener on ~s: ~0p~n",
                       [Proto, format(ListenOn), Reason])
     end,
     StopRet.
diff --git a/bin/nodetool b/bin/nodetool
index cf50509f9..1dfc93014 100755
--- a/bin/nodetool
+++ b/bin/nodetool
@@ -47,7 +47,7 @@ main(Args) ->
         {true, pong} ->
             ok;
         {false, pong} ->
-            io:format(standard_error, "Failed to connect to node ~p .\n", [TargetNode]),
+            io:format(standard_error, "Failed to connect to node ~p\n", [TargetNode]),
             halt(1);
         {_, pang} ->
             io:format(standard_error, "Node ~p not responding to pings.\n", [TargetNode]),
diff --git a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
index 047830e5b..db05cafb8 100644
--- a/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
+++ b/lib-opensource/emqx_management/src/emqx_mgmt_cli.erl
@@ -157,7 +157,7 @@ cluster(["join", SNode]) ->
         ignore ->
             emqx_ctl:print("Ignore.~n");
         {error, Error} ->
-            emqx_ctl:print("Failed to join the cluster: ~p~n", [Error])
+            emqx_ctl:print("Failed to join the cluster: ~0p~n", [Error])
     end;
 
 cluster(["leave"]) ->
@@ -166,7 +166,7 @@ cluster(["leave"]) ->
             emqx_ctl:print("Leave the cluster successfully.~n"),
             cluster(["status"]);
         {error, Error} ->
-            emqx_ctl:print("Failed to leave the cluster: ~p~n", [Error])
+            emqx_ctl:print("Failed to leave the cluster: ~0p~n", [Error])
     end;
 
 cluster(["force-leave", SNode]) ->
@@ -177,7 +177,7 @@ cluster(["force-leave", SNode]) ->
         ignore ->
             emqx_ctl:print("Ignore.~n");
         {error, Error} ->
-            emqx_ctl:print("Failed to remove the node from cluster: ~p~n", [Error])
+            emqx_ctl:print("Failed to remove the node from cluster: ~0p~n", [Error])
     end;
 
 cluster(["status"]) ->
@@ -536,7 +536,7 @@ listeners(["stop",  Name = "http" ++ _N | _MaybePort]) ->
         ok ->
             emqx_ctl:print("Stop ~s listener successfully.~n", [Name]);
         {error, Error} ->
-            emqx_ctl:print("Failed to stop ~s listener, error:~p~n", [Name, Error])
+            emqx_ctl:print("Failed to stop ~s listener: ~0p~n", [Name, Error])
     end;
 
 listeners(["stop", "mqtt:" ++ _ = Identifier]) ->
@@ -565,7 +565,7 @@ stop_listener(#{listen_on := ListenOn} = Listener, _Input) ->
         ok ->
             emqx_ctl:print("Stop ~s listener on ~s successfully.~n", [ID, ListenOnStr]);
         {error, Reason} ->
-            emqx_ctl:print("Failed to stop ~s listener on ~s - ~p~n.",
+            emqx_ctl:print("Failed to stop ~s listener on ~s: ~0p~n",
                            [ID, ListenOnStr, Reason])
     end.
 
diff --git a/src/emqx_listeners.erl b/src/emqx_listeners.erl
index fad205cb1..eadc93b97 100644
--- a/src/emqx_listeners.erl
+++ b/src/emqx_listeners.erl
@@ -109,7 +109,7 @@ start_listener(#{proto := Proto, name := Name, listen_on := ListenOn, opts := Op
         {ok, _} -> io:format("Start ~s listener on ~s successfully.~n",
                              [ID, format(ListenOn)]);
         {error, Reason} ->
-            io:format(standard_error, "Failed to start mqtt listener ~s on ~s - ~0p~n!",
+            io:format(standard_error, "Failed to start mqtt listener ~s on ~s: ~0p~n",
                       [ID, format(ListenOn), Reason]),
             error(Reason)
     end.