Event Sourcing and State Derivation

One of the most interesting aspects of the CQRS architecture is the ability to use event sourcing within your business applications. Representing all state changes as domain events has various benefits, not in the least being able to look back upon historical data to derive new information.

However, although historical information is nice, every system must also know its current state. Accessing the current state in traditional applications is just a database lookup. Can we do this in an event-based systems without adding a lot of complexity?

In my Java CQRS example application I used the following pattern:

 public class Lottery {  
   private final Set<LotteryTicket> tickets = new HashSet<LotteryTicket>();  
   private double ticketPrice;  
   private double prizeAmount;  
   private void onLotteryCreatedEvent(LotteryCreatedEvent event) {  
     this.ticketPrice = event.getTicketPrice();  
     this.prizeAmount = event.getPrizeAmount();  
   }  
   private void onTicketPurchasedEvent(LotteryTicketPurchasedEvent event) {  
     tickets.add(new LotteryTicket(aggregate, event.getTicketNumber(), event.getCustomerId().getId()));  
   }  
 }  

So for each type of event a method is added that updates the current state. Some framework magic (involving reflection and other dark arts) take care of taking the event stream and invoking the right method on the Lottery instance. This approach works quite well for simple cases, but the event handling code seems hard to reuse in other contexts, especially when multiple different events update the same state (not an issue in this example). Something like the one table per view pattern could become quite unwieldy this way.

Another problem is that this approach was only used by the core domain. The reporting side implements similar logic using plain old JDBC, so the UI can easily show the current state of the system to the user.

Reactive Programming

Fortunately, some interesting new frameworks have started to come out that explicitely deal with event based applications. One of these is the Reactive Extensions (RX) framework from Microsoft. Note that I've not been the only one to see that RX could be beneficial to CQRS applications.

Since I'm used to the Java VM and I also want to learn more about Scala and RX I took some of the concepts of RX and reimplemented these in Scala over the course of a weekend, mainly to find out how reactive programming can help when dealing with event-driven systems. In Scala the example becomes:

 class Lottery extends Subject[LotteryEvent] {  
   val prizeAmount = collect { case event: LotteryCreated => event.prizeAmount }.first  
   val ticketPrize = collect { case event: LotteryCreated => event.ticketPrize }.first  
   val tickets = collect { case event: TicketPurchased => new LotteryTicket(this, event.customerId, event.ticketNumber) }  
 }  

Here Lottery is a Subject[LotteryEvent], meaning that it can observe a stream of LotteryEvents and then republish these to other subscribers. To determine the Lottery's current prizeAmount, the collect method is used to subscribe to this Lottery instance and filter on LotteryCreated events. The first such event determines the prizeAmount. The ticketPrize is similarly defined.

The Lottery's tickets contains the information from all the TicketPurchased events.

Although the amount of code doesn't differ much, the definitions of the reactive versions can easily be reused in other contexts (with a little refactoring) and doesn't require any reflection. Furthermore, the reactive framework contains many more combinators that allow you to handle multiple event streams, corrolate events, etc.

Conclusion

The reactive version may not be a clear win, but it looks interesting enough to investigate further. Especially interesting will be extending this to more complex scenarios as well as persisting these derived states to a database for quick querying.

And finally, a word of caution: trivial examples like this can be very misleading.