Close connections using HTTP API

Greatly simplify `RabbitMQCtl` class.

Use `rabbitmq:management` docker image on Ubuntu

Author: lukebakken

.ci/ubuntu/gha-setup.sh

@@ -63,12 +63,13 @@ function start_rabbitmq
         --hostname "$rabbitmq_docker_name" \
         --publish 5671:5671 \
         --publish 5672:5672 \
+        --publish 15672:15672 \
         --network "$docker_network_name" \
         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/enabled_plugins:/etc/rabbitmq/enabled_plugins" \
         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro" \
         --volume "$GITHUB_WORKSPACE/.ci/certs:/etc/rabbitmq/certs:ro" \
         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/log:/var/log/rabbitmq" \
-        rabbitmq:latest
+        rabbitmq:management
 }
 
 function wait_rabbitmq

projects/Test/Common/Common.csproj

@@ -26,6 +26,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <PackageReference Include="EasyNetQ.Management.Client" Version="2.0.0" />
     <PackageReference Include="xunit" Version="2.7.0" />
     <PackageReference Include="xunit.abstractions" Version="2.0.3" />
   </ItemGroup>

projects/Test/Common/RabbitMQCtl.cs

@@ -32,54 +32,20 @@
 using System;
 using System.Diagnostics;
 using System.IO;
-using System.Text.RegularExpressions;
 using System.Threading.Tasks;
-using RabbitMQ.Client;
 using Xunit.Abstractions;
 
 namespace Test
 {
     public class RabbitMQCtl
     {
-        private static readonly char[] newLine = new char[] { '\n' };
-        // NOTE: \r?
-        // https://learn.microsoft.com/en-us/dotnet/standard/base-types/regular-expression-options#multiline-mode
-        private static readonly Regex s_getConnectionProperties =
-            new Regex(@"^(?<pid><[^>]*>)\s\[.*""connection_name"",""(?<connection_name>[^""]*)"".*\]\r?$", RegexOptions.Multiline | RegexOptions.Compiled);
-
         private readonly ITestOutputHelper _output;
 
         public RabbitMQCtl(ITestOutputHelper output)
         {
             _output = output;
         }
 
-        public async Task CloseConnectionAsync(IConnection conn)
-        {
-            string pid = await GetConnectionPidAsync(conn.ClientProvidedName);
-            await CloseConnectionAsync(pid);
-        }
-
-        public Task AddUserAsync(string username, string password)
-        {
-            return ExecRabbitMQCtlAsync($"add_user {username} {password}");
-        }
-
-        public Task ChangePasswordAsync(string username, string password)
-        {
-            return ExecRabbitMQCtlAsync($"change_password {username} {password}");
-        }
-
-        public Task SetPermissionsAsync(string username, string conf, string write, string read)
-        {
-            return ExecRabbitMQCtlAsync($"set_permissions {username} \"{conf}\" \"{write}\" \"${read}\" ");
-        }
-
-        public Task DeleteUserAsync(string username)
-        {
-            return ExecRabbitMQCtlAsync($"delete_user {username}");
-        }
-
         public async Task<string> ExecRabbitMQCtlAsync(string args)
         {
             try
@@ -139,28 +105,6 @@ private static ProcessStartInfo GetRabbitMqCtlStartInfo(string args)
             return CreateProcessStartInfo(path, args);
         }
 
-        private async Task<string> GetConnectionPidAsync(string connectionName)
-        {
-            string stdout = await ExecRabbitMQCtlAsync("list_connections --silent pid client_properties");
-            Match match = s_getConnectionProperties.Match(stdout);
-            while (match.Success)
-            {
-                if (match.Groups["connection_name"].Value == connectionName)
-                {
-                    return match.Groups["pid"].Value;
-                }
-
-                match = match.NextMatch();
-            }
-
-            throw new Exception($"No connection found with name: {connectionName}");
-        }
-
-        private Task CloseConnectionAsync(string pid)
-        {
-            return ExecRabbitMQCtlAsync($"close_connection \"{pid}\" \"Closed via rabbitmqctl\"");
-        }
-
         private static ProcessStartInfo CreateProcessStartInfo(string cmd, string arguments, string workDirectory = null)
         {
             return new ProcessStartInfo

projects/Test/Common/TestConnectionRecoveryBase.cs

@@ -170,7 +170,7 @@ internal async Task<AutorecoveringConnection> CreateAutorecoveringConnectionWith
 
         protected Task CloseConnectionAsync(IConnection conn)
         {
-            return _rabbitMQCtl.CloseConnectionAsync(conn);
+            return Util.CloseConnectionAsync(conn);
         }
 
         protected Task CloseAndWaitForRecoveryAsync()

projects/Test/Common/Util.cs

@@ -1,14 +1,23 @@
 using System;
+using System.Collections.Generic;
 using System.Globalization;
+using System.Linq;
+using System.Threading.Tasks;
+using EasyNetQ.Management.Client;
+using RabbitMQ.Client;
 
 namespace Test
 {
     public static class Util
     {
+        private static readonly TimeSpan s_closeConnectionDelay = TimeSpan.FromSeconds(2);
+        private static readonly ManagementClient s_managementClient;
         private static readonly bool s_isWindows = false;
 
         static Util()
         {
+            var managementUri = new Uri("http://localhost:15672");
+            s_managementClient = new ManagementClient(managementUri, "guest", "guest");
             s_isWindows = InitIsWindows();
         }
 
@@ -33,5 +42,53 @@ private static bool InitIsWindows()
 
             return false;
         }
+
+        public static async Task CloseConnectionAsync(IConnection conn)
+        {
+            ushort tries = 10;
+            EasyNetQ.Management.Client.Model.Connection connectionToClose = null;
+            do
+            {
+                IReadOnlyList<EasyNetQ.Management.Client.Model.Connection> connections;
+                try
+                {
+                    do
+                    {
+                        await Task.Delay(s_closeConnectionDelay);
+                        connections = await s_managementClient.GetConnectionsAsync();
+                    } while (connections.Count == 0);
+
+                    connectionToClose = connections.Where(c0 =>
+                        string.Equals((string)c0.ClientProperties["connection_name"], conn.ClientProvidedName,
+                        StringComparison.InvariantCultureIgnoreCase)).FirstOrDefault();
+
+                    if (connectionToClose == null)
+                    {
+                        tries--;
+                    }
+                    else
+                    {
+                        break;
+                    }
+                }
+                catch (ArgumentNullException)
+                {
+                    // Sometimes we see this in GitHub CI
+                    tries--;
+                }
+            } while (tries > 0);
+
+            if (tries == 0)
+            {
+                throw new InvalidOperationException($"Could not delete connection: '{conn.ClientProvidedName}'");
+            }
+
+            if (connectionToClose == null)
+            {
+                throw new InvalidOperationException($"connectionToClose should not be null here");
+            }
+
+            await s_managementClient.CloseConnectionAsync(connectionToClose);
+        }
     }
 }