Now we have the reader, heartbeaters and queue_collector all under the same connection_supervisor, with one connection_supervisor per connection. Tests pass. The framing_channel, channel, writer and limiter are all still just linked as before, without supervision

1 files changed, 12 insertions, 8 deletions

diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index 68ffc98a..6502c6d1 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -114,7 +114,7 @@ start() ->
                {rabbit_tcp_client_sup,
                 {tcp_client_sup, start_link,
                  [{local, rabbit_tcp_client_sup},
-                  {rabbit_reader,start_link,[]}]},
+                  {rabbit_connection_sup,start_link,[]}]},
                 transient, infinity, supervisor, [tcp_client_sup]}),
     ok.
 
@@ -201,9 +201,12 @@ on_node_down(Node) ->
 
 start_client(Sock, SockTransform) ->
     {ok, Child} = supervisor:start_child(rabbit_tcp_client_sup, []),
-    ok = rabbit_net:controlling_process(Sock, Child),
-    Child ! {go, Sock, SockTransform},
-    Child.
+    hd([begin
+            ok = rabbit_net:controlling_process(Sock, Reader),
+            Reader ! {go, Sock, SockTransform},
+            Reader
+        end || {reader, Reader, worker, [rabbit_reader]}
+                   <- supervisor:which_children(Child)]).
 
 start_client(Sock) ->
     start_client(Sock, fun (S) -> {ok, S} end).
@@ -226,8 +229,10 @@ start_ssl_client(SslOpts, Sock) ->
       end).
 
 connections() ->
-    [Pid || {_, Pid, _, _} <- supervisor:which_children(
-                                rabbit_tcp_client_sup)].
+    [Pid || {_, ConnSup, supervisor, _}
+                <- supervisor:which_children(rabbit_tcp_client_sup),
+            {reader, Pid, worker, [rabbit_reader]}
+                <- supervisor:which_children(ConnSup)].
 
 connection_info_keys() -> rabbit_reader:info_keys().
 
@@ -238,8 +243,7 @@ connection_info_all() -> cmap(fun (Q) -> connection_info(Q) end).
 connection_info_all(Items) -> cmap(fun (Q) -> connection_info(Q, Items) end).
 
 close_connection(Pid, Explanation) ->
-    case lists:any(fun ({_, ChildPid, _, _}) -> ChildPid =:= Pid end,
-                   supervisor:which_children(rabbit_tcp_client_sup)) of
+    case lists:member(Pid, connections()) of
         true  -> rabbit_reader:shutdown(Pid, Explanation);
         false -> throw({error, {not_a_connection_pid, Pid}})
     end.