Event Sourcing: Projections with Liquid Projections

Projections are an important yet pretty simple concept when working with event-centric or event sourcing systems. The concept is to build a state from a stream of events. In my previous post, Event Sourcing with SQL Stream Store, I made a pretty basic projection to keep the current account balance. In this post, I’ll use Liquid Projections to accomplish the same task.

I recommend checking out my post on SQL Stream Store as I’m using the same example/demo application in this post.

Liquid Projections

Liquid Projection is a library for building projections. The concept and API are pretty simple, and I’ll cover the basics in this post.

First I’ll update my Demo Application to use the Liquid Projections NuGet package.

	<Project Sdk="Microsoft.NET.Sdk">

	<PropertyGroup>
	<OutputType>Exe</OutputType>
	<TargetFramework>netcoreapp3.0</TargetFramework>
	</PropertyGroup>

	<ItemGroup>
	<PackageReference Include="SqlStreamStore" Version="1.1.3" />
	<PackageReference Include="Newtonsoft.Json" Version="12.0.2" />
	<PackageReference Include="LiquidProjections" Version="3.2.0" />
	</ItemGroup>

	</Project>

view raw csproj.xml hosted with ❤ by GitHub

Event Map

The first building block is creating an event map. The idea is to map an event to an action that we want to be invoked when the event occurs. We can use the EventMapBuild<TContext> to create all of our event mappings. The TContext can be anything but I’ll be using it to mutate the state that represents the current balance.

As a refresher from my previous post, here is the Balance

	public class Balance
	{
	public Balance(decimal amount, DateTime asOf)
	{
	Amount = amount;
	AsOf = asOf;
	}

	public decimal Amount { get; private set; }
	public DateTime AsOf { get;private set; }

	public void Add(decimal value)
	{
	AsOf = DateTime.UtcNow;
	Amount += value;
	}

	public void Subtract(decimal value)
	{
	AsOf = DateTime.UtcNow;
	Amount -= value;

	}
	}

view raw balance.cs hosted with ❤ by GitHub

Now using the EventMapBuilder<Balance> to handle the Deposited and Withdrawn events.

	public class BalanceProjection
	{
	private readonly IEventMap<Balance> _map;
	public Balance Balance { get; } = new Balance(0, DateTime.UtcNow);

	public BalanceProjection(IStreamStore streamStore, StreamId streamId)
	{
	var mapBuilder = new EventMapBuilder<Balance>();
	mapBuilder.Map<Deposited>().As((deposited, balance) =>
	{
	balance.Add(deposited.Amount);
	});
	mapBuilder.Map<Withdrawn>().As((withdrawn, balance) =>
	{
	balance.Subtract(withdrawn.Amount);
	});

	_map = mapBuilder.Build(new ProjectorMap<Balance>()
	{
	Custom = (context, projector) => projector()
	});

	streamStore.SubscribeToStream(streamId, null, StreamMessageReceived);
	}

	private async Task<object> DeserializeJsonEvent(StreamMessage streamMessage, CancellationToken cancellationToken)
	{
	var json = await streamMessage.GetJsonData(cancellationToken);

	switch (streamMessage.Type)
	{
	case "Deposited":
	return JsonConvert.DeserializeObject<Deposited>(json);
	case "Withdrawn":
	return JsonConvert.DeserializeObject<Withdrawn>(json);
	default:
	throw new InvalidOperationException("Unknown event type.");
	}
	}

	private async Task StreamMessageReceived(IStreamSubscription subscription, StreamMessage streamMessage, CancellationToken cancellationToken)
	{
	var @event = await DeserializeJsonEvent(streamMessage, cancellationToken);
	await _map.Handle(@event, Balance);
	}
	}

view raw map.cs hosted with ❤ by GitHub

You’ll notice call Build() on the EventMapBuilder<Balance> which will return us the EventMap<Balance>. At this point, you can call Handle() to pass in the events incoming from our subscription with our current Balance.

Conditions

There are also some nice additions such as providing conditions for when to execute a map. For example, if you wanted to handle only deposited events where the amount was greater than 100.

	mapBuilder
	.Map<Deposited>()
	.When(deposited => deposited.Amount > 100)
	.As(deposited =>
	{
	// do something
	});

view raw condition.cs hosted with ❤ by GitHub

Inheritance

There is also support for inheritance, which means since my Deposited and Withdrawn events inherit from the AccountEvent base class, I could also have done

	mapBuilder.Map<AccountEvent>().As((accountEvent, balance) =>
	{
	// do something
	});

view raw base.cs hosted with ❤ by GitHub

Event Store Integration

The next on my list is write this up directly with to something like Event Store or SQL Stream Store. Stay tuned.

Check out Liquid Projections on GitHub for more.

All the source code shown in this post is available on GitHub.

If you’d like a more detailed contented related to various aspects of event sourcing, let me know in the comments or on twitter.