event sourcing with php

Event Sourcingwith php

#sfPot @VeryLastRoom 2016/09Speaker: @sebastienHouze

Event Sourcingwith php

This talk is not yet another talk to:- Convince you to use ES because its qualities.- Let you listen me or having some rest, stay

aware!- Just talk about theoretical things, we will put

ES in practice (and in php, not that usual).- Sell you a had hoc solution, because as

everything in computer engineering - it depends™

Event Sourcingwith php

- core principles

- es in your domain

- projections

- persistence

main concepts definitions

use es in your root aggregate lifecycle to restore it at any state

how to adapt streams persistence in your infra

cherry on the cake!

Event Sourcingcore principles

Storing all the changes to the system, rather than just its current state.∫

Event Sourcingcore principles

change state/

Event Sourcingcore principles

change statevs

something that happenedresult of some event processingresult of some command handlingsnapshot at a given time

what we store in databaseswhat you probably don’t store?usually

Event Sourcingcore principles

a change is the result of an action on some entity / root aggregate

in an event sourced system

Changes of each root aggregate are persisted in a dedicated event stream

Event Sourcingevent stream

RegisteredVehicle

ParkedVehicle …

Vehicle CX897BC

Event Sourcingevent stream(s)

RegisteredVehicle

ParkedVehicle …

Vehicle AM069GG

RegisteredVehicle

ParkedVehicle …

Vehicle CX897BC

Event Sourcinges in your domain

Disclaimer: forget setters/reflection made by your ORM on your entities.∫

Event Sourcinges in your domain

Let’s build the next unicorn!

parkedLife™

PRAGMATIC USE CASE

Event Sourcinges in your domain

Let’s start with parkedLife app, a service which offer a pretty simple way to locate where your vehicle(s) (car, truck, ...) has been parked.

Your vehicle(s) need to be registered on the service the first time with a plate number.

When you have many vehicles you own a vehicle fleet.

Event Sourcinges in your domain

Let’s start with parkedLife app, a service which offer a pretty simple way to locate where your vehicle(s) (car, truck, ...) has been parked.

Your vehicle(s) need to be registered on the service the first time with a plate number.

When you have many vehicles you own a vehicle fleet.

Event Sourcinges in your domain

1. Emit change(s)2. Persist them in a stream3. Reconstitute state from

stream

Our goal: endless thing

Event Sourcinges in your domain

class VehicleFleet{ public function registerVehicle(string $platenumber, string $description) { $vehicle = Vehicle::register($platenumber, $this->userId); $vehicle->describe($description);

$this->vehicles[] = $vehicle;

return $vehicle; }}

FROM THE BASICS

Event Sourcinges in your domain

READY?

Event Sourcinges in your domain

class VehicleFleet{ public function registerVehicle(string $platenumber, string $description) { $this->whenVehicleWasRegistered(new VehicleWasRegistered($platenumber, (string)$this->userId)); $this->whenVehicleWasDescribed(new VehicleWasDescribed($platenumber, $description));

return $this->vehicleWithPlatenumber($platenumber); }

protected function whenVehicleWasRegistered($change) { $this->vehicles[] = Vehicle::register( $change->getPlatenumber(), new UserId($change->getUserId()) ); }

protected function describeVehicle(string $platenumber, string $description) { $this->whenVehicleWasDescribed(new VehicleWasDescribed($platenumber, $description)); }

public function whenVehicleWasDescribed($change) { $vehicle = $this->vehicleWithPlatenumber($change->getPlatenumber()); $vehicle->describe($change->getDescription()); }}

LET’S INTRODUCE EVENTS

event

event handler

class VehicleFleet{ public function registerVehicle(string $platenumber, string $description) { $changes = [ new VehicleWasRegistered($platenumber, (string)$this->userId), new VehicleWasDescribed($platenumber, $description) ];

foreach ($changes as $change) { $handler = sprintf('when%s', implode('', array_slice(explode('\\', get_class($change)), -1))); $this->{$handler}($change); }

return $this->vehicleWithPlatenumber($platenumber); }}

Event Sourcinges in your domain

AND THEN (VERY BASIC) ES

very basic local event stream

very basic sourcing of stream

1. Emit change(s)2. Persist them in a stream3. Reconstitute state from

stream

Event Sourcinges in your domain

Our goal: endless thing

MISSION

COMPLETE

Event Sourcinges in your domain

Well… no we don’t really permit to reconstitute the state from some event stream from the outside of the root aggregate, let’s refine that!

final class VehicleFleet extends AggregateRoot{ public function registerVehicle(string $platenumber, string $description) { $this->record(new VehicleWasRegistered($this->getAggregateId(), $platenumber)); $this->record(new VehicleWasDescribed($this->getAggregateId(), $platenumber, $description));

return $this->vehicleWithPlatenumber($platenumber); }

public function whenVehicleWasRegistered(VehicleWasRegistered $change) { $this->vehicles[] = Vehicle::register($change->getPlatenumber(), new UserId($change->getAggregateId())); }

public function describeVehicle(string $platenumber, string $description) { $this->record(new VehicleWasDescribed($this->getAggregateId(), $platenumber, $description)); }

public function whenVehicleWasDescribed(VehicleWasDescribed $change) { $vehicle = $this->vehicleWithPlatenumber($change->getPlatenumber()); $vehicle->describe($change->getDescription()); }}

Event Sourcinges in your domain

Look ‘ma, I’m an Aggregate root!

Generic logic managed by AggregateRoot

abstract class AggregateRoot{ private $aggregateId;

private $recordedChanges = [];

protected function __construct(string $aggregateId)

public function getAggregateId(): string

public static function reconstituteFromHistory(\Iterator $history)

public function popRecordedChanges(): \Iterator

protected function record(Change $change)}

Event Sourcinges in your domain

Prefer (explicit) named constructor if you’re applying DDD

That simple, we’re ready to source events, from an event store for example

Event Sourcinges in your domain

We’re done with our domain!let’s talk about how to persist our

events.

Event Sourcingpersistence

You just have to adapt your domain, choose your infra weapon!∫

Event Sourcingpersistence

Let’s try with one of the simplest implementations: filesystem event

store.

Event Sourcingpersistence

Pretty easy from the very deep nature of events

- Append only: as events happen- One file per stream (so by aggregate

root)

Event Sourcingpersistence

EVENT STORE INTERFACE

interface EventStore{ public function commit(Stream $eventStream);

public function fetch(StreamName $streamName): Stream;}

Advanced ES introduce at least a $version arg not covered by this talk

Event Sourcingpersistence

class FilesystemEventStore implements EventStore{ public function commit(Stream $eventStream) { $filename = $this->filename($eventStream->getStreamName()); $content = ''; foreach ($eventStream->getChanges() as $change) { $content .= $this->eventSerializer->serialize($change).PHP_EOL; }

$this->fileHelper->appendSecurely($filename, $content); }

public function fetch(StreamName $streamName): Stream { $filename = $this->filename($streamName); $lines = $this->fileHelper->readIterator($this->filename($streamName)); $events = new ArrayIterator();

foreach ($lines as $serializedEvent) { $events->append($this->eventSerializer->deserialize($serializedEvent)); }

$lines = null; // immediately removes the descriptor.

return new Stream($streamName, $events); }}

EVENT STORE IMPLEMENTATIONstream name to file name association

Event SourcingpersistenceAPP.PHP

use Shouze\ParkedLife\Domain\{Domain, EventSourcing, Adapters, Ports};

// 1. We start from pure domain code$userId = new Domain\UserId('shouze');$fleet = Domain\VehicleFleet::ofUser($userId);$platenumber = 'AM 069 GG';$fleet->registerVehicle($platenumber, 'My benz');$fleet->parkVehicle($platenumber, Domain\Location::fromString('4.1, 3.12'), new \DateTimeImmutable());

Event SourcingpersistenceAPP.PHP

use Shouze\ParkedLife\Domain\{Domain, EventSourcing, Adapters, Ports};

// 1. We start from pure domain code$userId = new Domain\UserId('shouze');$fleet = Domain\VehicleFleet::ofUser($userId);$platenumber = 'AM 069 GG';$fleet->registerVehicle($platenumber, 'My benz');$fleet->parkVehicle($platenumber, Domain\Location::fromString('4.1, 3.12'), new \DateTimeImmutable());

// 2. We build our sourceable stream$streamName = new EventSourcing\StreamName(sprintf('vehicle_fleet-%s', $userId));$stream = new EventSourcing\Stream($streamName, $fleet->popRecordedChanges());

Event SourcingpersistenceAPP.PHP

use Shouze\ParkedLife\Domain\{Domain, EventSourcing, Adapters, Ports};

// 1. We start from pure domain code$userId = new Domain\UserId('shouze');$fleet = Domain\VehicleFleet::ofUser($userId);$platenumber = 'AM 069 GG';$fleet->registerVehicle($platenumber, 'My benz');$fleet->parkVehicle($platenumber, Domain\Location::fromString('4.1, 3.12'), new \DateTimeImmutable());

// 2. We build our sourceable stream$streamName = new EventSourcing\StreamName(sprintf('vehicle_fleet-%s', $userId));$stream = new EventSourcing\Stream($streamName, $fleet->popRecordedChanges());

// 3. We adapt the domain to the infra through event sourcing$serializer = new EventSourcing\EventSerializer( new Domain\EventMapping, new Symfony\Component\Serializer\Serializer( [ new Symfony\Component\Serializer\Normalizer\PropertyNormalizer( null, new Symfony\Component\Serializer\NameConverter\CamelCaseToSnakeCaseNameConverter ) ], [ new Symfony\Component\Serializer\Encoder\JsonEncoder ] ));$eventStore = new Adapters\FilesystemEventStore(__DIR__.'/var/eventstore', $serializer, new Ports\FileHelper);$eventStore->commit($stream);

Event SourcingpersistenceAPP.PHP

use Shouze\ParkedLife\Domain\{Domain, EventSourcing, Adapters, Ports};

// 1. We start from pure domain code$userId = new Domain\UserId('shouze');$fleet = Domain\VehicleFleet::ofUser($userId);$platenumber = 'AM 069 GG';$fleet->registerVehicle($platenumber, 'My benz');$fleet->parkVehicle($platenumber, Domain\Location::fromString('4.1, 3.12'), new \DateTimeImmutable());

// 2. We build our sourceable stream$streamName = new EventSourcing\StreamName(sprintf('vehicle_fleet-%s', $userId));$stream = new EventSourcing\Stream($streamName, $fleet->popRecordedChanges());

// 3. We adapt the domain to the infra through event sourcing$serializer = …$eventStore = new Adapters\FilesystemEventStore(__DIR__.'/var/eventstore', $serializer, new Ports\FileHelper);$eventStore->commit($stream);

Event Sourcingpersistence

$ docker run -w /app --rm -v $(pwd):/app -it php:zts-alpine php app.php$ docker run -w /app --rm -v $(pwd):/app -it php:zts-alpine sh -c 'find var -type f | xargs cat' {"event_name":"vehicle_was_registered.fleet.parkedlife", "data":{"user_id":"shouze","platenumber":"AM 069 GG"}}{"event_name":"vehicle_was_described.fleet.parkedlife", "data":{"user_id":"shouze","platenumber":"AM 069 GG", "description":"My benz"}}{"event_name":"vehicle_was_parked.fleet.parkedlife", "data":{"user_id":"shouze","platenumber":"AM 069 GG","latitude":4.1,"longitude":3.12,"timestamp":1474838529}}

LET’S RUN IT

YEAH, IT’S OUR FIRST TRULY PERSISTED EVENT

STREAM!

Event Sourcingprojections

How to produce state representation(s) at any time from the very first event of your stream to any point of your stream.

∫

Event Sourcingprojections

Ok, we saw that actions on your system produce state (through

events)

But when the time come to read that state, did you notice that we often have use cases where we want to express it through many

representations?

Event Sourcingprojections

Projection is about deriving state from the stream of events.

As we can produce any state representation from the very first emitted event, we can

produce every up to date state representation derivation.

Event Sourcingprojections

Projections deserve an event bus as it permit to introduce eventual consistency (async build) of projection(s) and loose

coupling of course.

For projections of an aggregate you will (soon) need a projector.

Event Sourcingprojections

So you quickly need Read Models, very simple objects far from you root aggregate.

Event Sourcingprojections

EVENT BUS IMPLEMENTATIONclass InMemoryEventBus implements EventSourcing\EventBus{ public function __construct(EventSourcing\EventSerializer $eventSerializer, Symfony\Component\Serializer\Serializer $serializer) { $this->eventSerializer = $eventSerializer; $this->serializer = $serializer; $this->mapping = [ 'vehicle_was_registered.fleet.parkedlife' => 'VehicleWasRegistred', 'vehicle_was_described.fleet.parkedlife' => 'VehicleWasDescribed', 'vehicle_was_parked.fleet.parkedlife' => 'VehicleWasParked' ]; }

public function publish(EventSourcing\Change $change) { $eventNormalized = $this->eventSerializer->normalize($change); $projector = new Domain\ReadModel\VehicleFleetProjector(new Adapters\JsonProjector(__DIR__.'/var/eventstore', $this->serializer)); if (array_key_exists($eventNormalized['event_name'], $this->mapping)) { $handler = $this->mapping[$eventNormalized['event_name']]; $projector->{'project'.$handler}($eventNormalized['data']); } }}

Event Sourcingwith php

QUESTIONS?

Event Sourcingwith php

https://github.com/shouze/parkedLife.gitMORE CODE AT: