|
| 1 | +--- |
| 2 | +outline: deep |
| 3 | +--- |
| 4 | + |
| 5 | +# Polling |
| 6 | + |
| 7 | +To use OutboxKit with the MongoDB polling provider (and others as well), you'll need to choose between two paths: accept the library defaults, making your infra match them, or make use of the library's flexibility to adapt to your existing infrastructure. |
| 8 | + |
| 9 | +::: tip |
| 10 | +Don't skip the [important notes](#important-notes) section at the end of the page. |
| 11 | +::: |
| 12 | + |
| 13 | +## Using the defaults |
| 14 | + |
| 15 | +In the box you'll find the `Message` record, which looks something like this: |
| 16 | + |
| 17 | +```csharp |
| 18 | +public sealed record Message : IMessage |
| 19 | +{ |
| 20 | + public ObjectId Id { get; init; } |
| 21 | + public required string Type { get; init; } |
| 22 | + public required byte[] Payload { get; init; } |
| 23 | + public required DateTime CreatedAt { get; init; } |
| 24 | + public byte[]? TraceContext { get; init; } |
| 25 | +} |
| 26 | +``` |
| 27 | + |
| 28 | +By default, these messages will be stored in a collection named `outbox_messages`. Additionally, with these defaults, the `Id` property will be used to order the messages. |
| 29 | + |
| 30 | +Because MongoDB, unlike relational databases, doesn't support traditional locking mechanisms (e.g. `SELECT ... FOR UPDATE`), OutboxKit implements its own locking mechanism. This locking mechanism requires the use of an auxiliary collection, which by default is named `outbox_locks`. |
| 31 | + |
| 32 | +Assuming you're happy with the out of the box defaults, setting up the provider with DI would look something like this: |
| 33 | + |
| 34 | +```csharp |
| 35 | +services.AddOutboxKit(kit => |
| 36 | + kit.WithMongoDbPolling(p => |
| 37 | + p.WithDatabaseFactory((_, s) => s.GetRequiredService<IMongoDatabase>())); |
| 38 | +``` |
| 39 | + |
| 40 | +Note that this assumes a singleton `IMongoDatabase` is registered in the DI container, but you can do whatever you want in `WithDatabaseFactory`, as long as it returns an `IMongoDatabase` instance. |
| 41 | + |
| 42 | +## Making it your own |
| 43 | + |
| 44 | +Now, while the defaults are nice, one of the motivations for building OutboxKit in the first place, is to make it possible to adapt to specific applications and their infrastructure, which means there's a bunch of things that can be configured. |
| 45 | + |
| 46 | +Let's start with a snippet that shows all the things you can configure: |
| 47 | + |
| 48 | +```csharp |
| 49 | +services.AddOutboxKit(kit => |
| 50 | + kit |
| 51 | + .WithMongoDbPolling(p => |
| 52 | + p |
| 53 | + .WithDatabaseFactory((_, s) => s.GetRequiredService<IMongoDatabase>()) |
| 54 | + .WithBatchSize(100) |
| 55 | + .WithPollingInterval(TimeSpan.FromMinutes(5)) |
| 56 | + .WithCollection<OutboxMessage, ObjectId>(c => c |
| 57 | + .WithName("OutboxMessages") |
| 58 | + .WithIdSelector(m => m.Id) |
| 59 | + .WithSort(new SortDefinitionBuilder<OutboxMessage>().Ascending(m => m.Id)) |
| 60 | + .WithProcessedAtSelector(m => m.ProcessedAt)) |
| 61 | + .WithUpdateProcessed(u => u |
| 62 | + .WithCleanUpInterval(TimeSpan.FromHours(1)) |
| 63 | + .WithMaxAge(TimeSpan.FromDays(1))) |
| 64 | + .WithDistributedLock(l => l |
| 65 | + .WithCollectionName("OutboxLocks") |
| 66 | + .WithId("OutboxLock") |
| 67 | + .WithOwner(Environment.MachineName) |
| 68 | + .WithChangeStreamsEnabled(true)))); |
| 69 | +``` |
| 70 | + |
| 71 | +So, it's not massive, but there still are a few options available. |
| 72 | + |
| 73 | +Note that not everything is always mandatory, but there are some things that are dependent on each other, so if you set one, you'll need to set some others. |
| 74 | + |
| 75 | +`WithDatabaseFactory` is the way to get a `IMongoDatabase` instance for OutboxKit to interact with the database. It is the only configuration that is always required. |
| 76 | + |
| 77 | +`WithBatchSize` allows you to set the maximum number of messages that will be made available to the `IBatchProducer` in one go. |
| 78 | + |
| 79 | +`WithPollingInterval` allows you to customize how often polling should happen. |
| 80 | + |
| 81 | +`WithCollection` is where you can configure the collection you want OutboxKit to use. If you're fine with OutboxKit's defaults, no need to call `WithCollection`, but if you want to customize something, then you need to configure everything, using all the exposed builder methods (minus `WithProcessedAtSelector`, but we'll look at that later). |
| 82 | + |
| 83 | +`WithName` allows configuring the outbox message collection name. |
| 84 | + |
| 85 | +`WithIdSelector` allows you to configure the id selector for the outbox message, which will be used when acknowledging the messages produced. |
| 86 | + |
| 87 | +`WithSort` is the way to configure how to sort the messages as they're fetched from the outbox collection and made available to produce. |
| 88 | + |
| 89 | +Let's talk about `WithUpdateProcessed`, then come back to `WithProcessedAtSelector`. |
| 90 | + |
| 91 | +By default, OutboxKit will immediately delete the messages that have been produced. However, if you want to keep them around for a while, you can change the strategy to mark them as processed instead. To do this, you use `WithUpdateProcessed`. |
| 92 | + |
| 93 | +When using `WithUpdateProcessed`, you can configure how often the messages should be cleaned up using `WithCleanUpInterval`, and how old the messages should be before they are cleaned up using `WithMaxAge`. |
| 94 | + |
| 95 | +Note that, if you use `WithUpdateProcessed`, you must use `WithProcessedAtSelector`, in order for the library to do its magic. When marking the messages as processed, OutboxKit will set the column to a `DateTime` in UTC, obtained from a [`TimeProvider`](https://learn.microsoft.com/en-us/dotnet/api/system.timeprovider) it gets from DI. |
| 96 | +
|
| 97 | +As mentioned earlier, because MongoDB doesn't support traditional locking mechanisms, OutboxKit implements a distributed locking mechanism itself, making use of an auxiliary collection for it. You can tweak some aspects of this, by using `WithDistributedLock`. You can tweak the collection name with `WithCollectionName`, the id associated with the lock with `WithId` (should be the same in all instances of the application interacting with the outbox), the owner of the lock with `WithOwner` (should be different per instance of the application interacting with the outbox), and whether or not to use [MongoDB change streams](https://www.mongodb.com/docs/manual/changeStreams/) with `WithChangeStreamsEnabled`. The default values for these are `outbox_locks`, `outbox_lock`, [`Environment.MachineName`](https://learn.microsoft.com/en-us/dotnet/api/system.environment.machinename) and `false`, respectively. |
| 98 | + |
| 99 | +## Multi-database |
| 100 | + |
| 101 | +If your application uses multiple MongoDB databases, and you need an outbox for each of them (for example, you have a multi-tenant application, where each tenant uses a different database), everything we discussed so far still applies, you just need to tweak things very slightly. |
| 102 | + |
| 103 | +`WithMongoDbPolling` has an overload that takes a `string` as the first argument, allowing you to identify the outbox. |
| 104 | + |
| 105 | +Taking the defaults approach as an example, you could set up two outboxes like this: |
| 106 | + |
| 107 | +```csharp |
| 108 | +services.AddOutboxKit(kit => |
| 109 | + kit |
| 110 | + .WithMongoDbPolling( |
| 111 | + tenantOne, |
| 112 | + p => p.WithDatabaseFactory((k, s) => s.GetRequiredKeyedService<IMongoDatabase>(tenantOne))) |
| 113 | + .WithMongoDbPolling( |
| 114 | + tenantTwo, |
| 115 | + p => p.WithDatabaseFactory((k, s) => s.GetRequiredKeyedService<IMongoDatabase>(tenantTwo)))); |
| 116 | +``` |
| 117 | + |
| 118 | +As you can infer, this means you can not only have multiple databases, but you can also configure them differently (not sure it's the most relevant thing ever, but hey, it works). |
| 119 | + |
| 120 | +You'll notice we're using `GetRequiredKeyedService` instead of `GetRequiredService`. You don't have to do it like this, but it's a simple way to have different dependencies resolved for different contexts, like a multi-tenant application. |
| 121 | + |
| 122 | +As discussed in [Core/Producing messages](/core/producing-messages), the `IBatchProducer` `ProduceAsync` method receives an `OutboxKey`, composed by a provider key (`"mongodb_polling"` in this case) and a client key, which is what you passed to `WithMongoDbPolling`. If you only have one outbox and don't set the key, you'll get the `string` `"default"`. |
| 123 | + |
| 124 | +The `k` parameter shown in the example above in the `WithDatabaseFactory` method is also the aforementioned `OutboxKey`, and it's passed in case it's useful to resolve the `IMongoDatabase` instance. If you don't need it, you can just ignore it. |
| 125 | + |
| 126 | +## Important notes |
| 127 | + |
| 128 | +- Due to the need to implement a distributed locking mechanism, not just using something provided by the database itself, makes the likelihood of bugs in this provider higher than in others. Hopefully it's implemented well, but if you find any issues, please report them. |
| 129 | +- For simplicity and testability, the distributed locking mechanism uses the date/time of the machine running the application to determine lock expiration. For this reason, when running multiple instances of the application, it's important that the clocks are in sync, otherwise there might be unexpected behavior. Other approaches might be considered, but at this point, it seemed an acceptable approach. |
| 130 | +- This implementation of the MongoDB polling provider was designed exclusively to be used with a primary database instance. It wasn't thought or tested to be used with sharded clusters. |
0 commit comments