Merge pull request #54 from engineering87/feature/background-event-processing

engineering87 · web-flow · commit 253aaba4f536 · 2024-12-11T00:51:38.000+01:00
Feature/background event processing
diff --git a/src/WART-Client/WART-Client.csproj b/src/WART-Client/WART-Client.csproj
@@ -24,7 +24,7 @@
     <PackageReference Include="Microsoft.AspNetCore.SignalR.Client" Version="9.0.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration" Version="9.0.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="9.0.0" />
-    <PackageReference Include="System.IdentityModel.Tokens.Jwt" Version="8.2.1" />
+    <PackageReference Include="System.IdentityModel.Tokens.Jwt" Version="8.3.0" />
   </ItemGroup>
 
 </Project>
diff --git a/src/WART-Client/WartTestClient.cs b/src/WART-Client/WartTestClient.cs
@@ -53,7 +53,7 @@ public static async Task ConnectAsync(string wartHubUrl)
             }
             catch (Exception e)
             {
-                Console.WriteLine(e);
+                Console.WriteLine(e.Message);
             }
 
             await Task.CompletedTask;
diff --git a/src/WART-Client/WartTestClientJwt.cs b/src/WART-Client/WartTestClientJwt.cs
@@ -58,7 +58,7 @@ public static async Task ConnectAsync(string wartHubUrl, string key)
             }
             catch (Exception e)
             {
-                Console.WriteLine(e);
+                Console.WriteLine(e.Message);
             }
 
             await Task.CompletedTask;
diff --git a/src/WART-Core/Authentication/JWT/JwtServiceCollectionExtension.cs b/src/WART-Core/Authentication/JWT/JwtServiceCollectionExtension.cs
@@ -11,6 +11,8 @@
 using Microsoft.Extensions.Logging;
 using Microsoft.IdentityModel.Tokens;
 using System.Linq;
+using WART_Core.Hubs;
+using WART_Core.Services;
 
 namespace WART_Core.Authentication.JWT
 {
@@ -79,6 +81,12 @@ public static IServiceCollection AddJwtMiddleware(this IServiceCollection servic
                 };
             });
 
+            // Register WART event queue as a singleton service.
+            services.AddSingleton<WartEventQueueService>();
+
+            // Register the WART event worker as a hosted service.
+            services.AddHostedService<WartEventWorker<WartHubJwt>>();
+
             // Configure SignalR options, including error handling and timeouts
             services.AddSignalR(options =>
             {
diff --git a/src/WART-Core/Controllers/WartBaseController.cs b/src/WART-Core/Controllers/WartBaseController.cs
@@ -4,12 +4,11 @@
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.AspNetCore.SignalR;
 using Microsoft.Extensions.Logging;
-using System;
-using System.Collections.Generic;
 using System.Linq;
-using System.Threading.Tasks;
 using WART_Core.Entity;
 using WART_Core.Filters;
+using Microsoft.Extensions.DependencyInjection;
+using WART_Core.Services;
 
 namespace WART_Core.Controllers
 {
@@ -19,6 +18,8 @@ public abstract class WartBaseController<THub> : Controller where THub : Hub
         private readonly IHubContext<THub> _hubContext;
         private const string RouteDataKey = "REQUEST";
 
+        private WartEventQueueService _eventQueue;
+
         protected WartBaseController(IHubContext<THub> hubContext, ILogger logger)
         {
             _hubContext = hubContext;
@@ -39,7 +40,7 @@ public override void OnActionExecuting(ActionExecutingContext context)
         /// Processes the executed action and sends the event to the SignalR hub if applicable.
         /// </summary>
         /// <param name="context">The action executed context.</param>
-        public override async void OnActionExecuted(ActionExecutedContext context)
+        public override void OnActionExecuted(ActionExecutedContext context)
         {
             if (context?.Result is ObjectResult objectResult)
             {
@@ -52,107 +53,13 @@ public override async void OnActionExecuted(ActionExecutedContext context)
                     var response = objectResult.Value;
 
                     var wartEvent = new WartEvent(request, response, httpMethod, httpPath, remoteAddress);
-                    await SendToHub(wartEvent, [.. context.Filters]);
-                }
-            }
-
-            base.OnActionExecuted(context);
-        }
 
-        /// <summary>
-        /// Sends the current event to the SignalR hub.
-        /// </summary>
-        /// <param name="wartEvent">The current WartEvent.</param>
-        /// <param name="filters">The list of filters applied to the request.</param>
-        private async Task SendToHub(WartEvent wartEvent, List<IFilterMetadata> filters)
-        {
-            try
-            {
-                // Retrieve the target groups based on the filters.
-                var groups = GetTargetGroups(filters);
-
-                // If there are groups specified, send the event to each group in parallel.
-                if (groups.Any())
-                {
-                    // Sending events to multiple groups in parallel.
-                    var tasks = groups.Select(group => SendEventToGroup(wartEvent, group));
-                    await Task.WhenAll(tasks);
+                    _eventQueue = context.HttpContext?.RequestServices.GetService<WartEventQueueService>();
+                    _eventQueue?.Enqueue(new WartEventWithFilters(wartEvent, [.. context.Filters]));
                 }
-                else
-                {
-                    // If no specific groups are defined, send the event to all connected clients.
-                    await SendEventToAllClients(wartEvent);
-                }
-            }
-            catch (Exception ex)
-            {
-                _logger?.LogError(ex, "Error sending WartEvent to clients");
             }
-        }
-
-        /// <summary>
-        /// Retrieves the list of groups that the WartEvent should be sent to, based on the provided filters.
-        /// </summary>
-        /// <param name="filters">The list of filters that may contain group-related information.</param>
-        /// <returns>A list of group names to send the WartEvent to.</returns>
-        private IEnumerable<string> GetTargetGroups(List<IFilterMetadata> filters)
-        {
-            var groups = new List<string>();
 
-            // Check if there is a GroupWartAttribute filter indicating groups.
-            if (filters.Any(f => f.GetType().Name == nameof(GroupWartAttribute)))
-            {
-                var wartGroup = filters.FirstOrDefault(f => f.GetType() == typeof(GroupWartAttribute)) as GroupWartAttribute;
-                if (wartGroup != null)
-                {
-                    groups.AddRange(wartGroup.GroupNames);
-                }
-            }
-
-            return groups;
-        }
-
-        /// <summary>
-        /// Sends the WartEvent to a specific group of clients.
-        /// </summary>
-        /// <param name="wartEvent">The WartEvent to send to the group.</param>
-        /// <param name="group">The group name to which the event will be sent.</param>
-        /// <returns>A Task representing the asynchronous operation.</returns>
-        private async Task SendEventToGroup(WartEvent wartEvent, string group)
-        {
-            try
-            {
-                await _hubContext?.Clients
-                    .Group(group)
-                    .SendAsync("Send", wartEvent.ToString());
-
-                _logger?.LogInformation($"Group: {group}, WartEvent: {wartEvent}");
-            }
-            catch (Exception ex)
-            {
-                _logger?.LogError(ex, $"Error sending WartEvent to group {group}");
-            }
-        }
-
-        /// <summary>
-        /// Sends the WartEvent to all connected clients.
-        /// </summary>
-        /// <param name="wartEvent">The WartEvent to send to all clients.</param>
-        /// <returns>A Task representing the asynchronous operation.</returns>
-        private async Task SendEventToAllClients(WartEvent wartEvent)
-        {
-            try
-            {
-                // Send the WartEvent to all connected clients using SignalR.
-                await _hubContext?.Clients.All
-                    .SendAsync("Send", wartEvent.ToString());
-
-                _logger?.LogInformation("Event: {EventName}, Details: {EventDetails}", nameof(WartEvent), wartEvent.ToString());
-            }
-            catch (Exception ex)
-            {
-                _logger?.LogError(ex, "Error sending WartEvent to all clients");
-            }
+            base.OnActionExecuted(context);
         }
     }
 }
diff --git a/src/WART-Core/Entity/WartEventWithFilters.cs b/src/WART-Core/Entity/WartEventWithFilters.cs
@@ -0,0 +1,35 @@
+﻿// (c) 2024 Francesco Del Re <francesco.delre.87@gmail.com>
+// This code is licensed under MIT license (see LICENSE.txt for details)
+using Microsoft.AspNetCore.Mvc.Filters;
+using System.Collections.Generic;
+
+namespace WART_Core.Entity
+{
+    /// <summary>
+    /// Represents an event that contains additional filter metadata.
+    /// </summary>
+    public class WartEventWithFilters
+    {
+        /// <summary>
+        /// The main WartEvent object.
+        /// </summary>
+        public WartEvent WartEvent { get; set; }
+
+        /// <summary>
+        /// A list of filters applied to the event.
+        /// </summary>
+        public List<IFilterMetadata> Filters { get; set; }
+
+        /// <summary>
+        /// Initializes a new instance of the WartEventWithFilters class.
+        /// </summary>
+        /// <param name="wartEvent">The WartEvent to associate with the filters.</param>
+        /// <param name="filters">The list of filters applied to the event.</param>
+        public WartEventWithFilters(WartEvent wartEvent, List<IFilterMetadata> filters)
+        {
+            // Initialize the WartEvent and Filters properties
+            WartEvent = wartEvent;
+            Filters = filters;
+        }
+    }
+}
diff --git a/src/WART-Core/Hubs/WartHubBase.cs b/src/WART-Core/Hubs/WartHubBase.cs
@@ -122,5 +122,10 @@ public static int GetConnectionsCount()
         {
             return _connections.Count;
         }
+
+        /// <summary>
+        /// Returns a value indicating whether there are connected clients.
+        /// </summary>
+        public static bool HasConnectedClients => !_connections.IsEmpty;
     }
 }
diff --git a/src/WART-Core/Middleware/WartServiceCollectionExtension.cs b/src/WART-Core/Middleware/WartServiceCollectionExtension.cs
@@ -9,6 +9,8 @@
 using System.Linq;
 using WART_Core.Authentication.JWT;
 using WART_Core.Enum;
+using WART_Core.Hubs;
+using WART_Core.Services;
 
 namespace WART_Core.Middleware
 {
@@ -32,6 +34,12 @@ public static IServiceCollection AddWartMiddleware(this IServiceCollection servi
             // Add console logging.
             services.AddLogging(configure => configure.AddConsole());
 
+            // Register WART event queue as a singleton service.
+            services.AddSingleton<WartEventQueueService>();
+
+            // Register the WART event worker as a hosted service.
+            services.AddHostedService<WartEventWorker<WartHub>>();
+
             // Configure SignalR with custom options.
             services.AddSignalR(options =>
             {
diff --git a/src/WART-Core/Services/WartEventQueueService.cs b/src/WART-Core/Services/WartEventQueueService.cs
@@ -0,0 +1,43 @@
+﻿// (c) 2024 Francesco Del Re <francesco.delre.87@gmail.com>
+// This code is licensed under MIT license (see LICENSE.txt for details)
+using System.Collections.Concurrent;
+using WART_Core.Entity;
+
+namespace WART_Core.Services
+{
+    /// <summary>
+    /// A service that manages a concurrent queue for WartEvent objects with filters.
+    /// This class provides methods for enqueuing and dequeuing events.
+    /// </summary>
+    public class WartEventQueueService
+    {
+        // A thread-safe queue to hold WartEvent objects along with their associated filters.
+        private readonly ConcurrentQueue<WartEventWithFilters> _queue = new ConcurrentQueue<WartEventWithFilters>();
+
+        /// <summary>
+        /// Enqueues a WartEventWithFilters object to the queue.
+        /// </summary>
+        /// <param name="wartEventWithFilters">The WartEventWithFilters object to enqueue.</param>
+        public void Enqueue(WartEventWithFilters wartEventWithFilters)
+        {
+            // Adds the event with filters to the concurrent queue.
+            _queue.Enqueue(wartEventWithFilters);
+        }
+
+        /// <summary>
+        /// Attempts to dequeue a WartEventWithFilters object from the queue.
+        /// </summary>
+        /// <param name="wartEventWithFilters">The dequeued WartEventWithFilters object.</param>
+        /// <returns>True if an event was dequeued; otherwise, false.</returns>
+        public bool TryDequeue(out WartEventWithFilters wartEventWithFilters)
+        {
+            // Attempts to remove and return the event with filters from the queue.
+            return _queue.TryDequeue(out wartEventWithFilters);
+        }
+
+        /// <summary>
+        /// Gets the current count of events in the queue.
+        /// </summary>
+        public int Count => _queue.Count;
+    }
+}
diff --git a/src/WART-Core/Services/WartEventWorker.cs b/src/WART-Core/Services/WartEventWorker.cs
diff --git a/src/WART-WebApiRealTime/WART-Api.csproj b/src/WART-WebApiRealTime/WART-Api.csproj

Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ public static async Task ConnectAsync(string wartHubUrl)`
`53`	`53`	`}`
`54`	`54`	`catch (Exception e)`
`55`	`55`	`{`
`56`		`- Console.WriteLine(e);`
	`56`	`+ Console.WriteLine(e.Message);`
`57`	`57`	`}`
`58`	`58`
`59`	`59`	`await Task.CompletedTask;`
Original file line number	Diff line number	Diff line change
`@@ -58,7 +58,7 @@ public static async Task ConnectAsync(string wartHubUrl, string key)`
`58`	`58`	`}`
`59`	`59`	`catch (Exception e)`
`60`	`60`	`{`
`61`		`- Console.WriteLine(e);`
	`61`	`+ Console.WriteLine(e.Message);`
`62`	`62`	`}`
`63`	`63`
`64`	`64`	`await Task.CompletedTask;`
Original file line number	Diff line number	Diff line change
`@@ -122,5 +122,10 @@ public static int GetConnectionsCount()`
`122`	`122`	`{`
`123`	`123`	`return _connections.Count;`
`124`	`124`	`}`
	`125`	`+`
	`126`	`+ /// <summary>`
	`127`	`+ /// Returns a value indicating whether there are connected clients.`
	`128`	`+ /// </summary>`
	`129`	`+ public static bool HasConnectedClients => !_connections.IsEmpty;`
`125`	`130`	`}`
`126`	`131`	`}`