rabbit_networking: Support systems with distinct listeners for IPv4 and IPv6

[Why]
If the system uses distinct listeners for IPv4 and IPv6, the functions
listing or stopping listeners would crash because they assumed that the
system would listen implicitly to IPv4 when starting an IPv6 listener.
This is a common Linuxism.

[How]
The code now understands that a given protocol can have several
associated listeners.

The name and return value of several exported functions changed to
reflect that they work on several elements or return a list of elements.
Therefore it is a breaking change for plugins relying on these
functions.

Author: dumbbell

deps/rabbit/src/rabbit_networking.erl

@@ -33,8 +33,8 @@
          close_all_user_connections/2,
          force_connection_event_refresh/1, force_non_amqp_connection_event_refresh/1,
          handshake/2, handshake/3, tcp_host/1,
-         ranch_ref/1, ranch_ref/2, ranch_ref_of_protocol/1, ranch_ref_to_protocol/1,
-         listener_of_protocol/1, stop_ranch_listener_of_protocol/1,
+         ranch_ref/1, ranch_ref/2, ranch_refs_of_protocol/1, ranch_ref_to_protocol/1,
+         listeners_of_protocol/1, stop_ranch_listeners_of_protocol/1,
          list_local_connections_of_protocol/1]).
 
 %% Used by TCP-based transports, e.g. STOMP adapter
@@ -229,9 +229,9 @@ ranch_ref(undefined) ->
 ranch_ref(IPAddress, Port) ->
     {acceptor, IPAddress, Port}.
 
--spec ranch_ref_of_protocol(atom()) -> ranch:ref() | undefined.
-ranch_ref_of_protocol(Protocol) ->
-    ranch_ref(listener_of_protocol(Protocol)).
+-spec ranch_refs_of_protocol(atom()) -> [ranch:ref()].
+ranch_refs_of_protocol(Protocol) ->
+    [ranch_ref(Listener) || Listener <- listeners_of_protocol(Protocol)].
 
 -spec ranch_ref_to_protocol(ranch:ref()) -> atom() | undefined.
 ranch_ref_to_protocol({acceptor, IPAddress, Port}) ->
@@ -248,32 +248,32 @@ ranch_ref_to_protocol({acceptor, IPAddress, Port}) ->
 ranch_ref_to_protocol(_) ->
     undefined.
 
--spec listener_of_protocol(atom()) -> #listener{}.
-listener_of_protocol(Protocol) ->
+-spec listeners_of_protocol(atom()) -> [#listener{}].
+listeners_of_protocol(Protocol) ->
     MatchSpec = #listener{
                    protocol = Protocol,
                    _ = '_'
                   },
-    case ets:match_object(?ETS_TABLE, MatchSpec) of
-        []    -> undefined;
-        [Row] -> Row
-    end.
+    ets:match_object(?ETS_TABLE, MatchSpec).
 
--spec stop_ranch_listener_of_protocol(atom()) -> ok | {error, not_found}.
-stop_ranch_listener_of_protocol(Protocol) ->
-    case ranch_ref_of_protocol(Protocol) of
-        undefined -> ok;
-        Ref       ->
-            ?LOG_DEBUG("Stopping Ranch listener for protocol ~ts", [Protocol]),
-            ranch:stop_listener(Ref)
+-spec stop_ranch_listeners_of_protocol(atom()) -> ok.
+stop_ranch_listeners_of_protocol(Protocol) ->
+    case ranch_refs_of_protocol(Protocol) of
+        [] ->
+            ok;
+        Refs ->
+            ?LOG_DEBUG("Stopping Ranch listeners for protocol ~ts", [Protocol]),
+            lists:foreach(fun ranch:stop_listener/1, Refs)
     end.
 
 -spec list_local_connections_of_protocol(atom()) -> [pid()].
 list_local_connections_of_protocol(Protocol) ->
-    case ranch_ref_of_protocol(Protocol) of
-        undefined   -> [];
-        AcceptorRef -> ranch:procs(AcceptorRef, connections)
-    end.
+    Refs = ranch_refs_of_protocol(Protocol),
+    lists:flatten(
+      lists:map(
+        fun(Ref) ->
+                ranch:procs(Ref, connections)
+        end, Refs)).
 
 -spec start_tcp_listener(
         listener_config(), integer()) -> 'ok' | {'error', term()}.

deps/rabbitmq_management/src/rabbit_mgmt_wm_health_check_below_node_connection_limit.erl

@@ -55,10 +55,9 @@ to_json(ReqData, Context) ->
     end.
 
 protocol_connection_count(Protocol) ->
-    case rabbit_networking:ranch_ref_of_protocol(Protocol) of
-        undefined ->
-            0;
-        RanchRef ->
-            #{active_connections := Count} = ranch:info(RanchRef),
-            Count
-    end.
+    Refs = rabbit_networking:ranch_refs_of_protocol(Protocol),
+    lists:foldl(
+      fun(Ref, Acc) ->
+              #{active_connections := Count} = ranch:info(Ref),
+              Acc + Count
+      end, 0, Refs).

deps/rabbitmq_management/src/rabbit_mgmt_wm_health_check_ready_to_serve_clients.erl

@@ -49,10 +49,9 @@ to_json(ReqData, Context) ->
 check() ->
     case rabbit:is_serving() of
         true ->
-            RanchRefs0 = [
-                rabbit_networking:ranch_ref_of_protocol(amqp),
-                rabbit_networking:ranch_ref_of_protocol('amqp/ssl')
-            ],
+            RanchRefs0 = (
+              rabbit_networking:ranch_refs_of_protocol(amqp) ++
+              rabbit_networking:ranch_refs_of_protocol('amqp/ssl')),
             RanchRefs = [R || R <- RanchRefs0, R =/= undefined],
             case RanchRefs of
                 [_ | _] ->

deps/rabbitmq_mqtt/src/rabbit_mqtt_sup.erl

@@ -63,8 +63,8 @@ init([{Listeners, SslListeners0}]) ->
 
 -spec stop_listeners() -> ok.
 stop_listeners() ->
-    _ = rabbit_networking:stop_ranch_listener_of_protocol(?MQTT_TCP_PROTOCOL),
-    _ = rabbit_networking:stop_ranch_listener_of_protocol(?MQTT_TLS_PROTOCOL),
+    rabbit_networking:stop_ranch_listeners_of_protocol(?MQTT_TCP_PROTOCOL),
+    rabbit_networking:stop_ranch_listeners_of_protocol(?MQTT_TLS_PROTOCOL),
     ok.
 
 %%

deps/rabbitmq_stomp/src/rabbit_stomp_sup.erl

@@ -42,8 +42,8 @@ init([{Listeners, SslListeners0}, Configuration]) ->
                           SslListeners)}}.
 
 stop_listeners() ->
-    _ = rabbit_networking:stop_ranch_listener_of_protocol(?TCP_PROTOCOL),
-    _ = rabbit_networking:stop_ranch_listener_of_protocol(?TLS_PROTOCOL),
+    rabbit_networking:stop_ranch_listeners_of_protocol(?TCP_PROTOCOL),
+    rabbit_networking:stop_ranch_listeners_of_protocol(?TLS_PROTOCOL),
     ok.
 
 %%

deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_app.erl

@@ -45,8 +45,8 @@ prep_stop(State) ->
 
 -spec stop(_) -> ok.
 stop(_State) ->
-    _ = rabbit_networking:stop_ranch_listener_of_protocol(?TCP_PROTOCOL),
-    _ = rabbit_networking:stop_ranch_listener_of_protocol(?TLS_PROTOCOL),
+    rabbit_networking:stop_ranch_listeners_of_protocol(?TCP_PROTOCOL),
+    rabbit_networking:stop_ranch_listeners_of_protocol(?TLS_PROTOCOL),
     ok.
 
 init([]) -> {ok, {{one_for_one, 1, 5}, []}}.

deps/rabbitmq_web_stomp/src/rabbit_web_stomp_listener.erl

@@ -54,8 +54,8 @@ init() ->
     ok.
 
 stop(State) ->
-    _ = rabbit_networking:stop_ranch_listener_of_protocol(?TCP_PROTOCOL),
-    _ = rabbit_networking:stop_ranch_listener_of_protocol(?TLS_PROTOCOL),
+    rabbit_networking:stop_ranch_listeners_of_protocol(?TCP_PROTOCOL),
+    rabbit_networking:stop_ranch_listeners_of_protocol(?TLS_PROTOCOL),
     State.
 
 -spec list_connections() -> [pid()].
@@ -77,11 +77,12 @@ close_all_client_connections(Reason) ->
 %%
 
 connection_pids_of_protocol(Protocol) ->
-    case rabbit_networking:ranch_ref_of_protocol(Protocol) of
-        undefined   -> [];
-        AcceptorRef ->
-            lists:map(fun cowboy_ws_connection_pid/1, ranch:procs(AcceptorRef, connections))
-    end.
+    Refs = rabbit_networking:ranch_refs_of_protocol(Protocol),
+    lists:foldl(
+      fun(Ref, Acc) ->
+              Procs = lists:map(fun cowboy_ws_connection_pid/1, ranch:procs(Ref, connections)),
+              Acc ++ Procs
+      end, [], Refs).
 
 -spec cowboy_ws_connection_pid(pid()) -> pid().
 cowboy_ws_connection_pid(RanchConnPid) ->