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

fix

Browse Source
This commit is contained in:
Mateusz Skoczek
2026-01-28 23:52:28 +01:00
Unverified
parent e911ffb726
commit 98e2ae5c2c
1 changed files with 33 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,26 +1,32 @@
using System.Text.Json; using System.Text.Json;
using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Hosting;
namespace TimetableDesigner.Backend.Events.OutboxPattern; 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 TDbContext _databaseContext; private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly IEventQueuePublisher _eventQueuePublisher; private readonly IEventQueuePublisher _eventQueuePublisher;
public EventOutboxSender(TDbContext databaseContext, IEventQueuePublisher eventQueuePublisher) public EventOutboxSender(IServiceScopeFactory serviceScopeFactory, IEventQueuePublisher eventQueuePublisher)
{ {
_databaseContext = databaseContext; _serviceScopeFactory = serviceScopeFactory;
_eventQueuePublisher = eventQueuePublisher; _eventQueuePublisher = eventQueuePublisher;
} }
protected override async Task ExecuteAsync(CancellationToken stoppingToken) protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{ {
using (IServiceScope scope = _serviceScopeFactory.CreateScope())
using (TDbContext dbContext = scope.ServiceProvider.GetRequiredService<TDbContext>())
{
await dbContext.Database.EnsureCreatedAsync(stoppingToken);
while (!stoppingToken.IsCancellationRequested) while (!stoppingToken.IsCancellationRequested)
{ {
Event? eventData = await _databaseContext.Events.FirstOrDefaultAsync(x => x.LastRetryOn == null, stoppingToken) Event? eventData = await dbContext.Events.FirstOrDefaultAsync(x => x.LastRetryOn == null, stoppingToken)
?? await _databaseContext.Events.OrderBy(x => x.LastRetryOn).FirstOrDefaultAsync(stoppingToken); ?? await dbContext.Events.OrderBy(x => x.LastRetryOn).FirstOrDefaultAsync(stoppingToken);
if (eventData is null) if (eventData is null)
{ {
@@ -34,7 +40,7 @@ public class EventOutboxSender<TDbContext> : BackgroundService where TDbContext
{ {
await _eventQueuePublisher.PublishAsync(eventData.Payload, payloadType); await _eventQueuePublisher.PublishAsync(eventData.Payload, payloadType);
_databaseContext.Events.Remove(eventData); dbContext.Events.Remove(eventData);
} }
catch catch
{ {
@@ -43,7 +49,8 @@ public class EventOutboxSender<TDbContext> : BackgroundService where TDbContext
} }
finally finally
{ {
await _databaseContext.SaveChangesAsync(stoppingToken); await dbContext.SaveChangesAsync(stoppingToken);
}
} }
} }
} }