fix: bug on cleaner

CacheManagerClear.Rabbit/CachePublisher.cs

@@ -35,8 +35,14 @@ public CachePublisher(IConnection connection, string exchange)
 	/// </summary>
 	public async Task StopAsync()
 	{
+		if (_disposed)
+		{
+			return;
+		}
+
 		await _connection.DisposeAsync().ConfigureAwait(false);
 		await _connection.CloseAsync().ConfigureAwait(false);
+		_disposed = true;
 	}
 
 	/// <summary>

CacheManagerClear.Rabbit/CacheSubscriber.cs

@@ -23,7 +23,7 @@ public class CacheSubscriber : ICacheSubscriber
 	/// <param name="cacheManager">Cache manager instance</param>
 	/// <param name="exchange">RabbitMQ exchange name</param>
 	/// <param name="queue">RabbitMQ queue name</param>
-	public CacheSubscriber(IConnection connection, IEasyCacheManager cacheManager, string exchange, string queue)
+	public CacheSubscriber(IConnection connection, string exchange, string queue, IEasyCacheManager cacheManager)
 	{
 		_connection = connection ?? throw new ArgumentNullException(nameof(connection));
 		_cacheManager = cacheManager ?? throw new ArgumentNullException(nameof(cacheManager));
@@ -36,41 +36,45 @@ public CacheSubscriber(IConnection connection, IEasyCacheManager cacheManager, s
 	/// </summary>
 	public async Task SubscribeAsync(CancellationToken cancellationToken)
 	{
-		using var channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
-
-		await channel.ExchangeDeclareAsync(exchange: _exchange, type: ExchangeType.Fanout, durable: true, cancellationToken: cancellationToken).ConfigureAwait(false);
-		_ = await channel.QueueDeclareAsync(queue: _queue, durable: true, exclusive: false, autoDelete: false, cancellationToken: cancellationToken).ConfigureAwait(false);
-		await channel.QueueBindAsync(queue: _queue, exchange: _exchange, routingKey: string.Empty, cancellationToken: cancellationToken).ConfigureAwait(false);
+		try
+		{
+			using var channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
 
-		var consumer = new AsyncEventingBasicConsumer(channel);
+			await channel.ExchangeDeclareAsync(exchange: _exchange, type: ExchangeType.Fanout, durable: true, cancellationToken: cancellationToken).ConfigureAwait(false);
+			_ = await channel.QueueDeclareAsync(queue: _queue, durable: true, exclusive: false, autoDelete: false, cancellationToken: cancellationToken).ConfigureAwait(false);
+			await channel.QueueBindAsync(queue: _queue, exchange: _exchange, routingKey: string.Empty, cancellationToken: cancellationToken).ConfigureAwait(false);
 
-		consumer.ReceivedAsync += async (_, ea) =>
-		{
-			var message = Encoding.UTF8.GetString(ea.Body.ToArray());
+			var consumer = new AsyncEventingBasicConsumer(channel);
 
-			if (message == "*")
-			{
-				await _cacheManager.ClearAllCacheAsync().ConfigureAwait(false);
-			}
-			else
-			{
-				await _cacheManager.ClearCacheAsync(message).ConfigureAwait(false);
-			}
+			consumer.ReceivedAsync += async (_, ea) => await ProcessAsync(ea, channel, cancellationToken).ConfigureAwait(false);
 
-			await channel.BasicAckAsync(ea.DeliveryTag, false, cancellationToken).ConfigureAwait(false);
-		};
+			// Start consuming messages
+			_ = await channel.BasicConsumeAsync(queue: _queue, autoAck: false, consumer: consumer, cancellationToken: cancellationToken).ConfigureAwait(false);
+		}
+		catch (TaskCanceledException)
+		{
+		}
+		catch (Exception e)
+		{
+			Console.WriteLine(e);
 
-		// Start consuming messages
-		_ = await channel.BasicConsumeAsync(queue: _queue, autoAck: false, consumer: consumer, cancellationToken: cancellationToken).ConfigureAwait(false);
+			await Task.Delay(100, cancellationToken).ConfigureAwait(false);
+		}
 	}
 
 	/// <summary>
 	/// Stops the Kafka subscription process.
 	/// </summary>
 	public async Task StopAsync()
 	{
+		if (_disposed)
+		{
+			return;
+		}
+
 		await _connection.DisposeAsync().ConfigureAwait(false);
 		await _connection.CloseAsync().ConfigureAwait(false);
+		_disposed = true;
 	}
 
 	/// <summary>
@@ -88,4 +92,32 @@ public async ValueTask DisposeAsync()
 
 		_disposed = true;
 	}
+
+	private async Task ProcessAsync(BasicDeliverEventArgs ea, IChannel channel, CancellationToken cancellationToken)
+	{
+		try
+		{
+			var key = Encoding.UTF8.GetString(ea.Body.ToArray());
+
+			if (key.Equals(StaticData.ClearAllKey, StringComparison.Ordinal))
+			{
+				await _cacheManager.ClearAllCacheAsync().ConfigureAwait(false);
+			}
+			else
+			{
+				await _cacheManager.ClearCacheAsync(key).ConfigureAwait(false);
+			}
+
+			await channel.BasicAckAsync(ea.DeliveryTag, false, cancellationToken).ConfigureAwait(false);
+		}
+		catch (TaskCanceledException)
+		{
+		}
+		catch (Exception e)
+		{
+			Console.WriteLine(e);
+
+			await Task.Delay(100, cancellationToken).ConfigureAwait(false);
+		}
+	}
 }

CacheManagerClear.Rabbit/RabbitCacheClearBuilder.cs

@@ -89,7 +89,7 @@ public static CacheManagerClearBuilder UseRabbitSubscriber(this CacheManagerClea
 		}
 #endif
 
-		var subscriber = new CacheSubscriber(connection, cacheManager, exchange, queue);
+		var subscriber = new CacheSubscriber(connection, exchange, queue, cacheManager);
 
 		_ = builder.BuildSubscriber(subscriber);
 
@@ -141,7 +141,7 @@ public static CacheManagerClearBuilder UseRabbitPublisherAndSubscriber(this Cach
 #endif
 
 		var publisher = new CachePublisher(connection, exchange);
-		var subscriber = new CacheSubscriber(connection, cacheManager, exchange, queue);
+		var subscriber = new CacheSubscriber(connection, exchange, queue, cacheManager);
 
 		_ = builder.BuildPublisherAndSubscriber(publisher, subscriber);
 

CacheManagerClear.Redis/CachePublisher.cs

@@ -55,8 +55,14 @@ public async Task PublishClearAllCacheAsync(CancellationToken? cancellationToken
 	/// </summary>
 	public async Task StopAsync()
 	{
+		if (_disposed)
+		{
+			return;
+		}
+
 		await _redis.DisposeAsync().ConfigureAwait(false);
 		await _redis.CloseAsync().ConfigureAwait(false);
+		_disposed = true;
 	}
 
 	/// <summary>

CacheManagerClear.Redis/CacheSubscriber.cs

@@ -60,8 +60,14 @@ await subscriber.SubscribeAsync(channel, async (_, message) =>
 	/// </summary>
 	public async Task StopAsync()
 	{
+		if (_disposed)
+		{
+			return;
+		}
+
 		await _redis.DisposeAsync().ConfigureAwait(false);
 		await _redis.CloseAsync().ConfigureAwait(false);
+		_disposed = true;
 	}
 
 	/// <summary>

CacheManagerUnitTest/StaticData.cs

@@ -6,4 +6,6 @@ internal static class StaticData
 	internal const string Value = "test_value";
 	public const string ClearAllKey = "*";
 	public const string Topic = "cache_clear_topic";
+	public const string Exchange = "cache_clear_exchange";
+	public const string Queue = "cache_clear_queue";
 }