Remove messages history for removed hubs

* Small optimizations in Consumer service and factory

* Add removing message history from file when remove EventHub configuration

Author: dvlaskin

src/Application/Services/EventHubConsumerService.cs

@@ -13,8 +13,11 @@ public class EventHubConsumerService : IMessageConsumerService
     private readonly IMessageConsumerProvider messageConsumerProvider;
     private readonly ITextProcessingPipeline textProcessingPipeline;
 
-    private readonly Channel<EventHubMessage> channel = Channel.CreateUnbounded<EventHubMessage>();
-    private readonly Lock startLock = new();
+    private readonly Channel<EventHubMessage> channel = Channel
+        .CreateBounded<EventHubMessage>(new BoundedChannelOptions(100)
+        {
+            FullMode = BoundedChannelFullMode.Wait
+        });
     private bool isProcessing;
 
 
@@ -47,23 +50,15 @@ public async IAsyncEnumerable<EventHubMessage> StartReceiveMessageAsync(
 
     public async Task StopReceiveMessageAsync()
     {
-        lock (startLock)
-        {
-            isProcessing = false;
-        }
+        Interlocked.Exchange(ref isProcessing, false);
         await messageConsumerProvider.StopReceiveMessageAsync();
     }
 
 
     private async Task ReadMessageAsync(CancellationToken cancellationToken)
     {
-        lock (startLock)
-        {
-            if (isProcessing)
-                return;
-            
-            isProcessing = true;
-        }
+        if (Interlocked.CompareExchange(ref isProcessing, true, false))
+            return;
 
         var taskResult = messageConsumerProvider.StartReceiveMessageAsync(async message =>
         {

src/Application/Services/FileBasedMessageHistory.cs

@@ -74,4 +74,15 @@ public async Task RemoveMessageAsync(Guid input, string message)
             await messagesStorageProvider.SaveDataAsync(fullHistory);
         }
     }
+    
+    public async Task RemoveAllAsync(Guid input)
+    {
+        MessagesHistory? fullHistory = await messagesStorageProvider.GetDataAsync();
+        
+        if (fullHistory is null)
+            return;
+        
+        if (fullHistory.Messages.Remove(input))
+            await messagesStorageProvider.SaveDataAsync(fullHistory);
+    }
 }

src/Domain/Interfaces/Services/IMessageHistory.cs

@@ -5,4 +5,5 @@ public interface IMessageHistory<in TInput, TResult>
     Task<TResult> GetHistoryAsync(TInput input);
     Task AddMessageAsync(TInput input, string message);
     Task RemoveMessageAsync(TInput input, string message);
+    Task RemoveAllAsync(Guid input);
 }

src/Infrastructure/Factories/MessageConsumerFactory.cs

@@ -30,10 +30,10 @@ IServiceProvider serviceProvider
 
     public IMessageConsumerService CreateConsumer(Guid configId)
     {
-        logger.LogInformation("Creating producer for configId: {ConfigId}", configId);
+        logger.LogInformation("Creating consumer for configId: {ConfigId}", configId);
         var eventHubConfig = config.CurrentValue.EventHubsConfigs.First(x => x.Id == configId);
 
-        IMessageConsumerProvider ehConsumerProvider = eventHubConfig.UseCheckpoints 
+        IMessageConsumerProvider ehConsumerProvider = eventHubConfig.UseCheckpoints
             ? CreateConsumerWithStorage(eventHubConfig) 
             : CreateConsumerWithoutStorage(eventHubConfig);
 
@@ -46,18 +46,12 @@ public IMessageConsumerService CreateConsumer(Guid configId)
 
     private EventHubConsumerProviderWithStorage CreateConsumerWithStorage(EventHubConfig eventHubConfig)
     {
-        return ActivatorUtilities.CreateInstance<EventHubConsumerProviderWithStorage>(
-            serviceProvider,
-            eventHubConfig
-        );
+        return ActivatorUtilities.CreateInstance<EventHubConsumerProviderWithStorage>(serviceProvider, eventHubConfig);
     }
 
     private EventHubConsumerProviderWithoutStorage CreateConsumerWithoutStorage(EventHubConfig eventHubConfig)
     {
-        return ActivatorUtilities.CreateInstance<EventHubConsumerProviderWithoutStorage>(
-            serviceProvider,
-            eventHubConfig
-        );
+        return ActivatorUtilities.CreateInstance<EventHubConsumerProviderWithoutStorage>(serviceProvider, eventHubConfig);
     }
 
 
@@ -76,7 +70,7 @@ private IMessageFormatter[] GetActiveMessageFormatters(EventHubConfig eventHubCo
             .MessageFormatters
             .Where(x => x.Value)
             .Select(s => s.Key)
-            .ToArray();
+            .ToHashSet();
 
         var messageFormattersList = serviceProvider
             .GetServices<IMessageFormatter>()

src/WebUI/Components/Pages/Configuration.razor

@@ -4,6 +4,7 @@
 @using Domain.Interfaces.Services
 
 @inject IFileStorageProvider<AppConfiguration> ConfigurationProvider
+@inject IMessageHistory<Guid, List<string>> MessagesHistoryService
 
 @rendermode InteractiveServer
 
@@ -181,6 +182,10 @@ else
 
     private void RemoveConfig(int index)
     {
+        var configId = appConfiguration?.EventHubsConfigs[index].Id;
+        if (configId is not null)
+            MessagesHistoryService.RemoveAllAsync(configId.Value);
+        
         appConfiguration?.EventHubsConfigs.RemoveAt(index);
     }
 

src/WebUI/Components/Pages/EventHub.razor

@@ -9,7 +9,6 @@
 @using Microsoft.Extensions.Options
 @rendermode InteractiveServer
 @inject IJSRuntime JsRuntime
-@implements IDisposable
 @implements IAsyncDisposable
 
 <PageTitle>EventHub - @title</PageTitle>
@@ -349,27 +348,6 @@
         _ = toast?.ShowToast($"{infoMessage}: {ex.Message}", ToastType.Error, TimeSpan.FromSeconds(10));
     }
 
-    
-    public void Dispose()
-    {
-        // Cancel any pending operations first
-        receiverCts.Cancel();
-        delayedSenderCts.Cancel();
-        
-        // Dispose of services
-        messageProducer?.DisposeAsync().GetAwaiter().GetResult();
-        messageConsumer?.DisposeAsync().GetAwaiter().GetResult();
-        
-        // Dispose of cancellation tokens
-        receiverCts.Dispose();
-        delayedSenderCts.Dispose();
-        
-        // Clear collections
-        receivedMessages.Clear();
-        messagesHistoryList.Clear();
-        
-        GC.SuppressFinalize(this);
-    }
 
     public async ValueTask DisposeAsync()
     {