Merge pull request #4 from TimetableDesigner/dev

performance upgrade

TimetableDesigner.Backend.Events.OutboxPattern/Event.cs

@@ -11,9 +11,9 @@ public class Event
     public DateTimeOffset? LastRetryOn { get; set; }
     public uint RetryCount { get; set; }
     
-    public Event(object payload)
+    public static Event Create<T>(T payload) where T : class => new Event
     {
-        Payload = JsonSerializer.Serialize(payload);
+        Payload = JsonSerializer.Serialize<T>(payload),
-        PayloadType = payload.GetType().FullName!;
+        PayloadType = payload.GetType().FullName!
-    }
+    };
 }

TimetableDesigner.Backend.Events.OutboxPattern/EventOutboxSender.cs

@@ -1,5 +1,6 @@
 using System.Text.Json;
 using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 
@@ -7,17 +8,23 @@ namespace TimetableDesigner.Backend.Events.OutboxPattern;
 
 public class EventOutboxSender<TDbContext> : BackgroundService where TDbContext : DbContext, IEventOutboxDbContext
 {
+    private readonly IConfiguration _configuration;
     private readonly IServiceScopeFactory _serviceScopeFactory;
     private readonly IEventQueuePublisher _eventQueuePublisher;
     
-    public EventOutboxSender(IServiceScopeFactory serviceScopeFactory, IEventQueuePublisher eventQueuePublisher)
+    public EventOutboxSender(IConfiguration configuration, IServiceScopeFactory serviceScopeFactory, IEventQueuePublisher eventQueuePublisher)
     {
+        _configuration = configuration;
         _serviceScopeFactory = serviceScopeFactory;
         _eventQueuePublisher = eventQueuePublisher;
     }
     
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
+        IConfigurationSection section = _configuration.GetSection("Workers:EventOutboxSender");
+        int delayBetweenQueries = section.GetValue("DelayBetweenEmptyQueries", 10);
+        int limitInSingleQuery = section.GetValue("LimitInSingleQuery", 50);
+        
         using (IServiceScope scope = _serviceScopeFactory.CreateScope())
         using (TDbContext dbContext = scope.ServiceProvider.GetRequiredService<TDbContext>())
         {
@@ -25,29 +32,37 @@ public class EventOutboxSender<TDbContext> : BackgroundService where TDbContext
             
             while (!stoppingToken.IsCancellationRequested)
             {
-                Event? eventData = await dbContext.Events.FirstOrDefaultAsync(x => x.LastRetryOn == null, stoppingToken)
+                IAsyncEnumerable<Event>? events = dbContext.Events
-                                ?? await dbContext.Events.OrderBy(x => x.LastRetryOn).FirstOrDefaultAsync(stoppingToken);
+                                                           .OrderBy(x => x.LastRetryOn)
+                                                           .Take(limitInSingleQuery)
+                                                           .ToAsyncEnumerable();
 
-                if (eventData is null)
+                bool any = false;
+                await foreach (Event eventData in events)
                 {
-                    continue;
+                    any = true;
+                    
+                    Type payloadType = Type.GetType(eventData.PayloadType)!;
+                    JsonSerializer.Deserialize(eventData.Payload, payloadType);
+
+                    try
+                    {
+                        await _eventQueuePublisher.PublishAsync(eventData.Payload, payloadType);
+                    
+                        dbContext.Events.Remove(eventData);
+                    }
+                    catch
+                    {
+                        eventData.LastRetryOn = DateTimeOffset.UtcNow;
+                        eventData.RetryCount++;
+                    }
                 }
                 
-                Type payloadType = Type.GetType(eventData.PayloadType)!;
+                if (!any)
-                JsonSerializer.Deserialize(eventData.Payload, payloadType);
-
-                try
                 {
-                    await _eventQueuePublisher.PublishAsync(eventData.Payload, payloadType);
+                    await Task.Delay(TimeSpan.FromSeconds(delayBetweenQueries), stoppingToken);
-                    
-                    dbContext.Events.Remove(eventData);
                 }
-                catch
+                else
-                {
-                    eventData.LastRetryOn = DateTimeOffset.UtcNow;
-                    eventData.RetryCount++;
-                }
-                finally
                 {
                     await dbContext.SaveChangesAsync(stoppingToken);
                 }