TimetableDesigner/TimetableDesigner.Backend.Events.OutboxPattern
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Activity

performance upgrade

Browse Source
This commit is contained in:
Mateusz Skoczek
2026-01-29 23:44:41 +01:00
Unverified
parent f0ccc557e1
commit 8f7c858cc5
1 changed files with 33 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
TimetableDesigner.Backend.Events.OutboxPattern/EventOutboxSender.cs
View File

@@ -1,5 +1,6 @@
using System.Text.Json; using System.Text.Json;
using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Hosting;
@@ -7,17 +8,23 @@ namespace TimetableDesigner.Backend.Events.OutboxPattern;
public class EventOutboxSender<TDbContext> : BackgroundService where TDbContext : DbContext, IEventOutboxDbContext public class EventOutboxSender<TDbContext> : BackgroundService where TDbContext : DbContext, IEventOutboxDbContext
{ {
private readonly IConfiguration _configuration;
private readonly IServiceScopeFactory _serviceScopeFactory; private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly IEventQueuePublisher _eventQueuePublisher; private readonly IEventQueuePublisher _eventQueuePublisher;
public EventOutboxSender(IServiceScopeFactory serviceScopeFactory, IEventQueuePublisher eventQueuePublisher) public EventOutboxSender(IConfiguration configuration, IServiceScopeFactory serviceScopeFactory, IEventQueuePublisher eventQueuePublisher)
{ {
_configuration = configuration;
_serviceScopeFactory = serviceScopeFactory; _serviceScopeFactory = serviceScopeFactory;
_eventQueuePublisher = eventQueuePublisher; _eventQueuePublisher = eventQueuePublisher;
} }
protected override async Task ExecuteAsync(CancellationToken stoppingToken) 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 (IServiceScope scope = _serviceScopeFactory.CreateScope())
using (TDbContext dbContext = scope.ServiceProvider.GetRequiredService<TDbContext>()) using (TDbContext dbContext = scope.ServiceProvider.GetRequiredService<TDbContext>())
{ {
@@ -25,29 +32,37 @@ public class EventOutboxSender<TDbContext> : BackgroundService where TDbContext
while (!stoppingToken.IsCancellationRequested) 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); await dbContext.SaveChangesAsync(stoppingToken);
} }