Domain Events with .NET

Introduction

Domain Events is a pattern and concept made popular by Domain Driven Design. A Domain Event is something that happened on your system and you want interested-parties to know about. It is a way to publish information in a decoupled, usually asynchronous, way. A publisher publishes (raises) an event, and interested subscribers receive a notification and act accordingly. A subscriber can also be a publisher, of course.

In this post I'm going to present a Domain Events library for .NET. It's not the first time that I write about decoupling messages and subscribers in an application, I implemented a few years ago the Postal.NET library which does something similar.

Mind you, there are many implementations out there, MediatR coming first to my mind, but this is mine. I obviously got some ideas from other implementations, but mine is substantially different from all the others.

Concepts

First, a few concepts:

  • An event is just some class that implements IDomainEvent
  • A publisher is where you publish messages (events) to
  • A subscriber is where you register your intent to receive notifications about a specific event type
  • A mediator is what links together publishers and subscribers
  • A subscription is a concrete registration of an handler to an event
  • A dispatcher is what actually calls the subscription handlers
  • An interceptor is an handler that can perform operations on an event before and after it is sent to the subscribers
The declarations, first, a marker interface for domain events, IDomainEvent:
public interface IDomainEvent { }

As you can see, there's nothing much to it, it is merely used to mark some class as a domain event. Of course, your event classes can be arbitrarily complex.

Now, the publisher, IEventsPublisher:

public interface IEventsPublisher
{
    Task Publish<TEvent>(TEvent @event, CancellationToken cancellationToken = default) where TEvent : IDomainEvent;
}
Still quite simple, it merely declares a generic method, Publish, where the generic parameter is a domain event. The Publish method is declared as asynchronous.

The subscriber is IEventsSubscriber:

public interface IEventsSubscriber
{
    Subscription Subscribe<TEvent>(Action<TEvent> action) where TEvent : IDomainEvent;
}

Again, simple enough: the Subscribe generic method takes a generic action delegate (Action<T>) and returns a Subscription instance. The action will be called whenever an event of type TEvent is published.

The mediator, IEventsMediator:

public interface IEventsMediator
{
    Task Publish<TEvent>(TEvent @event, CancellationToken cancellationToken = default) where TEvent : IDomainEvent;
    Subscription Subscribe<TEvent>(Action<TEvent> action) where TEvent : IDomainEvent;
    bool Unsubscribe(Subscription subscription);
    void AddInterceptor(IDomainEventInterceptor interceptor);
}

As you can see, the IEventsMediator interface has some methods coming from IEventsSubscriber and IEventsPublisher, plus a couple more, to unsubscribe and to register an interceptor. It will act as a mediator between the different classes. However, you should not touch these methods directly, except for AddInterceptor.

A subscription is implemented by class Subscription, which is disposable:

public class Subscription : IDisposable
{
    //details not included here
}

The reason this class is disposable is, if we want to cancel an existing subscription, we need only call its Dispose method. It is an opaque reference, you needn't worry about its contents.

Approaching the end, here is the IEventsDispatcher interface:

public interface IEventsDispatcher
{
    Task Dispatch<TEvent>(TEvent @event, IEnumerable<Subscription> subscriptions, CancellationToken cancellationToken = default);
}

This is what is actually responsible for calling the event handlers, and the framework includes many implementations:

  • SequentialEventsDispatcher: calls all event dispatchers synchronously and sequentially
  • TaskEventsDispatcher: calls event handlers asynchronously and sequentially; this is the default
  • ParallelEventsDispatcher: calls event handlers in parallel
  • ThreadEventsDispatcher: uses threads to call event handlers
  • ThreadPoolEventsDispatcher: uses threads from the managed thread pool to call event handlers

And, finally, the interceptor, IDomainEventInterceptor:

public interface IDomainEventInterceptor
{
    Task BeforePublish(IDomainEvent @event, CancellationToken cancellationToken = default);
    Task AfterPublish(IDomainEvent @event, CancellationToken cancellationToken = default);
}

There's also an abstract class for making it easier for us:

public abstract class DomainEventInterceptor : IDomainEventInterceptor
{
    public virtual Task BeforePublish(IDomainEvent @event, CancellationToken cancellationToken = default) { }
    public virtual Task AfterPublish(IDomainEvent @event, CancellationToken cancellationToken = default) { }
}

You can inherit from DomainEventInterceptor and just implement the method you want. As you can see, the IDomainEventInterceptor class is designed to intercept all events published in the system, this is why it takes a IDomainEvent parameter, which is the interface all events must adhere to.

Usage

We must first register the domain events framework to the dependency injection (DI):

builder.Services.AddDomainEvents();

And then we can get the interfaces from it:

var subscriber = serviceProvider.GetRequiredService<IEventsSubscriber>();
var publisher = serviceProvider.GetRequiredService<IEventsPublisher>();

Of course, the most usual way to obtain these instances is from DI, for example, by injecting into a controller action:

public async Task<IActionResult> Publish([FromServices] IEventsPublisher publisher, CancellationToken cancellationToken)
{
    await publisher.Publish(new SomeEvent(), cancellationToken);
    //...
}

Of course, in order for this to do something productive, we must first register an event handler:

var subscription = subscriber.Subscribe<SomeEvent>(evt =>
{
    //do something with the event
});

Now, every time we publish an event (Publish), the register handlers will be called. When we are done with it, and don't want to receive more event notifications, we can dispose of the subscription:

subscription.Dispose();

And, if for some reason, we want to add an interceptor to the system, we can do so through the IEventsMediator interface: as I said, it is actually the only time we'll need to access it directly:

mediator.AddInterceptor(new SomeEventInterceptor());

This method takes an interceptor instance, later on we'll see an alternative.

Advanced Usages

Typed Subscriptions

Sometimes things can be more complex. For example, say we want to have a proper class for handling subscriptions. For that, we have the ISubscription<TEvent> interface:

public interface ISubscription<TEvent> where TEvent : IDomainEvent

{

Task OnEvent(TEvent @event, Subscription subscription);

}

The ISubscription<TEvent> interface specifies a single method, OnEvent, that receives the event being published and also the subscription to which the subscription refers to. It is safe to cancel the subscription from this method, the way I've shown before. This class is instantiated by DI, meaning, you can even inject other types to it.

And here is how to add a subscription:

builder.Services.AddDomainEvents()
    .AddSubscription<SomeEvent, SomeSubscription>();

Where SomeSubscription is a class that implements ISubscription<SomeEvent>. The AddSubscription method can be called on the return of AddDomainEvents.

Typed Interceptors

We've seen how the interceptors work. Essentially, they get called whenever an event is published, no matter what it's type. If we want to, however, we can have a typed interceptor that is called only for a specific event type:

public interface IDomainEventInterceptor<TEvent> where TEvent : IDomainEvent
{
    Task OnEvent(TEvent event, Subscription subscription);
}

The way to register it is also by calling AddInterceptor following a call to AddDomainEvents:

builder.Services.AddDomainEvents()
    .AddInterceptor<SomeEvent, SomeInterceptor>();

Where SomeInterceptor implements IDomainEventInterceptor<SomeEvent>. I am not including the code for SomeEvent or SomeInterceptor because they are not relevant here. It is important to know that SomeInterceptor is instantiated by the DI framework, 

We can also register a vanilla (non-generic) IDomainInterceptor this way:

builder.Services.AddDomainEvents()
    .AddInterceptor<SomeInterceptor>();

Notice that in this case, this AddInterceptor extension method only takes a single parameter, that of a type that implements IDomainInterceptor.

Supplying Your Own Implementations

It is possible to supply our own implementation of one or more of the following types:

  • IEventsPublisher
  • IEventsSubscriber
  • IEventsMediator
  • IEventsDispatcher

The way to do it is by registering the type on the DI engine before calling AddDomainEvents / AddDomainEventsFromAssembly. Of course, you should know what you're doing before changing any of these!

Auto Wiring Types

As an alternative to adding all subscriptions and interceptors manually, there is an extension method that automatically finds all relevant types and registers them, it's AddDomainEventsFromAssembly:

builder.Services.AddDomainEventsFromAssembly(typeof(Program).Assembly);

Specifically, it registers all non-abstract and non-generic occurrences of these types:

  • ISubscription<TEvent>
  • IDomainEventInterceptor<TEvent>
  • IDomainEventInterceptor

A final word on this: you can combine AddDomainEventsFromAssembly with AddInterceptor and AddSubscription, if, for example the types are located on a different assembly.

Domain Events Options

The DomainEventsOptions class can be used to pass options to the domain events framework. As of now, it contains a single property:

public class DomainEventsOptions
{
    public bool FailOnNoSubscribers { get; set; }
}
The FailOnNoSubscribers property tells the framework to throw an exception if there are no registered event handlers for a specific event. The way to set this option is through an overload to AddDomainEvents / AddDomainEventsFromAssembly method:
 builder.Services.AddDomainEvents(options =>
{
    options.FailOnNoSubscribers = true;
});

Conclusion

And this is it. You can see for yourself by grabbing the NetDomainEvents Nuget package or by looking at the GitHub repo. As always, keen to hear your thoughts, questions, criticism, etc!

Source Code: https://github.com/rjperes/DomainEvents

Nuget: https://www.nuget.org/packages/NetDomainEvents

Location: Coimbra, Portugal

Comments

Post a Comment

Popular posts from this blog

ASP.NET Core Pitfalls – Posting a String

This is another post on my ASP.NET Core pitfalls series . It is actually related to this one  ASP.NET Core Pitfalls – Null Models in Post Requests . What happens if you try to submit a string containing JSON as a POST? I mean, you always submit strings, but this time you don't want for it to be parsed into some class. Like this: [HttpPost] public IActionResult PostString([FromBody] string json) { ... } You may be surprised to find out that this fails with a HTTP 415 Unsupported Media Type . Now, you might be tempted to add an [Consumes] attribute to it, such as: [HttpPost] [Consumes("application/json")] public IActionResult PostString([FromBody] string json) { ... } And this will actually work! The problem is, or could be, that it requires the sender to send the appropriate Content-Type header (" application/json "). Another, better, solution is to just read the string from the from the request stream: [HttpPost] public IActionResult PostString() { using v
Read more

Domain Events with .NET - New Features

Introduction You may remember my post on the Domain Events framework I created, a few days ago. Well, I've been working on it, cleaning things a bit, correcting a few bugs, and I introduced two new features/concept:  Event interceptors can now return a boolean from the BeforePublish event in order to cancel the publication of the event Event transformers Changes to Event Interceptors The changes to the IDomainEventInterceptor / IDomainEventInterceptor<TEvent> interface look like this: public interface IDomainEventInterceptor { ValueTask<bool> BeforePublish(IDomainEvent @event, CancellationToken cancellationToken = default); Task AfterPublish(IDomainEvent @event, CancellationToken cancellationToken = default); } public interface IDomainEventInterceptor<TEvent> where TEvent : IDomainEvent { ValueTask<bool> BeforePublish(TEvent @event, CancellationToken cancellationToken = default); Task AfterPublish(TEvent @event, CancellationToken cancellat
Read more