Event Sourcing

Search…
Event Sourcing
Event Sourcing PHP
Ecotone comes with Prooph Event Store integration. This is well known and stable solution providing event storage over databases like Postgres, MySQL or MariaDB. 

Ecotone provides Event Sourced Aggregates, which are stored as series of events. 
Thanks to that we will be able to have the whole history of what happened with specific Aggregate.
And build up projections, that targets specific view, in order to keep our code simple, even with really complicated queries. This divide Commands from Queries (CQRS). 
Do you want to find out, how to apply Event Sourcing in your business? Try out InspectIO!
Installation
1
composer require ecotone/pdo-event-sourcing
Copied!
And go to DBAL Support in order to configure the connection.
Ecotone Event Sourcing is based on well known and stable Prooph's Event Store.
It does provide support for three databases:
- PostgreSQL
- MySQL
- MariaDB
Event Sourced Aggregate
There are two ways, we can make use of Event Sourced Aggregates. 

Pure Event Sourced Aggregate
This way of handling events does allow for pure functions. Writes are clearly separated from writes. 
1
#[EventSourcingAggregate] // 1
2
class Ticket
3
{
4
    use WithAggregateVersioning; // 2
5
​
6
    #[AggregateIdentifier] // 1
7
    private string $ticketId;
8
    private string $ticketType;
9
​
10
    #[CommandHandler] // 2
11
    public static function register(RegisterTicket $command) : array
12
    {
13
        return [new TicketWasRegistered($command->getTicketId(), $command->getTicketType())];
14
    }
15
​
16
    #[CommandHandler] // 2
17
    public function close(CloseTicket $command) : array
18
    {
19
        return [new TicketWasClosed($this->ticketId)];
20
    }
21
​
22
    #[EventSourcingHandler] // 4
23
    public function applyTicketWasRegistered(TicketWasRegistered $event) : void
24
    {
25
        $this->ticketId       = $event->getTicketId();
26
        $this->ticketType     = $event->getTicketType();
27
    }
28
}
Copied!
1.
EventSourcingAggregate and AggregateIdentifier works exactly the same as State-Stored Aggregate.
2.
Event Sourced Aggregate must provide version. You may leave it to Ecotone using WithAggregateVersioning or you can implement it yourself.
3.
CommandHandlerfor event sourcing returns events generated by specific method. This will be passed to the Repository to be stored. 
4.
EventSourcingHandler is method responsible for reconstructing Aggregate from previously created events. At least one event need to be handled in order to provide AggregateIdentifier.
Internal Recorder Aggregate
This way of handling events allow for similarity with State Stored Aggregates.
1
#[EventSourcingAggregate(true)] // 1
2
class Basket
3
{
4
    use WithAggregateEvents;
5
    use WithAggregateVersioning;
6
​
7
    #[AggregateIdentifier]
8
    private string $id;
9
​
10
    #[CommandHandler] // 2
11
    public static function create(CreateBasket $command) : static
12
    {
13
        $basket = new static();
14
        $basket->recordThat(new BasketWasCreated($command->getId()));
15
​
16
        return $basket;
17
    }
18
​
19
    #[CommandHandler] // 2
20
    public function addProduct(AddProduct $command) : void
21
    {
22
        $this->recordThat(new ProductWasAddedToBasket($this->id, $command->getProductName()));
23
    }
24
​
25
    #[EventSourcingHandler]
26
    public function applyBasketWasCreated(BasketWasCreated $basketWasCreated)
27
    {
28
        $this->id = $basketWasCreated->getId();
29
    }
30
}
Copied!
1.
In order to make use of alternative way of handling events, we need to set attribute to trueEventSourcingAggregate(true)
2.
Command Handlers instead of returning events are acting the same as State Stored Aggregates.
All events which will be published using recordThatwill be passed to the Repository to be stored. 
Projections
Projections are about deriving current state from the stream of events.
Projections can be added in any moment of the application lifetime and be built up from existing history, till the current time. 
This is powerful concept as it allow for building views quickly and discarding them without pain, when they are not longer needed. 
1
#[Projection("inProgressTicketList", Ticket::class] // 1
2
class InProgressTicketList
3
{
4
    private Connection $connection;
5
​
6
    public function __construct(Connection $connection)
7
    {
8
        $this->connection = $connection;
9
    }
10
​
11
    #[EventHandler] // 2
12
    public function addTicket(TicketWasRegistered $event, array $metadata) : void
13
    {
14
        $result = $this->connection->executeStatement(<<<SQL
15
    INSERT INTO in_progress_tickets VALUES (?,?)
16
SQL, [$event->getTicketId(), $event->getTicketType()]);
17
    }
18
​
19
    #[QueryHandler("getInProgressTickets")] // 3
20
    public function getTickets() : array
21
    {
22
        return $this->connection->executeQuery(<<<SQL
23
    SELECT * FROM in_progress_tickets
24
SQL)->fetchAllAssociative();
25
    }    
26
}
Copied!
1.
This tells Ecotone that specific class is Projection. The first parameter is the name of the projection and the second is name of the stream (the default is the name of the Aggregate) that this projection subscribes to. 
2.
Events that this projection subscribes to
3.
Optional Query Handlers for this projection. They can be placed in different classes depending on preference. 
Running Projection
Synchronously Event Driven Projection
By default Ecotone runs the projections synchronously. There is no additional configuration needed for this.
This kind of running configuration can be used to avoid eventual consistency or for testing purposes. 
However when you expect concurrent access to your Aggregates, you may consider using different approach to limit the time of your database transaction to minimum.
Polling Projection
You may run your projection in the background. 
It will query the database within constant time intervals, to look if new events have been registered. 
Each projection is running as separate process. 
To register Polling Projection make use of ServiceContext.
1
#[ServiceContext]
2
public function basketList()
3
{
4
    return ProjectionRunningConfiguration::createPolling("basket_list");
5
}
Copied!
After setting up Pollable Channel we can run the endpoint:
Symfony
Laravel
Lite
1
bin/console ecotone:run basket_list -vvv
Copied!
1
artisan ecotone:run basket_list -vvv
Copied!
1
$messagingSystem->run("basket_list");
Copied!
Asynchronously Event Driven Projection
You may pass your projections in event driven manner using asynchronous channels.
1
#[Asynchronous("asynchronous_projections")]
2
#[Projection("basket_list")]
3
class BasketList
Copied!

The difference between Polling and Event Driven projection is the way they are triggered. 
The Event Driven is only triggered when new event comes to the system. This avoid the pitfall of continues database access while using Polling Projection.
The second strength of Asynchronously Event Driven Projection is possibility of registering multiple projections under same channel (which is same consumer).
Consider using Dbal Message Channels to avoid 2PC transactions problems. 
Thanks to that Ecotone will store events and publish them within same transaction. The solution is based on Outbox Pattern. 
With Event Driven projections, your projections will not be created on startup, as they will be called when event happens.
If this is problem, you may consider starting with Polling Projection and then switching to Event Driven Projection.
Projection Actions
Projection initialization
As projection can be restarted, deleted and created differently. It's easier to maintain, when the projection knows how to setup it itself, instead of depending on migrations. 
Method with attribute #[ProjectionInitialization] will be called on startup of the projection.
1
#[ProjectionInitialization]
2
public function initialization() : void
3
{
4
$this->connection->executeStatement(<<<SQL
5
    CREATE TABLE IF NOT EXISTS in_progress_tickets (
6
        ticket_id VARCHAR(36) PRIMARY KEY,
7
        ticket_type VARCHAR(25)
8
    )
9
SQL);
10
}
Copied!
Resetting the projection
In order to restart the projection in case we want to provide incompatible change, we can simply reset the projection and it will build up from the beginning. 
Symfony
Laravel
Lite
1
bin/console ecotone:es:reset-projection {projectionName}
Copied!
1
artisan ecotone:es:reset-projection {projectionName}
Copied!
1
$messagingSystem->runConsoleCommand("ecotone:es:reset-projection", ["name" => $projectionName]);
Copied!
And inside the projection we need to implement ProjectionReset to tell Ecotone what to do:
1
#[ProjectionReset]
2
public function reset() : void
3
{
4
$this->connection->executeStatement(<<<SQL
5
    DELETE FROM in_progress_tickets
6
SQL);
7
}
Copied!
Deleting the projection
If we want to delete the projection 
Symfony
Laravel
Lite
1
bin/console ecotone:es:delete-projection {projectionName}
Copied!
1
artisan ecotone:es:delete-projection {projectionName}
Copied!
1
$messagingSystem->runConsoleCommand("ecotone:es:delete-projection", ["name" => $projectionName]);
Copied!
And inside the projection we need to implement ProjectionDelete to tell Ecotone what to do:
1
#[ProjectionDelete]
2
public function delete() : void
3
{
4
$this->connection->executeStatement(<<<SQL
5
    DROP TABLE in_progress_tickets
6
SQL);
7
}
Copied!
Choosing Event Streams
The Projection is deriving from Event Stream.
There may be situations when we will want to derive the projection from multiple streams however. 
Let's see what options do we have:
From Single Stream
If we are interested in single stream, we can listen directly for specific aggregate
1
#[Projection("basketList", Basket::class)]
2
class BasketList
3
{
4
    #[EventHandler]
5
    public function addBasket(BasketWasCreated $event) : void
6
    {
7
        // do something
8
    }
9
}
Copied!
In here we are handling events from single Basket's Aggregate stream. It will contain all the events in relation to this aggregate.
From Multiple Streams
There may be situations, when we will want to handle different streams together.
1
#[Projection("log_projection", [Ticket::class, Basket::class])]
2
class Logger
Copied!
From Category
In case if using Stream Per Aggregate Persistence Strategy we will need to use categories to target.
If we would listen on Domain\Ticket stream using Stream Per Aggregate then we would not target any event, as the streams that are created are suffixed by the identifier Domain\Ticket-123. 
In that case we can make use of categories in order to target Domain\Ticket-*.
1
#[Projection("category_projection", fromCategories: Ticket::class)]
2
class FromCategoryUsingAggregatePerStreamProjection
Copied!
Storing And Handling Events By Names
If you want to avoid storing class names of your events in the Event Store you may mark them with name.
1
#[NamedEvent("basket.was_created")]
2
class BasketWasCreated
3
{
4
    public const EVENT_NAME = "basket.was_created";
5
​
6
    private string $id;
7
​
8
    public function __construct(string $id)
9
    {
10
        $this->id = $id;
11
    }
12
​
13
    public function getId(): string
14
    {
15
        return $this->id;
16
    }
17
}
Copied!
And tell the projection to make use of it
1
#[Projection("basket_list")]
2
class BasketList
3
{
4
    #[EventHandler("basket.was_created")]
5
    public function addBasket(BasketWasCreated $event) : void
6
    {
7
        // do something with $event
8
    }
Copied!
Speeding Up Projection Restore
If projections are handling the events by names, then there is no need to deserialization of the event to the class and simple array can be used. In case of thousands of events during resetting the projection it will speed up the process.
1
    #[EventHandler("basket.was_created")]
2
    public function addBasket(array $event) : void
3
    {
4
        // do something with $event
5
    }
Copied!
Event Sourcing Configuration
To change Event Sourcing configuration we will use Service Context.
Persistence Strategy
Single Stream Strategy
The default persistence strategy is Single Stream Strategy.
This persistence stores all instances of specific aggregate, within same stream.
1
#[ServiceContext]
2
public function persistenceStrategy()
3
{
4
    return \Ecotone\EventSourcing\EventSourcingConfiguration::createWithDefaults()
5
        ->withSingleStreamPersistenceStrategy();
6
}
Copied!
1
namespace Domain\Ticket;
2
​
3
#[EventSourcingAggregate]
4
class Ticket
Copied!
All instances of Ticket will be stored within Domain\Ticket\Ticket stream. 
Read more about this strategy under SingleStreamStrategy:
http://docs.getprooph.org/event-store/implementations/pdo_event_store/variants.html#SingleStreamStrategy​
Stream Per Aggregate Strategy
This persistence creates stream per aggregate instance.
1
#[ServiceContext]
2
public function persistenceStrategy()
3
{
4
    return \Ecotone\EventSourcing\EventSourcingConfiguration::createWithDefaults()
5
        ->withStreamPerAggregatePersistenceStrategy();
6
}
Copied!
1
namespace Domain\Ticket;
2
​
3
#[EventSourcingAggregate]
4
class Ticket
Copied!
Instances of Ticket will be stored within Domain\Ticket\Ticket-{ticketId} stream where ticketId is identifier of specific aggregate. 
Read more about this strategy under AggregateStreamStrategy:
http://docs.getprooph.org/event-store/implementations/pdo_event_store/variants.html#AggregateStreamStrategy​
Custom Stream Name
If you want to make use of custom stream name (default is Aggregate class name), then you can apply Stream attribute to your aggregate.
1
#[Stream("basket_stream")]
2
class Basket
Copied!
Then tell the projection to make use of it:
1
#[Projection(self::PROJECTION_NAME, "basket_stream")]
2
class BasketList
Copied!
Modelling - Previous
Asynchronous Handling
Next - Modelling
Scheduling
Last modified 5mo ago