p(meta). 12 Feb 2009
h1. Introduction
There has been some discussions lately about Event Sourcing. For example, Greg Young "recently discussed":http://www.infoq.com/interviews/greg-young-ddd how they were using Event Sourcing and explicit state transitions together with Domain-Driven Design (DDD) to build a highly scalable and loosely coupled system.
So what is Event Sourcing? Martin Fowler wrote an "excellent article":http://martinfowler.com/eaaDev/EventSourcing.html about some years ago and there is no use repeating it here, so please read (or at least skim) that article before reading further.
What I will do in this article is to show you how you can implement Event Sourcing using asynchronous message-passing based on actors. Actors are generally an excellent paradigm to implement asynchronous event-based systems and they allow you to easily get explicit state transitions working nicely together with an immutable domain model. This gives a concurrent system that scales very well, with the side-effect/feature of "Eventual Consistency":http://www.allthingsdistributed.com/2008/12/eventually_consistent.html.
h1. Domain model
I will reuse the example Martin Fowler used in his article but rewrite it using Scala Actors. So without further ado let's start hacking. Martin's example implements a simple Ship management system.
First, let's define the simplistic domain model; Ship, Port and Country.
The Ship class is worth discussing a bit. It is an actor, which means that it is an isolated 'lightweight process' with its own state, which is only accessible and modifiable using messages (in our case, events). The Ship actor responds to four different events; set arrival and departure, query for current port and finally reset the state.
classShip(valname:String,valhome:Port)extendsActor{defact=loop(home)privatedefloop(current:Port){react{caseArrivalEvent(time,port,_)=>println(toString+" ARRIVED at port "+port+" @ "+time)loop(port)caseDepartureEvent(time,port,_)=>println(toString+" DEPARTED from port "+port+" @ "+time)loop(Port.AT_SEA)caseReset=>println(toString+" has been reset")loop(home)caseCurrentPort=>reply(current)loop(current)caseunknown=>error("Unknown event: "+unknown)}}overridedeftoString="Ship("+name+")"}classPort(valcity:String,valcountry:Country){overridedeftoString="Port("+city+")"}objectPort{valAT_SEA=newPort("AT SEA",Country.AT_SEA)}caseclassCountry(valname:String)objectCountry{valUS=newCountry("US")valCANADA=newCountry("CANADA")valAT_SEA=newCountry("AT_SEA")}
Note: In this example I have been managing the state in the actors (Ship and EventProcessor) by passing it on in the recursive 'loop', using stack-confinement. This is a slick technique but not possible if you need to persist the state in some way, either using something like Terracotta or store it in a database. Then you would have to put the state in private field(s) in the actor, something that will not affect the correctness or performance.
h1. Events
Now let's define our events, implementing the explicit state transitions DepartureEvent and ArrivalEvent. In Scala these are best defined as 'case classes' which supports pattern matching and attribute destructing. These two events encapsulate their state transition in the 'process' method. We also define one event for asking the Ship for its current port and one for resetting its state to its "home" port.
h1. Event processor
Finally, let's define the event processor. This class is an actor which responds to any event that is a subtype of StateTransitionEvent, e.g. either DepartureEvent or ArrivalEvent. It also holds a history list ('log') with all events that it has processed. Something that we will make use of later on.
h2. Test run 1
Now we have the basis for our Ship Management Event Sourcing framework. Let's create some tests to drive the thing. Since each event submission is processed asynchronously we have to interleave them with calls to 'Thread.sleep(500)' in order to see what is going on.
Pretty nice.
But now, let's start to take advantage of the event persistence. Let's implement event replay.
h1. Replay
Implementing replay is actually very simple now when we have an event log. First we define a Replay event.
caseobjectReplayextendsEvent
Then we need the EventProcessor to respond to this new event by first reversing the order of the event log (since functional lists are concatenated in reverse order) and then for each event invoke 'process'.
Done deal.
h2. Test run 2
Let's try it out by adding a new test method to our suite. Here we make use of the Reset event which resets the ship to its initial state before replaying all state transitions.
defresetAndReplayEventLog={println("\n===> resetAndReplayEventLog")shipKR!Resetprocessor!ReplayThread.sleep(500)assert(portSFO==(shipKR!?CurrentPort))this}(newActorBasedEventSourcingTest).setup.arrivalSetsShipsLocation.departurePutsShipOutToSea.smallTrip.resetAndReplayEventLog// new test method
.tearDown
h1. Replay up to a specific point in time
Finally, (my last example, I promise) let's add the possibility of replaying the event log up to a specific date to get a snapshot of the system's state at a particular point in time.
You know the drill by now, first define a new event; ReplayUpTo, holding the date.
caseclassReplayUpTo(date:Date)extendsEvent
Here the event processor first reverses the log, then it applies a filter to the list which filters out all events that has been created after the date specified and finally run 'process' on all events in the resulting filtered list.
h2. Test run 3
So we add a last test method to our suite, one that replays all events created in earlier tests up to the date '2009/2/4'.
defresetAndReplayEventLogUpToDate={println("\n===> resetAndReplayEventLogUpToDate")shipKR!Resetprocessor!ReplayUpTo(newDate(2009,2,4))Thread.sleep(500)assert(Port.AT_SEA==(shipKR!?CurrentPort))this}(newEventSourcingWithActorsTest).setup.arrivalSetsShipsLocation.departurePutsShipOutToSea.smallTrip.resetAndReplayEventLog.resetAndReplayEventLogUpToDate// new test method
.tearDown
That's all there's to it. We have only scratched the surface on what can be done with asynchronous Event Sourcing, and as in all these kind of articles, the example is almost too simplistic to fully understand the power and flexibility of the solution. But I hope that you have understood the underlying principle enough to be able to apply it to a real-world enterprise system.
