Event Sourcing

Example: Tracking Ships (C#)

Here is a very simple example of Event Sourcing to get the basic idea over. For this example I'm deliberately being ultra-simple to act as a starting point - then I'll use further examples to explore some of the more complicated issues.

The domain model is a simple one of ships that carry cargo and move between ports.

There are four kinds of events that affect the model:

• Arrival: ship arrives at a port
• Departure: ship leaves a port
• Load: cargo is loaded on a ship
• Unload: cargo is unloaded from a ship

Let's take a simple example of moving a ship around.

class Tester...

  Ship kr;
  Port sfo, la, yyv;
  Cargo refact;
  EventProcessor eProc;

  [SetUp]
  public void SetUp() {
    eProc = new EventProcessor(); 
    refact = new Cargo ("Refactoring");
    kr = new Ship("King Roy");
    sfo = new Port("San Francisco", Country.US);
    la = new Port("Los Angeles", Country.US);
    yyv = new Port("Vancouver", Country.CANADA) ;
  }

[Test]
public void ArrivalSetsShipsLocation() {
  ArrivalEvent ev = new ArrivalEvent(new DateTime(2005,11,1), sfo, kr);
  eProc.Process(ev);
  Assert.AreEqual(sfo, kr.Port);
}
[Test]
public void DeparturePutsShipOutToSea() {
  eProc.Process(new ArrivalEvent(new DateTime(2005,10,1), la, kr));
  eProc.Process(new ArrivalEvent(new DateTime(2005,11,1), sfo, kr));
  eProc.Process(new DepartureEvent(new DateTime(2005,11,1), sfo, kr));
  Assert.AreEqual(Port.AT_SEA, kr.Port);    
}

To make these tests work we just need the arrival and departure events. The event processor is very simple.

class EventProcessor...

  IList log = new ArrayList();  
  public void Process(DomainEvent e) {
    e.Process();
    log.Add(e);
  }

Each event has a process method.

class DomainEvent...

  DateTime _recorded, _occurred;
  internal DomainEvent (DateTime occurred) {
    this._occurred = occurred;
    this._recorded = DateTime.Now;
  }
  abstract internal void Process();

The arrival event simply captures the data and has a process method that simply forwards the event to an appropriate domain object.

class DepartureEvent...

  Port _port;
  Ship _ship;
  internal Port Port  {get { return _port; }}
  internal Ship Ship  {get { return _ship; }}
  internal DepartureEvent(DateTime time, Port port, Ship ship) : base (time)  {
    this._port = port;
    this._ship = ship;
  }
  internal override void Process() {
    Ship.HandleDeparture(this);
  }

So here the event just does the processing selection logic. The processing domain logic is done by the ship.

class Ship...

  public Port Port;
  public void HandleDeparture(DepartureEvent ev) {
    Port = Port.AT_SEA;
  }

A departure event just sets the Ship's port to a Special Case. You'll notice I'm passing the event into the domain object. There is a choice here about whether the event should pass in just the data the domain object needs for its processing, or pass in the event itself. By passing the event the event doesn't need to know exactly what data the domain logic needs. If the event gains additional data later, there's no need to update signatures. The downside of passing the event is that the domain logic is now aware of the event itself.

That simple test just shows how the basic event processing works. Now I'll show a little domain logic to see how that would work. Our ships are transporting books to satisfy my fantasies of having an entire shipping line to move my books around the world. As I'm sure you know, it's very dangerous to send books through Canada, since it's this runs the risk of the books being contaminated with lots of “eh”. I've seen books where nearly every sentence has a eh (longer sentences can get two or three).

So my perfect book shipping system would be able to detect if a cargo had passed through Canada.

class Tester...

  [Test]
  public void VisitingCanadaMarksCargo() {
    eProc.Process(new LoadEvent(new DateTime(2005,11,1), refact, kr));
    eProc.Process(new ArrivalEvent(new DateTime(2005,11,2), yyv, kr));
    eProc.Process(new DepartureEvent(new DateTime(2005,11,3), yyv, kr ));
    eProc.Process(new ArrivalEvent(new DateTime(2005,11,4), sfo, kr));
    eProc.Process(new UnloadEvent(new DateTime(2005,11,5), refact, kr));
    Assert.IsTrue(refact.HasBeenInCanada);
  }

Since the cargo may be moved between ships and offloaded, it's the cargo that has the responsibility of knowing whether it's been exposed to these northern dangers. Fortunately the risk only occurs when actually in a port, just being in the waters is safe. So our arrival event has to keep track of this.

class ArrivalEvent...

  Port _port;
  Ship _ship; 
  internal ArrivalEvent (DateTime occurred, Port port, Ship ship) : base (occurred) {
    this._port = port;
    this._ship = ship;
  } 
  internal Port Port {get {return _port;}}
  internal Ship Ship {get{return _ship;}}
  
  internal override void Process() {
    Ship.HandleArrival(this);
  }

Again the handler is the Ship object.

class Ship...

  IList cargo;
  public void HandleArrival (ArrivalEvent ev) {
    Port = ev.Port;
    foreach (Cargo c in cargo) c.HandleArrival(ev);
  }

The ship isn't responsible for tracking Canadian visits so it passes the arrival notification on to the cargo.

class Cargo...

  public bool HasBeenInCanada = false;
  public void HandleArrival(ArrivalEvent ev) {
    if (Country.CANADA == ev.Port.Country) 
      HasBeenInCanada = true;
  }

Consider the alternative of where the event process method holds the domain logic - it would need to have a lot of knowledge of the domain model as the domain logic gets more complex. In this approach the domain objects pass the event around to relevant objects so they can handle it to do what they need in response.

Example: Updating an External System (C#)

One of the great features of Event Sourcing is that you can reprocess events as much as you like. However if processing events should cause interactions with external systems, then this is a Bad Thing. The best that could happen is they'll tired of all your Spam events.

An easy way to help deal with this is to ensure that your system calls external systems through gateways that can be configured to ensure that no messages go out unless you are processing an event 'for real'.

I'll illustrate this with a simple example using ships and ports (no cargo this time). Let's say that whenever a ship enters a port it must notify the local customs authority. We can make this happen in the event processing domain logic.

class Port...

  public void HandleArrival (ArrivalEvent ev) {
    ev.Ship.Port = this;
    Registry.CustomsNotificationGateway.Notify(ev.Occurred, ev.Ship, ev.Port);
  }

Notice that this code just invokes the notification on the gateway object, it doesn't care whether this is a real processing or some kind of replay. The general principle here is that the domain logic should never care about the context of the running of the events.

It's the gateway's responsibility to figure out whether to actually send the message on or not. Since this case is pretty simple, it does this simply by having a link to the event processor and checking to see if the processor is active.

class CustomsEventGateway...

  EventProcessor processor;
  public void Notify (DateTime arrivalDate, Ship ship, Port port) {
    if (processor.isActive) 
      SendToCustoms(BuildArrivalMessage(arrivalDate, ship, port));
  }

The event processor simply makes itself active when doing regular processing.

class EventProcessor...

  public void Process(DomainEvent e) {
    isActive = true;
    e.Process();
    isActive = false;
    log.Add(e);
  }

Although this case is very simple, the fundamental principles are the same. Gateways decide whether to send an external message, not the domain logic. The gateways decide this based on information they gather about the context of the processing. In this case a simple boolean state from the processor is enough.

Example: Reversing an Event (C#)

Here we'll take the shipping example and see how to reverse the events. The critical thing we need for reversal is to ensure that we can accurately calculate the prior state of any object that has changed state due to the event.

A good place to store this prior data is on the event itself, something that works quite well with the example's approach of passing the event around the domain objects. Since the domain objects have the event to hand, they can easily store information on the event for them.

The load event makes a simple example. The event carries the following source data.

class LoadEvent...

  int _shipCode;
  string _cargoCode;
  internal LoadEvent(DateTime occurred, string cargo, int ship) : base(occurred){
    this._shipCode = ship;
    this._cargoCode = cargo;
  }
  internal Ship Ship {get { return Ship.Find(_shipCode); }}
  internal Cargo Cargo {get { return Cargo.Find(_cargoCode); }}

The processing is handed off to the cargo object, which needs to store the port that was the prior location of the cargo.

class LoadEvent...

  internal override void Process() {
    Cargo.HandleLoad(this);
  }

internal Port priorPort;

class Cargo...

  internal void HandleLoad(LoadEvent ev) {
    ev.priorPort = _port;
    _port = null;
    _ship = ev.Ship;
    _ship.HandleLoad(ev);
  }

To reverse the event we add a reverse method that mirrors the process method, calling a reversal method on the domain object.

class LoadEvent...

  internal override void Reverse() {
    Cargo.ReverseLoad(this);
  }

class Cargo...

  public void ReverseLoad(LoadEvent ev) {
    _ship.ReverseLoad(ev);
    _ship = null;
    _port = ev.priorPort;
  }

In this case the event takes on a bunch of mutable prior data which is part of its processing data. In cases like this, simple fields suffice. Other cases can require a more sophisticated data structure. When the cargo handles an arrival event, it keeps track of whether it's been in Canada eh. It can do this with a simple boolean field. To store the prior value on the event needs more than a simple field since many cargoes can be affected by an arrival. So in this case I use a map indexed by the cargo.

class Cargo...

  public void HandleArrival(ArrivalEvent ev) {
    ev.priorCargoInCanada[this] = _hasBeenInCanada;
    if ("CA" == ev.Port.Country) 
      _hasBeenInCanada = true;
  }
  private bool _hasBeenInCanada = false;
  public bool HasBeenInCanada {get { return _hasBeenInCanada;}}

class ArrivalEvent...

  internal Port priorPort;
  internal IDictionary priorCargoInCanada = new Hashtable();

Then to reverse:

class Cargo...

  public void ReverseArrival(ArrivalEvent ev) {
    _hasBeenInCanada = (bool) ev.priorCargoInCanada[this];
  }

This example nicely illustrates how the source data on the event and the error handling affects how we do reversal. For the load event we need to store the port that the cargo was at when it was loaded. If the event included this on its source data we wouldn't have to do this. A bit of extra source data removes the need to add prior data. This doesn't work everywhere: the arrival event can't source the prior Canadian state of its cargoes.

What's deemed correct processing can also make a difference. In this system we have both arrival events and departure events for ships. Assuming all works correctly we should always get these interleaved. Thus a ship could reverse an arrival event by setting its port field to Port.OUT_TO_SEA. What happens when we get two arrival events in succession? To reverse this we need to store the prior port on the ship. Our alternative would be to declare a second arrival without a departure as an error, we don't need to do this storage.

Example: External Query (C#)

External queries are awkward even for basic Event Sourcing because if you want to rebuild the application state you need to do this using external query responses that were made in the past.

Let's imagine a case where a ship has to determine the value of its cargoes when it enters a port. This valuation is done by an external service. If a ship enters a port on November 3rd and I process the event immediately, I'll get the value of that cargo as at November 3rd. If I rebuild my application state on December 5th I'll want the same value of the cargo, even if its value has changed in the meantime.

For this example I'll show one way of handling this by keeping a record of external queries and using them to provide values when reprocessing an event. In many ways it's a similar approach to that you use with an external update - turn the interaction into events at the boundary of the system and use the record of events to remember what happened.

Using a similar style to event handling that I'm using elsewhere in these examples, we pass the event to the domain objects to handle. In this case the cargo initiates the call to the external system and saves the value. (We'll assume we're going to actually do something useful with this value, but that's not relevant to this example.)

class Cargo...

  public void HandleArrival(ArrivalEvent ev) {
    declaredValue = Registry.PricingGateway.GetPrice(this);
  }
  private Money declaredValue;

As is my habit, I encapsulate all external system access through a Gateway that I control. The basic gateway will just translate the call into whatever is needed for the external interaction. To support replaying events I just wrap this with a class that logs these queries.

Figure 6: Wrapping the gateway with a logging gateway to support proper rebuilding from events.

The logging gateway checks each call to see if it has an old request that matches this one, if not it makes a new request.

class LoggedPricingGateway...

  public Money GetPrice(Cargo cargo) {
    GetPriceRequest oldReq = oldRequest(cargo);
    if (null != oldReq) return (Money) oldReq.Result; 
    else return newRequest(cargo);
  }

If the request is a new one, it turns the request into an event object, invokes the external query, and stores it in the log.

class LoggedPricingGateway...

  private Money newRequest(Cargo cargo) {
    GetPriceRequest request = new GetPriceRequest(cargo);
    request.Result = gateway.GetPrice(cargo);
    log.Store(request);
    return (Money) request.Result;
  }

private class GetPriceRequest : QueryEvent {
  private Cargo cargo;
  public GetPriceRequest(Cargo cargo) : base() {
    this.cargo = cargo;
  }

class QueryEvent...

  DomainEvent _eventBeingProcessed;
  Object result;
  public QueryEvent() {
    _eventBeingProcessed = Registry.EventProcessor.CurrentEvent;
  }
  public object Result {
    get { return result; }
    set { result = value; }
  }
  public DomainEvent EventBeingProcessed {
    get { return _eventBeingProcessed; }
  }
}

So to find an old request, it searches its log.

class LoggedPricingGateway...

  private GetPriceRequest oldRequest(Cargo cargo) {
    IList candidates =
      log.FindBy(EventProcessor.CurrentEvent, typeof (GetPriceRequest));
    foreach (GetPriceRequest request in candidates) {
      if (request.Cargo.RegistrationCode == cargo.RegistrationCode)
        return request;
    }
    return null;
  }

The query log is generic, so we can issue a query to get a few items by using the Domain Event that was processed and the type of request. This gives us a small set that needs further checking in a gateway-specific manner.

The log of requests will need to be persisted in the same way as the log of Domain Events is persisted as it's needed to rebuild the application state.