### Install AsyncMonolith NuGet packages Source: https://timmoth.github.io/AsyncMonolith/quickstart Choose one database-specific package to install. The packages are AsyncMonolith.Ef, AsyncMonolith.MySql, AsyncMonolith.MsSql, AsyncMonolith.PostgreSql, and AsyncMonolith.MariaDb. ```bash // Pick one dotnet add package AsyncMonolith.Ef dotnet add package AsyncMonolith.MySql dotnet add package AsyncMonolith.MsSql dotnet add package AsyncMonolith.PostgreSql dotnet add package AsyncMonolith.MariaDb ``` -------------------------------- ### AsyncMonolith consumer example Source: https://timmoth.github.io/AsyncMonolith/posts/transactional-outbox Shows a consumer class that handles the OrderCreated event, with a placeholder for creating the order. ```csharp public class CreateShipment : BaseConsumer { public override Task Consume(OrderCreated message, CancellationToken cancellationToken) { ... // Create Order } } ``` -------------------------------- ### Register AsyncMonolith services and configure settings Source: https://timmoth.github.io/AsyncMonolith/quickstart Register required services and configure AsyncMonolith settings. The example uses AddPostgreSqlAsyncMonolith, but you can choose one of the other provider-specific methods (AddEfAsyncMonolith, AddMsSqlAsyncMonolith, AddMySqlAsyncMonolith, AddPostgreSqlAsyncMonolith, AddMariaDbAsyncMonolith). Settings include retry delays, processor counts, and timeouts. ```csharp // Register required services builder.Services.AddLogging(); builder.Services.AddSingleton(TimeProvider.System); // Register AsyncMonolith using either: // services.AddEfAsyncMonolith // services.AddMsSqlAsyncMonolith // services.AddMySqlAsyncMonolith // services.AddPostgreSqlAsyncMonolith // services.AddMariaDbAsyncMonolith builder.Services.AddPostgreSqlAsyncMonolith(settings => { settings.RegisterTypesFromAssembly(Assembly.GetExecutingAssembly()); settings.AttemptDelay = 10, // Seconds before a failed message is retried settings.MaxAttempts = 5, // Number of times a failed message is retried settings.ProcessorMinDelay = 10, // Minimum millisecond delay before the next batch is processed settings.ProcessorMaxDelay = 1000, // Maximum millisecond delay before the next batch is processed settings.ProcessorBatchSize = 5, // The number of messages to process in a single batch settings.ConsumerMessageProcessorCount = 2, // The number of concurrent consumer message processors to run in each app instance settings.ScheduledMessageProcessorCount = 1, // The number of concurrent scheduled message processors to run in each app instance settings.DefaultConsumerTimeout = 10 // The default number of seconds before a consumer will timeout }); ``` -------------------------------- ### AsyncMonolith producer service example Source: https://timmoth.github.io/AsyncMonolith/posts/transactional-outbox Shows how AsyncMonolith simplifies the outbox pattern by using a producer service to add the event to the outbox, then saving changes. ```csharp // Create order _dbContext.Orders.Add(newOrder); // Insert event into outbox await _producerService.Produce(new OrderCreated() { OrderId = order.Id }); await _dbContext.SaveChangesAsync(); ``` -------------------------------- ### Consumer example with timeout and attempts attributes Source: https://timmoth.github.io/AsyncMonolith/guides/consuming-messages Example of a consumer class that handles a UserDeleted message. It uses the [ConsumerTimeout(5)] attribute to set a 5-second timeout and [ConsumerAttempts(1)] to move the message to the poisoned table after one failed attempt. The consumer injects ApplicationDbContext and overrides the Consume method to perform database operations. ```csharp [ConsumerTimeout(5)] // Consumer timeouts after 5 seconds [ConsumerAttempts(1)] // Consumer messages moved to poisoned table after 1 failed attempt public class DeleteUsersPosts : BaseConsumer { private readonly ApplicationDbContext _dbContext; public DeleteUsersPosts(ApplicationDbContext dbContext) { _dbContext = dbContext; } public override Task Consume(UserDeleted message, CancellationToken cancellationToken) { ... await _dbContext.SaveChangesAsync(cancellationToken); } } ``` -------------------------------- ### ConsumerTestBase Example: CancelShipmentTests Source: https://timmoth.github.io/AsyncMonolith/tests Shows how to write a consumer test using ConsumerTestBase. It sets up an in-memory database and real AsyncMonolith base services, then processes an OrderCancelled event and asserts the shipment status is updated to 'cancelled'. ```csharp public class CancelShipmentTests : ConsumerTestBase { public CancelShipmentTests(ITestOutputHelper testOutputHelper) : base(testOutputHelper) { } private readonly string _inMemoryDatabaseName = Guid.NewGuid().ToString(); protected override Task Setup(IServiceCollection services) { services.AddDbContext((sp, options) => { options.UseInMemoryDatabase(_inMemoryDatabaseName); } ); services.AddRealAsyncMonolithBaseServices(settings => settings.RegisterTypesFromAssemblyContaining()); return Task.CompletedTask; } [Fact] public async Task OrderCancelled_Sets_Shipment_Status_To_Cancelled() { // Given var model = new Shipment { Id = "test-shipment-id", OrderId = "test-order-id", Status = "pending", }; using (var scope = Services.CreateScope()) { var dbContext = scope.ServiceProvider.GetRequiredService(); dbContext.Shipments.Add(model); await dbContext.SaveChangesAsync(); } // When await Process(new OrderCancelled() { OrderId = model.OrderId }); // Then using (var scope = Services.CreateScope()) { var dbContext = scope.ServiceProvider.GetRequiredService(); var shipment = await dbContext.Shipments.FirstOrDefaultAsync(c => c.Id == model.Id); shipment.Status.Should().Be("cancelled"); await dbContext.SaveChangesAsync(); } } } ``` -------------------------------- ### AsyncMonolith Consumers for OrderCancelled Source: https://timmoth.github.io/AsyncMonolith/posts/mediator Example consumers for the OrderCancelled message using AsyncMonolith's BaseConsumer class. ```csharp public class CancelShipment : BaseConsumer { public override Task Consume(OrderCancelled message, CancellationToken cancellationToken) { ... shipment.Cancel(); _dbContext.Shipments.Update(shipment); await _dbContext.SaveChangesAsync(); } } ``` ```csharp public class SendOrderCancelledEmail : BaseConsumer { public override Task Consume(OrderCancelled message, CancellationToken cancellationToken) { ... // Send order cancelled email } } ``` -------------------------------- ### Mediator Handlers for OrderCancelled Source: https://timmoth.github.io/AsyncMonolith/posts/mediator Example handlers for the OrderCancelled event: CancelShipmentHandler and CancellationEmailHandler. ```csharp public class CancelShipmentHandler { public async Task Handle(OrderCancelled orderCancelled) { ... shipment.Cancel(); _dbContext.Shipments.Update(shipment); await _dbContext.SaveChangesAsync(); } } ``` ```csharp public class CancellationEmailHandler { public async Task Handle(OrderCancelled orderCancelled) { ... // Send order cancelled email } } ``` -------------------------------- ### Setup ApplicationDbContext with AsyncMonolith Source: https://timmoth.github.io/AsyncMonolith/quickstart Define a DbContext that includes the required DbSets (ConsumerMessage, PoisonedMessage, ScheduledMessage) and calls ConfigureAsyncMonolith() in OnModelCreating. This is required for AsyncMonolith to work with EF Core. ```csharp public class ApplicationDbContext : DbContext { public ApplicationDbContext(DbContextOptions options) : base(options) { } // Add Db Sets public DbSet ConsumerMessages { get; set; } = default!; public DbSet PoisonedMessages { get; set; } = default!; public DbSet ScheduledMessages { get; set; } = default!; // Configure the ModelBuilder protected override void OnModelCreating(ModelBuilder modelBuilder) { modelBuilder.ConfigureAsyncMonolith(); base.OnModelCreating(modelBuilder); } } ``` -------------------------------- ### Create a branch Source: https://timmoth.github.io/AsyncMonolith/contributing Create and switch to a new branch for your changes. ```bash git checkout -b my-feature-branch ``` -------------------------------- ### Create order then publish event (fails) Source: https://timmoth.github.io/AsyncMonolith/posts/transactional-outbox Illustrates the failure scenario where the order is saved but publishing the event fails, leading to inconsistency. ```csharp // Create order _dbContext.Orders.Add(newOrder); await _dbContext.SaveChangesAsync(); // Dispatch event channel.BasicPublish(...); \\ This fails ``` -------------------------------- ### Publish event then create order (fails) Source: https://timmoth.github.io/AsyncMonolith/posts/transactional-outbox Illustrates the failure scenario where the event is published but the order creation fails, leading to inconsistency. ```csharp // Dispatch event channel.BasicPublish(...); // Create order _dbContext.Orders.Add(newOrder); await _dbContext.SaveChangesAsync(); \\ This fails ``` -------------------------------- ### Clone your fork Source: https://timmoth.github.io/AsyncMonolith/contributing Clone the forked repository to your local machine using the provided URL. ```bash git clone https://github.com/Timmoth/AsyncMonolith.git ``` -------------------------------- ### Configure OpenTelemetry Tracing with AsyncMonolith Source: https://timmoth.github.io/AsyncMonolith/guides/opentelemetry Add AsyncMonolithInstrumentation.ActivitySourceName to your OpenTelemetry tracing configuration to enable consumer and scheduled processor tracing. ```csharp builder.Services.AddOpenTelemetry() .WithTracing(x => { if (builder.Environment.IsDevelopment()) x.SetSampler(); x.AddSource(AsyncMonolithInstrumentation.ActivitySourceName); x.AddConsoleExporter(); }) .ConfigureResource(c => c.AddService("async_monolith.demo").Build()); ``` -------------------------------- ### Commit your changes Source: https://timmoth.github.io/AsyncMonolith/contributing Commit your changes with a descriptive message. ```bash git commit -m 'Add some feature' ``` -------------------------------- ### Produce and schedule messages Source: https://timmoth.github.io/AsyncMonolith/quickstart Inject IProducerService and IScheduleService to produce messages immediately, with a delay, or with a deduplication key. Schedule recurring messages using a cron expression and timezone. Save changes to persist the scheduled messages. ```csharp // Inject services private readonly IProducerService _producerService; private readonly IScheduleService _scheduleService; // Produce a message to be processed immediately await _producerService.Produce(new ValueSubmitted() { Value = newValue }); // Produce a message to be processed in 10 seconds await _producerService.Produce(new ValueSubmitted() { Value = newValue }, 10); // Produce a message to be processed in 10 seconds, but only once for a given userId await _producerService.Produce(new ValueSubmitted() { Value = newValue }, 10, $"user:{userId}"); // Publish a message every Monday at 12pm (UTC) with a tag that can be used to modify / delete related scheduled messages. _scheduleService.Schedule(new ValueSubmitted { Value = newValue }, "0 0 12 * * MON", "UTC", "id:{id}"); // Save changes await _dbContext.SaveChangesAsync(cancellationToken); ``` -------------------------------- ### Push to the branch Source: https://timmoth.github.io/AsyncMonolith/contributing Push your branch to the remote repository. ```bash git push origin my-feature-branch ``` -------------------------------- ### Add order and shipment in one transaction Source: https://timmoth.github.io/AsyncMonolith/posts/transactional-outbox Shows adding an order and a shipment to the database context and saving changes in a single transaction, ensuring both succeed or fail together. ```csharp _dbContext.Orders.Add(newOrder); _dbContext.Shipments.Add(newShipment); await _dbContext.SaveChangesAsync(); ``` -------------------------------- ### Transactional outbox: add order and outbox entry Source: https://timmoth.github.io/AsyncMonolith/posts/transactional-outbox Shows the core outbox pattern: adding the order and an outbox entry in the same transaction, ensuring both are committed together. ```csharp // Create order _dbContext.Orders.Add(newOrder); // Insert event into outbox _dbContext.Outbox.Add(orderCreated) await _dbContext.SaveChangesAsync(); ``` -------------------------------- ### Create a consumer payload and consumer Source: https://timmoth.github.io/AsyncMonolith/quickstart Define a consumer payload implementing IConsumerPayload and a consumer class inheriting BaseConsumer. The consumer uses attributes like [ConsumerTimeout] and [ConsumerAttempts] to control timeout and retry behavior. The Consume method processes the message. ```csharp // Define Consumer Payload public class ValueSubmitted : IConsumerPayload { [JsonPropertyName("value")] public required double Value { get; set; } } // Define Consumer [ConsumerTimeout(5)] // Consumer timeouts after 5 seconds [ConsumerAttempts(1)] // Consumer messages moved to poisoned table after 1 failed attempt public class ValueSubmittedConsumer : BaseConsumer { private readonly ApplicationDbContext _dbContext; private readonly IProducerService _producerService; public ValueSubmittedConsumer(ApplicationDbContext dbContext, IProducerService producerService) { _dbContext = dbContext; _producerService = producerService; } public override Task Consume(ValueSubmitted message, CancellationToken cancellationToken) { ... await _dbContext.SaveChangesAsync(cancellationToken); } } ``` -------------------------------- ### AsyncMonolith Producer for OrderCancelled Source: https://timmoth.github.io/AsyncMonolith/posts/mediator Shows how to produce an OrderCancelled message using AsyncMonolith's producer service. ```csharp order.Cancel(); _dbContext.Orders.Update(order); await _producerService.Produce(new OrderCancelled() { OrderId = order.Id }); await _dbContext.SaveChangesAsync(); ``` -------------------------------- ### Produce a delayed message with a 60-second delay Source: https://timmoth.github.io/AsyncMonolith/guides/producing-messages Pass a delay in seconds as the second argument to Produce to schedule the message for later consumption. ```csharp order.Cancel(); _dbContext.Orders.Update(order); await _producerService.Produce(new OrderCancelled() { OrderId = order.Id, CancelledAt = _timeProvider.UtcNow() }, 60); // Save changes await _dbContext.SaveChangesAsync(cancellationToken); ``` -------------------------------- ### Define a consumer payload class implementing IConsumerPayload Source: https://timmoth.github.io/AsyncMonolith/guides/producing-messages Define a class that implements IConsumerPayload and is serializable by System.Text.Json.JsonSerializer. Keep the payload small as it is stored serialized in the database. ```csharp public class OrderCancelled : IConsumerPayload { [JsonPropertyName("order_id")] public string OrderId { get; set; } [JsonPropertyName("cancelled_at")] public DateTimeOffset CancelledAt { get; set; } } ``` -------------------------------- ### Produce an immediate message with IProducerService Source: https://timmoth.github.io/AsyncMonolith/guides/producing-messages Inject IProducerService and call Produce with the payload. The message is produced transactionally with domain changes when SaveChangesAsync is called. For MySql, MsSql, MariaDb, or PostgreSQL packages, wrap changes in a transaction. ```csharp order.Cancel(); _dbContext.Orders.Update(order); await _producerService.Produce(new OrderCancelled() { OrderId = order.Id, CancelledAt = _timeProvider.UtcNow() }); // Save changes await _dbContext.SaveChangesAsync(cancellationToken); ``` -------------------------------- ### Wrap produce and SaveChanges in an explicit transaction for MySql/MsSql/MariaDb/PostgreSql Source: https://timmoth.github.io/AsyncMonolith/guides/producing-messages For MySql, MsSql, MariaDb, or PostgreSQL packages, the Produce method uses ExecuteSqlRawAsync, so wrap all changes in an explicit transaction to ensure messages are inserted transactionally with domain changes. ```csharp await using var dbContextTransaction = await _dbContext.Database.BeginTransactionAsync(cancellationToken); order.Cancel(); _dbContext.Orders.Update(order); await _producerService.Produce(new OrderCancelled() { OrderId = order.Id, CancelledAt = _timeProvider.UtcNow() }); await _dbContext.SaveChangesAsync(cancellationToken); await dbContextTransaction.CommitAsync(cancellationToken); ``` -------------------------------- ### AsyncMonolith producer for transactional event emission Source: https://timmoth.github.io/AsyncMonolith/posts/idempotency Uses AsyncMonolith's producer service to emit an OrderCancelled event transactionally with the domain changes, avoiding the early-return issue. ```csharp if(order.Status == "cancelled"){ return; } order.Status = "cancelled"; await _producerService.Produce(new OrderCancelled() { OrderId = order.Id }); await _dbContext.SaveChangesAsync(); ``` -------------------------------- ### Idempotent order cancellation handler with early return Source: https://timmoth.github.io/AsyncMonolith/posts/idempotency Adds an early return if the order is already cancelled, preventing duplicate emails. However, if the email service throws, the retry will exit early because the status is already set. ```csharp if(order.Status == "cancelled"){ return; } order.Status = "cancelled"; await _dbContext.SaveChangesAsync(); await _emailService.SendCancellationEmail(order.Id); ``` -------------------------------- ### Tightly Coupled Order Cancellation Source: https://timmoth.github.io/AsyncMonolith/posts/mediator Shows the initial approach where OrderService directly calls shipment and email services, leading to tight coupling. ```csharp order.Cancel(); _dbContext.Orders.Update(order); await _dbContext.SaveChangesAsync(); _shipmentService.CancelShipment(order.ShipmentId); _emailService.SendCancellationEmail(order.Id); ``` -------------------------------- ### AsyncMonolith consumer for sending cancellation email Source: https://timmoth.github.io/AsyncMonolith/posts/idempotency Consumer class that handles the OrderCancelled event and sends the email. This is part of the AsyncMonolith solution. ```csharp public class SendOrderCancelledEmail : BaseConsumer { public override Task Consume(OrderCancelled message, CancellationToken cancellationToken) { ... // Send order cancelled email } } ``` -------------------------------- ### Produce a deduplicated message with insertId Source: https://timmoth.github.io/AsyncMonolith/guides/producing-messages Pass a delay and an insertId to Produce to ensure only one message with the same consumer type and insertId is pending at a time. Useful for aggregating events like page views. ```csharp pageView.Increment(); _dbContext.PageViews.Update(pageView); await _producerService.Produce(new PageViewed() { PageId = pageView.Id, }, 10, $"page_id:{pageView.Id}"); // Save changes await _dbContext.SaveChangesAsync(cancellationToken); ``` -------------------------------- ### Non-idempotent order cancellation handler Source: https://timmoth.github.io/AsyncMonolith/posts/idempotency Sets order status to cancelled, saves changes, and sends an email. If the event is processed twice, a duplicate email is sent. ```csharp order.Status = "cancelled"; await _dbContext.SaveChangesAsync(); await _emailService.SendCancellationEmail(order.Id); ``` -------------------------------- ### Schedule a message with IScheduleService Source: https://timmoth.github.io/AsyncMonolith/guides/scheduling-messages Inject IScheduleService to schedule a message with a cron expression, timezone, and tag. Ensure SaveChangesAsync is called on the DbContext to persist the schedule transactionally. ```csharp // Publish 'CacheRefreshScheduled' every Monday at 12pm (UTC) with a tag that can be used to modify / delete related scheduled messages. _scheduleService.Schedule(new CacheRefreshScheduled { Id = id }, "0 0 12 * * MON", "UTC", "id:{id}"); await _dbContext.SaveChangesAsync(cancellationToken); ``` -------------------------------- ### Mediator-Based Order Cancellation Source: https://timmoth.github.io/AsyncMonolith/posts/mediator Shows the refactored OrderService that sends an OrderCancelled event to the mediator instead of directly calling other services. ```csharp order.Cancel(); _dbContext.Orders.Update(order); await _dbContext.SaveChangesAsync(); _mediator.Send(new OrderCancelled(order.Id)); ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.