Support for automatically append events from aggregates

[tim-cools]

When event sourcing is implemented, the appended events are often stored in a collection in a base Aggregate class. This is explained in this blog post: https://lostechies.com/jimmybogard/2014/05/13/a-better-domain-events-pattern/

Example: https://github.com/yreynhout/AggregateSource/blob/master/src/Core/AggregateSource.Content.ExplicitRouting/AggregateRootEntity.cs

This enables that the events can be appended to the store before a transactions is completed.
( See SaveChanged https://lostechies.com/jimmybogard/2014/05/13/a-better-domain-events-pattern/ )

This is difficult in the case of aggregates because the Marten session does not contain the aggregates. Is this something we can support too? We might hold the aggregates in a separated collection, so that we can enumerate them in a listener.

Or do you have something else in in mind to append events to the session in an automatic way?

[jeremydmiller]

I'm familiar with this pattern, and even used it on a project years ago using Raven. I'm not sure I'd say that this is a 1.0 requirement though.

We can handle fetching the aggregate inside of DocumentSession before it does the commit, and have that vary according to the aggregate type. We might wanna consider using the forthcoming partial updates in #18 to append events to the aggregate at capture time. That's what I ended up doing in our Raven project years ago.

[jvdvleuten]

@tim-cools shouldn't it be better to use a CommandHandler and (and maybe even decorate it) to handle this sort of stuff?

public class SignUpUserCommandHandler : ICommandHandler<SignUpUserCommand>
{
    private readonly IDomainRepository<User> _userDomainRepository;

    public SignUpUserCommandHandler(IDomainRepository<User> userDomainRepository)
    {
        _userDomainRepository = userDomainRepository;
    }

    public async Task HandleAsync(SignUpUserCommand command)
    {
        var userFromDb = await _userDomainRepository.GetAsync(command.UserId).ConfigureAwait(false);

        if (userFromDb != null)
        {
            throw new ArgumentException("UserId already exists", nameof(command.UserId));
        }

        var user = new User(command.UserId, command.EmailAddress);

        _userDomainRepository.Store(user);
    }
}

The User is implemented like this:

public class User : AggregateRoot
{
    public User() { }

    public User(Guid userId, string emailAddress) : base(userId)
    {
        RaiseEvent(new UserCreatedEvent { UserId = userId, EmailAddress = emailAddress });
    }

    public void Apply(UserCreatedEvent @event)
    {
        Id = @event.UserId;
    }
}

The IDomainRepository (or call it a AggregateRootRepository if you want) will handle saving of the events raised by the Aggregate.

public class MartenEventStoreDomainRepository<TAggregateRoot> : IDomainRepository<TAggregateRoot> 
    where TAggregateRoot : class, IAggregateRoot, new()
{
    private readonly MartenSession _session;

    public MartenEventStoreDomainRepository(MartenSession session)
    {
        _session = session;
    }

    public async Task<TAggregateRoot> GetAsync(Guid id)
    {
        var aggregator = new Aggregator<TAggregateRoot>();

        var stream = await _session.DocumentSession.Events.FetchStreamAsync(id).ConfigureAwait(false);

        if (!stream.Any())
        {
            return null;
        }

        var aggregate = aggregator.Build(stream);

        return aggregate;
    }

    public void Store(TAggregateRoot aggregateRoot)
    {
        object[] uncommittedEvents = aggregateRoot.GetUncommittedEvents().ToArray();

        _session.DocumentSession.Events.StartStream<TAggregateRoot>(aggregateRoot.Id, uncommittedEvents);
    }
}

You can decorate the CommandHandler to handle transactions, and after a transaction is really comitted, publish the comitted events or dispatch it to handlers.

[tim-cools]

That is similar to what I do now. I use a base class for my command handlers (https://github.com/tim-cools/RealTimeWeb.NET/blob/0c0713a3db8f792bee5546a724f63d9e1dbe5005/src/Soloco.RealTimeWeb.Common/Messages/CommandHandler.cs#L20) and my command handlers update the data (https://github.com/tim-cools/RealTimeWeb.NET/blob/master/src/Soloco.RealTimeWeb.Membership/RefreshTokens/Handlers/CreateRefreshTokenHandler.cs).

But I don't use the repository abstraction anymore as I think it is an unnecessary abstraction in most cases. I unit test my aggregates and use integration tests for my handlers mostly anyway...

You are right that I probably doesn't need to be in Marten as it is part of the infrastructure code. I was playing around with other implementations while preparing demo's for my talks next week in the #dddbe user group. And I didn't wanted to introduce handlers in the demo. Currently I explicitly call the Start and Append method: https://github.com/tim-cools/soloco-talks/blob/master/20160516-PolyglotPersistence/Soloco.Talks.PolyglotPersistence/G_EventSourcing/Example.cs#L37
And this is enough for the demo...

Disclaimer: both repo's are still WIP and not yet finished. But feedback always welcome ;-)

[bojanv91]

This is how I do it for dispatching the domain events which are stored in-memory of my entities with Marten:

    public class SimpleSessionListener : DocumentSessionListenerBase
    {
        private readonly IMediator _mediator;

        public SimpleSessionListener(IMediator mediator)
        {
            _mediator = mediator;
        }

        public override async Task BeforeSaveChangesAsync(IDocumentSession session, CancellationToken token)
        {
            var domainEvents = session
                .PendingChanges
                .AllChangedFor<AggregateRoot>()
                .SelectMany(entity => entity.UncommitedEvents);

            foreach (var evt in domainEvents)
            {
                await _mediator.PublishAsync(evt);
            }
        }
    }