worker_pool

Author: dumbbell

deps/rabbit/src/rabbit_amqqueue_process.erl

@@ -354,17 +354,11 @@ terminate_delete(EmitStats, Reason0, ReplyTo,
         end,
         case ReplyTo of
             _ when ReplyTo =/= none ->
-                %% This try-catch block transforms throws to errors since
-                %% throws are not logged. When mnesia is removed this `try`
-                %% can be removed: Khepri returns errors as error tuples
-                %% instead.
-                Reply = try rabbit_amqqueue:internal_delete(Q, ActingUser, Reason0) of
+                Reply = case delete_queue_record(Q, ActingUser, Reason0) of
                             ok ->
                                 {ok, Len};
                             {error, _} = Err ->
                                 Err
-                        catch
-                            {error, ReasonE} -> error(ReasonE)
                         end,
                 send_reply(ReplyTo, Reply);
             none ->
@@ -385,28 +379,38 @@ terminate_delete(EmitStats, Reason0, ReplyTo,
                 %% with the `node_up' message from `rabbit_node_monitor'. This
                 %% calls `rabbit_amqqueue:on_node_up/1' which will delete any
                 %% transient queues.
-                _ = infinite_internal_delete(Q, ActingUser, Reason0),
+                %%
+                %% This infinite delete attempts loop is executed in a
+                %% separate process to let this queue process exits. This way,
+                %% connections will be notified that the queue process is
+                %% gone.
+                worker_pool:submit_async(
+                  fun() ->
+                          _ = infinite_internal_delete(Q, ActingUser, Reason0)
+                  end),
                 ok
         end,
         BQS1
     end.
 
 infinite_internal_delete(Q, ActingUser, Reason) ->
-    %% This try-catch block transforms throws to errors since throws are not
-    %% logged. When mnesia is removed this `try` can be removed: Khepri returns
-    %% errors as error tuples instead.
-    Ret = try
-              rabbit_amqqueue:internal_delete(Q, ActingUser, Reason)
-          catch
-              {error, ReasonE} -> error(ReasonE)
-          end,
-    case Ret of
+    case delete_queue_record(Q, ActingUser, Reason) of
         {error, timeout} ->
             infinite_internal_delete(Q, ActingUser, Reason);
-        _ ->
+        Ret ->
             Ret
     end.
 
+delete_queue_record(Q, ActingUser, Reason) ->
+    %% This try-catch block transforms throws to errors since throws are not
+    %% logged. When mnesia is removed this `try` can be removed: Khepri returns
+    %% errors as error tuples instead.
+    try
+        rabbit_amqqueue:internal_delete(Q, ActingUser, Reason)
+    catch
+        {error, ReasonE} -> error(ReasonE)
+    end.
+
 terminated_by({terminated_by, auto_delete}) ->
     ?INTERNAL_USER;
 terminated_by({terminated_by, ActingUser}) ->