tried to fix unit test

Author: kerryjiang

tests/Test/MySQLFixture.cs

@@ -13,51 +13,63 @@ public class MySQLFixture : IDisposable
         private const string _username = "root";
         private const string _password = "root";
 
-        private readonly MySqlConnection _connection;
-
         public IReplicationClient Client { get; private set; }
 
+        private SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
+
         public static MySQLFixture Instance { get; } = new MySQLFixture();
 
         private MySQLFixture()
         {
-            _connection = new MySqlConnection($"Server={_host};Database=garden;Uid={_username};Pwd={_password};");
             Client = new ReplicationClient();
             ConnectAsync().Wait();
         }
 
         private async Task ConnectAsync()
         {
-            await _connection.OpenAsync();
             await Client.ConnectAsync(_host, _username, _password, 1);
         }
 
+        private MySqlConnection GetConnection()
+        {
+            var connection = new MySqlConnection($"Server={_host};Database=garden;Uid={_username};Pwd={_password};");
+            connection.OpenAsync().Wait();
+            return connection;
+        }
+
         public MySqlCommand CreateCommand()
         {
-            return _connection.CreateCommand();
+            return GetConnection().CreateCommand();
         }
 
         public async Task<TLogEvent> ReceiveAsync<TLogEvent>(CancellationToken cancellationToken = default)
             where TLogEvent : LogEvent
         {
-            while (!cancellationToken.IsCancellationRequested)
-            {
-                var logEvent = await Client.ReceiveAsync();
+            await _semaphore.WaitAsync(cancellationToken);
 
-                if (logEvent is TLogEvent requiredLogEvent)
+            try
+            {
+                while (!cancellationToken.IsCancellationRequested)
                 {
-                    return requiredLogEvent;
+                    var logEvent = await Client.ReceiveAsync();
+
+                    if (logEvent is TLogEvent requiredLogEvent)
+                    {
+                        return requiredLogEvent;
+                    }
                 }
             }
+            finally
+            {
+                _semaphore.Release();
+            }
 
             return default;
         }
 
         public void Dispose()
         {
             Client?.CloseAsync().AsTask().Wait();
-            _connection?.CloseAsync().Wait();
-            _connection?.Dispose();
         }
     }
 }