Publishing Events to the Bus

When an Aggregate Root publishes an Event, what do we do with it? Looking back at why Domain Events are so important - they are how we achieve consistency with parties that are outside of transactional consistency boundaries.

There are consistency boundaries in various places, one primary boundary being the one that surrounds an Aggregate Root. In Implementing Domain Driven Design, Vaughn Vernon gives us a rule of thumb when it comes to Aggregate consistency (through discussion with Eric Evans):

“When examining the use case (or story), ask whether it’s the job of the user executing the use case to make the data consistent. If it is, try to make it transactionally consistent, but only by adhering to the other rules of Aggregates. If it another user’s job, or the job of the system, allow it to be eventually consistent.” (Vernon)

We should always try to focus on the true business invariants that require different kinds of consistency rather than following a technical preference. Does Nick require his inventory to be updated as soon as he ships a batch of roasted beans? Is a customer allowed to reserve more than what Nick has planned to roast on a given day? These are the kinds of questions we should ask when we’re deciding whether to be transactionally or eventually consistent.

The Bus plays an important role in eventual consistency across both inter-Aggregate and Write/Read stack boundaries. Let’s talk about how we keep our CQRS Write and Read Models eventually consistent using Domain Events and the Bus.

The Bus’s Message Router allows us to associate multiple Event Handlers to a given Event type, just like we have done with Command Handlers to Command types so far.

FakeBus implements IEventPublisher, who has a Publish<T>(T @event) method:

public interface IEventPublisher {
    void Publish<T>(T @event) where T : Event;
}

In FakeBus, the implementation of `Publish(T @event) looks like this:

public void Publish<T>(T @event) where T : Event {
    List<Action<Message>> handlers;

    if (!_routes.TryGetValue(@event.GetType(), out handlers)) return;

    foreach (var handler in handlers) {
        Task.Run(() => handler(@event));
    }
}

The only difference between the IEventPublisher.Publish<T> method and the ICommandSender.Send<T> method is that Publish<T> can dispatch an Event to multiple Event Handler’s Handle(Event @event) methods. We’ll write a quick test that the Bus can register Event Handlers for Events:

public class When_publishing_an_event_to_the_fakebus_with_a_single_event_handler_registered : EventSpecification<TestAggregateRoot, TestEvent> {

    private readonly FakeBus _bus;
    private TestEventHandler _handler;
    private TestEvent _event;
    public When_publishing_an_event_to_the_fakebus_with_a_single_event_handler_registered() {
        _bus = new FakeBus();
        _bus.RegisterHandler<TestEvent>(_handler.Handle);
    }

    protected override TestEvent When() {
        _event = new TestEvent();
        return _event;
    }

    protected override IEventHandler<TestEvent> EventHandler() {
        _handler = new TestEventHandler();
        return _handler;
    }

    [Then]
    public void Then_handle_method_is_invoked_on_command_handler() {
        _handler.HandledEventIds.First().Should().Be(_event.Id);
    }
}

Right now, we’re not so much concerned with publishing Events to various Aggregates and Bounded Contexts. We will. For now, we’ll focus on how our Read Model consumes published Events in a process known as Projection. Stay tuned for the next post to find out more!

previous…

I will be releasing most of this content only to subscribers. Make sure you sign up by clicking the big red button!

Related Posts: