rabbit_networking: Support systems with distinct listeners for IPv4 and IPv6

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
@@ -197,8 +197,7 @@ tcp_listener_spec(NamePrefix, {IPAddress, Port, Family}, SocketOpts,
      transient, infinity, supervisor, [tcp_listener_sup]}.
 
 -spec ranch_ref(#listener{} | [{atom(), any()}] | 'undefined') -> ranch:ref() | undefined.
-ranch_ref(#listener{port = Port}) ->
-    [{IPAddress, Port, _Family} | _] = tcp_listener_addresses(Port),
+ranch_ref(#listener{ip_address = IPAddress, port = Port}) ->
     {acceptor, IPAddress, Port};
 ranch_ref(Listener) when is_list(Listener) ->
     Port = rabbit_misc:pget(port, Listener),
@@ -230,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}) ->
@@ -249,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) ->