Introduction

In the previous post I showed how we can use Yeoman to scaffold a new ASP.NET vNext application on Ubuntu. I then implemented a simple RESTful API and also started to implement the basics of a domain model. In this part I will improve the domain model and also implement a repository for GetEventStore (GES) to save the events generated by our domain aggregates.

The code accompanying this series of posts can be found here on GitHub.

Implementing the GES repository

In the previous post we have defined the interface of the repository that we are using in our application. Just as a reminder it looks like this

Let’s now write an implementation for GES v3.0.1. First we need to add the client library for .NET which is available as Nuget package to our solution. Later on we will also need the Newtonsoft.JSON library in our code. Thus we need to modify the project.json file and add the following lines to the dependencies collection

All access to GES happens via a GES connection object. We will inject this connection into our repository. We will also inject an aggregate factory into the repository. The repository will use this aggregate factory to create new instances of aggregates. Thus we start with the following code

Next we are going to implement the Get method

In this method we essentially read all event events of a stream representing the particular aggregate instance that we want to re-hydrate. On line 6 we determine the stream name from the type of the requested aggregate and the given id. Thus if the requested aggregate type is RecipeAggregate and the id is equal to say 11 then the stream name will be RecipeAggregate-11. We need to read all events of the given stream to be able to reconstruct the current state of the requested aggregate. Thus we loop and read events in batches of 200 until there are no more left. Then, on line 15 we use the aggregate factory to create a new instance and pass it the collection of events. Finally we return the aggregate to the caller.

Please note lines 9-11; the client library of GES only offers asynchronous methods. Since we want to read all events synchronously we use the .Result property of the Task returned by the ReadStreamEventsForwardAsync method. This will block execution until the asynchronous operation has been completed.

With each read operation we get a slice of the stream. The events of this slice we have to unwrap and deserialize into domain specific events. On line 13 we use the extension method DeserializeEvent() to do so.

The helper method GetStreamName used on line 6 looks like this

Now we also need to implement the Save method. This is the code we’ll be using

On line 4 we ask the aggregate if there are any uncommitted events it has for use to save. If there are none then we bail out immediately. Otherwise we determine the stream name to which we want to append the given events. The stream name is constructed from the type name of the aggregate to save and the id of the aggregate instance. We use optimistic concurrency and thus have to calculate what the version of the aggregate was when we originally loaded it. This is done on lines 7 and 8. GES will throw an exception if the stored aggregate has a different version than the one we expect.

An event as it is defined in GES consists of the data (or payload) and some metadata also called header data. On lines 10 through 14 we define what we want to store as part of the metadata. In this case it is a commit ID and the fully qualified name of the aggregate we are dealing with.

On line 15 we pack our domain events (generated by the aggregate) and the header data into GES type events. For this we use the extension method ToEventData.

On line 16 we synchronously append the events  to the stream. Finally, on line 17 we tell the aggregate to clear its collection of uncommitted events since everything is now persisted.

The two extension methods used to wrap and unwrap our domain events are defined as follows

We use the Newtonsoft.Json library (added previously to our solution) to serialize and de-serialize the events.

This is all we need to have a fully functional repository for GetEventStore.

In the Get method of the repository we used the aggregate factory to create a new instance of the requested type of aggregate. This results in the following interface

Let’s implement this interface. We will have this code

Here we use the generic parameter T to determine the type of the requested aggregate. We use this information to decide what we have to construct. We then pass the collection of events to the constructor of the aggregate. The aggregate will then re-hydrate its own state from the given collection of events.

If during the development of our application we define any other type of aggregate we can just extend the above class.

The aggregate base class

Given the code we used to implement in the GES repository we can deduce how the final version of the interface IAggregate has to look like. Lines 4, 7, 8 and 17 of the Save method define all the members we need

Every aggregate needs to implement this interface. To not have repetitive code all over it makes sense to implement an aggregate base class. Here is the code

The RedirectToWhen class used on line 37 whose implementation is shown below I originally stole from Rinat Abdullin and his of Lokad CQRS library. It is used to dispatch a given event to a matching When method that has the correct event as its (single) parameter. We will see how it is working when we show the implementation of the RecipeAggregate.

Please note the Apply method. This method will be used by the concrete aggregates to apply an event. This method adds the event to a uncommitted events collection, increases the version by one and then dispatches the event to an (optional) When method (implemented in the concrete aggregate).

The recipe aggregate

Now that we have implemented all the infrastructure we need let’s turn to the application specific (business) logic and implement the recipe aggregate. From the previous post we already know that the aggregate needs a create method. The aggregate will inherit from the base aggregate and thus we need to do some adjustments there too. All in all we will start with the following code

Now, what is the aggregate supposed to do when the Create method is called? When using event sourcing as the/an architectural pattern then an aggregate is supposed to emit an event as a result of a command in the case the command can be handled and does not violate any invariants or business rules. The event which makes sense in this context is called RecipeCreated and contains all the necessary information to inform consumers about what happened (and not more). Consequently we add the following code to the Create method

On line 3 we create the event using the information we have and use the Apply method defined in the base class to further process the event. Amongst other things the Apply method will store the event in the un-committed events collection where it can then be retrieved by the repository during the save operation.

Now in the Apply method of the aggregate base class we use the RedirectToWhen class to dispatch events to the appropriate When method. In our case we implement the correct When method in the recipe aggregate as follows

The only thing we do is storing the id in the state of the event

You might ask yourself: do we not need to also store all the other property values of the event? The answer is no, not necessarily. We only need to store state that further down the row during the life cycle of the aggregate we need to e.g. check the invariants, etc.

One thing that remains is to update the recipe application service where we call the Create method of the aggregate and also pass the id as a first parameter

That’s it folks. What we have to do now is to wire up everything and test it.

Testing what we have so far

First we need to wire up all components and initialize a connection to our GES. Open the Startup class and make sure your code look like this

Note how we define and open a connection on lines 7 through 9. On line 11 we construct our GES repository which then on line 18 we register as an instance with our IoC container.

We can now compile our solution (kpm build) and run the development server (k kestrel). We also need to start the GES and can do so in a new terminal. Navigate to the folder where GES is installed (in my case in ~/dev/tool/GES-v3.0.1) and run the following command

This will start GES on the IP address 127.0.0.1 (loopback) listening to incoming IP requests on port 1113 and on incoming HTTP requests on port 2113. You can make sure that GES is up an running by opening a browser window and navigating to localhost:1113/web/index.html. Use the login credentials admin/changeit to authenticate yourself. If that works we can now test the application using our Postman REST client on Chrome. On Linux kestrel listens on port 5004. Thus we have

In this case I try to create a new recipe with name=”Spaghetti Napoli”, etc. The response from the server is 1 which is the ID of the newly created recipe. If you add another recipe the answer should be 2 and so on.

No back in the GES client navigate to the Stream Browser tab and you should find a stream called RecipeAggregate-1 in the list. Select this stream and the GES client will show the list of events found in this stream. You can click on the JSON hyperlink of any event to see its details. You should get something like this

Great, that’s it. All works fine…

Summary</p> </p>

In this post I have shown how to implement a repository for GetEventStore (GES) and wire it up. We have then implemented an aggregate base class that implements all the logic that is common for all types of aggregates. Last we have implemented the recipe aggregate as a child of the aggregate base class. All this was done on a Ubuntu Linux distro using the new open source ASP.NET vNext. The very same application should also run without modification on Windows and OS-X.

In the next post we will implement the logic necessary to create the read model from the events. Stay tuned!