AsyncMonolith

AsyncMonolith is a lightweight library that facilitates simple, durable and asynchronous messaging in dotnet apps.

Overview:

• Speed up your API by offloading tasks to a background worker.
• Distribute workload amongst multiple app instances.
• Execute tasks at specific times or after a delay.
• Execute tasks at regular intervals.
• Simplify complex processes by building out an event-driven architecture.
• Decouple your services by utilizing the Mediator pattern.
• Improve your application's resiliency by utilizing the Transactional Outbox pattern.
• Improve your application's resiliency by utilizing automatic retries.
• Keep your infrastructure simple without using a message broker.
• Simplify testability.

Note

Despite its name, AsyncMonolith can be used within a microservices architecture. The only requirement is that the producers and consumers share the same database i.e messaging within a single project. However, it is not suitable for passing messages between different projects in a microservices architecture, as microservices should not share the same database.

Find out more 🤔

• Overview ✅
• Quick start ▶️
• Internals 🧠
• Releases 📒
• Warnings ⚠️
• Support 🛟
• Demo App ✨
• Contributing 🙏
• Tests 🐞

Check out this video from Derek Comartin, for some thoughts on using a database as a queue

Guides

• Producing messages
• Scheduling messages
• Consuming messages
• Changing messages
• Open Telemetry

Posts

• What is the Transactional Outbox?
• What is the Mediator pattern?
• What is Idempotency?

Quick Demo

Producing and scheduling messages

  // Publish 'UserDeleted' to be processed in 60 seconds
  await _producerService.Produce(new UserDeleted()
  {
    Id = id
  }, 60);
  
 // 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);

Consuming messages

[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 ValueSubmittedConsumer(ApplicationDbContext dbContext)
    {
        _dbContext = dbContext;
    }

    public override Task Consume(UserDeleted message, CancellationToken cancellationToken)
    {
        ...
		await _dbContext.SaveChangesAsync(cancellationToken);
    }
}

Collaboration 🙏

Like the idea and want to get involved? Check out the open issues or shoot me a message if you've got any ideas / feedback!

Support 🛟

Need help? Ping me on linkedin and I'd be more then happy to jump on a call to debug, help configure or answer any questions.

Discord

Come say hello on the Discord server

Support the project 🤝

• 🌟 Star this repository: It means a lot to me and helps with exposure.
• 🪲 Report bugs: Report any bugs you find by creating an issue.
• 📝 Contribute: Read the contribution guide then pick up or create an issue.