AsyncMonolith

https://timmoth.github.io/AsyncMonolith/
Admin
AsyncMonolith is a lightweight .NET library that facilitates simple, durable, and asynchronous...
Tokens:8,713
Snippets:98
Trust Score:9.7
Update:1 month ago
Show doc for...
Tokens:
Raw
### 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<OrderCreated>
{
    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<ApplicationDbContext>(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<UserDeleted>
{
    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<ApplicationDbContext>((sp, options) =>
            {
                options.UseInMemoryDatabase(_inMemoryDatabaseName);
            }
        );

        services.AddRealAsyncMonolithBaseServices<ApplicationDbContext>(settings =>
            settings.RegisterTypesFromAssemblyContaining<Program>());
        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<ApplicationDbContext>();
            dbContext.Shipments.Add(model);
            await dbContext.SaveChangesAsync();
        }

        // When
        await Process<CancelShipmentConsumer, OrderCancelled>(new OrderCancelled()
        {
            OrderId = model.OrderId
        });

        // Then
        using (var scope = Services.CreateScope())
        {
            var dbContext = scope.ServiceProvider.GetRequiredService<ApplicationDbContext>();
            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<OrderCancelled>
{
    public override Task Consume(OrderCancelled message, CancellationToken cancellationToken)
    {
        ...
        shipment.Cancel();
        _dbContext.Shipments.Update(shipment);
        await _dbContext.SaveChangesAsync();
    }
}
```

```csharp
public class SendOrderCancelledEmail : BaseConsumer<OrderCancelled>
{
    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<ApplicationDbContext> options) : base(options)
        {
        }

        // Add Db Sets
        public DbSet<ConsumerMessage> ConsumerMessages { get; set; } = default!;
        public DbSet<PoisonedMessage> PoisonedMessages { get; set; } = default!;
        public DbSet<ScheduledMessage> 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<AlwaysOnSampler>();

                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<T>. 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<ValueSubmitted>
    {
        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<OrderCancelled>
{
    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));
```
Tokens:
Raw