Skip to content

Commit 6d38988

Browse files
author
Ken
committed
Fix and test mongo outbox implementation.
1 parent a41e35e commit 6d38988

20 files changed

+181
-83
lines changed

‎MinimalDomainEvents.sln

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,9 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MinimalDomainEvents.Outbox.
5555
EndProject
5656
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MinimalDomainEvents.Outbox.Worker.Abstractions", "src\MinimalDomainEvents.Outbox.Worker.Abstractions\MinimalDomainEvents.Outbox.Worker.Abstractions.csproj", "{03BBC09D-9E45-487F-AB2A-C0ADA46B58A1}"
5757
EndProject
58-
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MinimalDomainEvents.Outbox.Worker.UnitTests", "test\MinimalDomainEvents.Outbox.Worker.UnitTests\MinimalDomainEvents.Outbox.Worker.UnitTests.csproj", "{F3C36E6D-6A1A-4407-87A6-D77B24C0576D}"
58+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MinimalDomainEvents.Outbox.Worker.UnitTests", "test\MinimalDomainEvents.Outbox.Worker.UnitTests\MinimalDomainEvents.Outbox.Worker.UnitTests.csproj", "{F3C36E6D-6A1A-4407-87A6-D77B24C0576D}"
59+
EndProject
60+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MongoTestContainer", "test\MongoTestContainer\MongoTestContainer.csproj", "{99C8CE04-4BC6-4206-B86C-1CE649C7FC19}"
5961
EndProject
6062
Global
6163
GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -131,6 +133,10 @@ Global
131133
{F3C36E6D-6A1A-4407-87A6-D77B24C0576D}.Debug|Any CPU.Build.0 = Debug|Any CPU
132134
{F3C36E6D-6A1A-4407-87A6-D77B24C0576D}.Release|Any CPU.ActiveCfg = Release|Any CPU
133135
{F3C36E6D-6A1A-4407-87A6-D77B24C0576D}.Release|Any CPU.Build.0 = Release|Any CPU
136+
{99C8CE04-4BC6-4206-B86C-1CE649C7FC19}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
137+
{99C8CE04-4BC6-4206-B86C-1CE649C7FC19}.Debug|Any CPU.Build.0 = Debug|Any CPU
138+
{99C8CE04-4BC6-4206-B86C-1CE649C7FC19}.Release|Any CPU.ActiveCfg = Release|Any CPU
139+
{99C8CE04-4BC6-4206-B86C-1CE649C7FC19}.Release|Any CPU.Build.0 = Release|Any CPU
134140
EndGlobalSection
135141
GlobalSection(SolutionProperties) = preSolution
136142
HideSolutionNode = FALSE
@@ -156,6 +162,7 @@ Global
156162
{2DEA751C-8CE8-4452-A365-68F1AAE76551} = {AB124D7A-09CE-4168-B706-F319426E6383}
157163
{03BBC09D-9E45-487F-AB2A-C0ADA46B58A1} = {AB124D7A-09CE-4168-B706-F319426E6383}
158164
{F3C36E6D-6A1A-4407-87A6-D77B24C0576D} = {A37BECD1-4461-4E64-BF71-B48ACEBB8447}
165+
{99C8CE04-4BC6-4206-B86C-1CE649C7FC19} = {A37BECD1-4461-4E64-BF71-B48ACEBB8447}
159166
EndGlobalSection
160167
GlobalSection(ExtensibilityGlobals) = postSolution
161168
SolutionGuid = {5C04EBF0-7B5F-4CCE-8D3C-55C5B17C2BEB}

‎src/MinimalDomainEvents.Dispatcher.MediatR/IDomainEventDispatcherBuilderExtensions.cs

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,11 +6,13 @@ namespace MinimalDomainEvents.Dispatcher.MediatR;
66

77
public static class IDomainEventDispatcherBuilderExtensions
88
{
9-
public static IServiceCollection AddMediatorDispatcher(this IDomainEventDispatcherBuilder builder)
9+
public static IDomainEventDispatcherBuilder AddMediatorDispatcher(this IDomainEventDispatcherBuilder builder)
1010
{
11-
return builder.Services
11+
builder.Services
1212
.AddScoped<IDispatchDomainEvents, MediatorDispatcher>()
1313
.AddTransient(typeof(IPipelineBehavior<,>), typeof(DomainEventDispatchBehavior<,>))
1414
;
15+
16+
return builder;
1517
}
1618
}

‎src/MinimalDomainEvents.Outbox.Abstractions/IOutboxTransaction.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
namespace MinimalDomainEvents.Outbox.Abstractions;
22
public interface IOutboxTransaction : IDisposable
33
{
4-
Task StartTransaction(CancellationToken cancellationToken = default);
4+
Task StartTransaction(Action? onCommit = null, CancellationToken cancellationToken = default);
55
Task Commit(CancellationToken cancellationToken = default);
66
}

‎src/MinimalDomainEvents.Outbox.MongoDb/IOutboxDispatcherBuilderExtensions.cs

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,9 +29,10 @@ private static void RegisterDefaultServices(IOutboxDispatcherBuilder builder)
2929
builder.Services.TryAddScoped<MongoSessionProvider>();
3030
builder.Services.TryAddScoped<IMongoSessionProvider>(sp => sp.GetRequiredService<MongoSessionProvider>());
3131
builder.Services.TryAddScoped<IMongoSessionProviderInitializer>(sp => sp.GetRequiredService<MongoSessionProvider>());
32-
builder.Services.TryAddScoped<IOutboxRecordCollectionProvider, OutboxRecordCollectionProvider>();
32+
builder.Services.TryAddScoped<OutboxRecordCollectionProvider>();
33+
builder.Services.TryAddScoped<IOutboxRecordCollectionProvider>(sp => sp.GetRequiredService<OutboxRecordCollectionProvider>());
34+
builder.Services.TryAddScoped<IOutboxRecordCollectionInitializer>(sp => sp.GetRequiredService<OutboxRecordCollectionProvider>());
3335
builder.Services.TryAddScoped<ITransactionProvider, MongoDbTransactionProvider>();
3436
builder.Services.TryAddScoped<IRetrieveOutboxRecords, MongoDbOutboxRecordRetriever>();
35-
builder.Services.TryAddScoped<ICleanupOutboxRecords, MongoDbOutboxRecordCleaner>();
3637
}
3738
}

‎src/MinimalDomainEvents.Outbox.MongoDb/MongoDbOutboxRecordCleaner.cs

Lines changed: 0 additions & 30 deletions
This file was deleted.

‎src/MinimalDomainEvents.Outbox.MongoDb/MongoDbOutboxTransaction.cs

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,20 +7,24 @@ internal sealed class MongoDbOutboxTransaction : IOutboxTransaction
77
{
88
public IClientSessionHandle ClientSessionHandle { get; }
99

10+
private Action? _onCommit;
11+
1012
public MongoDbOutboxTransaction(IClientSessionHandle clientSessionHandle)
1113
{
1214
ClientSessionHandle = clientSessionHandle;
1315
}
1416

15-
public Task StartTransaction(CancellationToken cancellationToken = default)
17+
public Task StartTransaction(Action? onCommit = null, CancellationToken cancellationToken = default)
1618
{
1719
ClientSessionHandle.StartTransaction();
20+
_onCommit = onCommit;
1821
return Task.CompletedTask;
1922
}
2023

21-
public Task Commit(CancellationToken cancellationToken = default)
24+
public async Task Commit(CancellationToken cancellationToken = default)
2225
{
23-
return ClientSessionHandle.CommitTransactionAsync(cancellationToken);
26+
await ClientSessionHandle.CommitTransactionAsync(cancellationToken);
27+
_onCommit?.Invoke();
2428
}
2529

2630
public void Dispose()

‎src/MinimalDomainEvents.Outbox.MongoDb/MongoDbTransactionProvider.cs

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,15 @@ public async Task<IOutboxTransaction> NewTransaction(CancellationToken cancellat
2121

2222
var session = await _mongoClient.StartSessionAsync(null, cancellationToken);
2323
_currentTransaction = new MongoDbOutboxTransaction(session);
24-
await _currentTransaction.StartTransaction(cancellationToken);
24+
await _currentTransaction.StartTransaction(ClearTransaction, cancellationToken);
2525
return _currentTransaction;
2626
}
2727

28+
private void ClearTransaction()
29+
{
30+
_currentTransaction = null;
31+
}
32+
2833
public bool TryGetCurrentTransaction(out IOutboxTransaction? outboxTransaction)
2934
{
3035
outboxTransaction = _currentTransaction;

‎src/MinimalDomainEvents.Outbox.MongoDb/OutboxRecordCollectionProvider.cs

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ internal sealed class OutboxRecordCollectionProvider : IOutboxRecordCollectionPr
1717
private readonly MongoClient _mongoClient;
1818

1919
private const string EnqueuedAtIndexName = "EnqueuedAt_asc";
20-
private const string DispatchedAtIndexName = "ExpiresAt_asc";
20+
private const string DispatchedAtIndexName = "OutboxCleanup";
2121

2222
public OutboxRecordCollectionProvider(OutboxSettings outboxSettings, MongoClient mongoClient)
2323
{
@@ -77,7 +77,7 @@ private static async Task CreateEnqueuedAtIndex(IMongoCollection<OutboxRecord> c
7777

7878
private static async Task CreateDispatchedAtIndex(IMongoCollection<OutboxRecord> collection, CancellationToken cancellationToken)
7979
{
80-
var indexKeysDefinition = Builders<OutboxRecord>.IndexKeys.Ascending(or => or.ExpiresAt);
80+
var indexKeysDefinition = Builders<OutboxRecord>.IndexKeys.Ascending(or => or.DispatchedAt);
8181
var createIndexModel = new CreateIndexModel<OutboxRecord>(indexKeysDefinition, new()
8282
{
8383
ExpireAfter = TimeSpan.FromDays(7),
@@ -92,6 +92,7 @@ private static bool TryRegisterOutboxRecordClassMap()
9292
return BsonClassMap.TryRegisterClassMap<OutboxRecord>(cm =>
9393
{
9494
cm.MapIdProperty(or => or.Id);
95+
cm.AutoMap();
9596
cm.UnmapProperty(or => or.ExpiresAt);
9697
cm.SetIgnoreExtraElements(true);
9798
});

‎src/MinimalDomainEvents.Outbox.Worker.Abstractions/ICleanupOutboxRecords.cs

Lines changed: 0 additions & 5 deletions
This file was deleted.

‎src/MinimalDomainEvents.Outbox.Worker/BackgroundDispatchWorker.cs

Lines changed: 19 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
using Microsoft.Extensions.Hosting;
22
using MinimalDomainEvents.Dispatcher.Abstractions;
33
using MinimalDomainEvents.Outbox.Abstractions;
4-
using MinimalDomainEvents.Outbox.Worker.Abstractions;
4+
using System.Diagnostics;
55

66
namespace MinimalDomainEvents.Outbox.Worker;
77
internal sealed class BackgroundDispatchWorker : BackgroundService
@@ -11,41 +11,44 @@ internal sealed class BackgroundDispatchWorker : BackgroundService
1111
private readonly IOutboxRecordCollectionInitializer _outboxRecordCollectionInitializer;
1212
private readonly ITransactionProvider _transactionFactory;
1313
private readonly IDomainEventRetriever _domainEventRetriever;
14-
private readonly ICleanupOutboxRecords _cleanupOutboxRecords;
1514
private readonly IEnumerable<IDispatchDomainEvents> _dispatchers;
1615

1716
public BackgroundDispatchWorker(IOutboxRecordCollectionInitializer outboxRecordCollectionInitializer,
1817
ITransactionProvider transactionFactory,
1918
IDomainEventRetriever domainEventRetriever,
20-
ICleanupOutboxRecords cleanupOutboxRecords,
2119
IEnumerable<IDispatchDomainEvents> dispatchers)
2220
{
2321
_domainEventRetriever = domainEventRetriever;
2422
_outboxRecordCollectionInitializer = outboxRecordCollectionInitializer;
2523
_transactionFactory = transactionFactory;
26-
_cleanupOutboxRecords = cleanupOutboxRecords;
2724
_dispatchers = dispatchers;
2825
}
2926

3027
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
3128
{
32-
await _outboxRecordCollectionInitializer.Initialize(stoppingToken);
33-
34-
while (!stoppingToken.IsCancellationRequested)
29+
try
3530
{
36-
using var transaction = await _transactionFactory.NewTransaction(stoppingToken);
37-
var domainEvents = await _domainEventRetriever.GetAndMarkAsDispatched(stoppingToken);
31+
await _outboxRecordCollectionInitializer.Initialize(stoppingToken);
3832

39-
if (domainEvents is not null && domainEvents.Count > 0)
33+
while (!stoppingToken.IsCancellationRequested)
4034
{
41-
foreach (var dispatcher in _dispatchers)
42-
await dispatcher.Dispatch(domainEvents);
43-
}
35+
using var transaction = await _transactionFactory.NewTransaction(stoppingToken);
36+
var domainEvents = await _domainEventRetriever.GetAndMarkAsDispatched(stoppingToken);
37+
38+
if (domainEvents is not null && domainEvents.Count > 0)
39+
{
40+
foreach (var dispatcher in _dispatchers)
41+
await dispatcher.Dispatch(domainEvents);
42+
}
4443

45-
await _cleanupOutboxRecords.CleanupExpiredOutboxRecords(stoppingToken);
46-
await transaction.Commit(stoppingToken);
44+
await transaction.Commit(stoppingToken);
4745

48-
await Task.Delay(Delay, stoppingToken);
46+
await Task.Delay(Delay, stoppingToken);
47+
}
48+
}
49+
catch (TaskCanceledException)
50+
{
51+
Debug.WriteLine("BackgroundDispatchWorker stopped. Reason: TaskCanceledException.");
4952
}
5053
}
5154
}

0 commit comments

Comments
 (0)