Merge pull request #14317 from rabbitmq/channel-reuse

Fix channel number reuse

Author: dumbbell

deps/rabbit/src/rabbit_amqp_reader.erl

@@ -17,7 +17,8 @@
 -export([init/1,
          info/2,
          mainloop/2,
-         set_credential/2]).
+         set_credential/2,
+         notify_session_ending/3]).
 
 -export([system_continue/3,
          system_terminate/4,
@@ -79,6 +80,11 @@ set_credential(Pid, Credential) ->
     Pid ! {set_credential, Credential},
     ok.
 
+-spec notify_session_ending(pid(), pid(), non_neg_integer()) -> ok.
+notify_session_ending(ConnPid, SessionPid, ChannelNum) ->
+    ConnPid ! {session_ending, SessionPid, ChannelNum},
+    ok.
+
 %%--------------------------------------------------------------------------
 
 recvloop(Deb, State = #v1{pending_recv = true}) ->
@@ -233,6 +239,8 @@ handle_other({set_credential, Cred}, State) ->
 handle_other(credential_expired, State) ->
     Error = error_frame(?V_1_0_AMQP_ERROR_UNAUTHORIZED_ACCESS, "credential expired", []),
     handle_exception(State, 0, Error);
+handle_other({session_ending, SessionPid, ChannelNum}, State) ->
+    untrack_channel(ChannelNum, SessionPid, State);
 handle_other(Other, _State) ->
     %% internal error -> something worth dying for
     exit({unexpected_message, Other}).

deps/rabbit/src/rabbit_amqp_session.erl

@@ -638,10 +638,11 @@ log_error_and_close_session(
   Error, State = #state{cfg = #cfg{reader_pid = ReaderPid,
                                    writer_pid = WriterPid,
                                    channel_num = Ch}}) ->
-    End = #'v1_0.end'{error = Error},
     ?LOG_WARNING("Closing session for connection ~p: ~tp",
                  [ReaderPid, Error]),
-    ok = rabbit_amqp_writer:send_command_sync(WriterPid, Ch, End),
+    rabbit_amqp_reader:notify_session_ending(ReaderPid, self(), Ch),
+    ok = rabbit_amqp_writer:send_command_sync(
+           WriterPid, Ch, #'v1_0.end'{error = Error}),
     {stop, {shutdown, Error}, State}.
 
 %% Batch confirms / rejects to publishers.
@@ -1178,9 +1179,11 @@ handle_frame(Detach = #'v1_0.detach'{handle = ?UINT(HandleInt)},
     reply_frames(Reply, State);
 
 handle_frame(#'v1_0.end'{},
-             State0 = #state{cfg = #cfg{writer_pid = WriterPid,
+             State0 = #state{cfg = #cfg{reader_pid = ReaderPid,
+                                        writer_pid = WriterPid,
                                         channel_num = Ch}}) ->
     State = send_delivery_state_changes(State0),
+    rabbit_amqp_reader:notify_session_ending(ReaderPid, self(), Ch),
     ok = try rabbit_amqp_writer:send_command_sync(WriterPid, Ch, #'v1_0.end'{})
          catch exit:{Reason, {gen_server, call, _ArgList}}
                  when Reason =:= shutdown orelse