I've been looking at ES for a while now. We tend to do DDD in PHP and actually have our aggregates release events to act as our audit log and for event-driven messaging between bounded contexts. However, your article raised a couple of questions for me.
First, re. projections you said:
"you build it in the background, storing the intermediate results in a database"
How do you do this in your PHP applications? The only two options I can think of are:
to have some watcher PHP process constantly polling your event stream and checking for new events (e.g. it internally stores "I last processed event 7481" and passes any new events on to one or more registered projectors)
to have projections happen synchronously when the event is generated i.e. a request comes into the application, PHP takes the request and calls methods on your aggregate which creates events, those events are persisted (in a transaction, in case any of the projections fail) and then that event is passed on to one or more registered projectors.
Do you do either of these or do you do something else (maybe involving a message bus of some kind)?
Second, you said:
"You load the events and replay them in memory, "projecting" them, to build up your dataset"
What do you mean by "data set"? Do you mean an instance of a PHP class (i.e. take a set of events for a given aggregate, pass them all to the apply(Event $e) (or similar) method on that aggregate's class and then work with that object) or do you mean something else?
Lead Developer and Solutions Architect, I specialise in Event Sourcing, DDD and Event Driven systems. PHP and GoLang developer. Enjoys being a smart ass and having a nice whiskey.
Location
Ireland
Education
MSc in Computer Science, Trinity College, Dublin
Work
Lead Developer and Solutions Architect at Contractor
On the first one, we have a background PHP process that listens for events and then passes them to the appropriate projection whenever they are received. When an aggregate is stored, it's events are pushed to a queue (Beanstalkd for local env, SQS for staging/production). The queue's PHP client waits for new messages, so it doesn't have to constantly poll. It will timeout eventually, but then you just reconnect and try again.
We use Supervisord to keep the process alive and ensure there's only one instance running.
For queues, we're planning to switch to Kafka in the near future, as it allows each projection to listen to the event queue, and keep track of it's own position, allowing them to update independently.
You could easily make these projections immediately consistent, and I'd actually recommend that for the start, while it's a single monolith, easier to manage.
On the second, that's exactly what I mean, you get the aggregate to replay it's events, building up it's internal data set, you then use this dataset to ensure you're aggregate is making valid state transitions. Eg. Can't login a user is they were never registered in the first place.
Thanks for your detailed reply, Barry. A few follow-up questions (my apologies for the lack of basic understanding they betray!) ↓
“When an aggregate is stored, it's events are pushed to a queue”
Do you also store those events in a local database? If so, how do you ensure that events are persisted and received by the queue? I could imagine it's problematic if events end up in your database but not in the queue, and it could show the user an error in the UI after the events were persisted locally but before they got onto the queue (leading the user to believe their operation failed despite that being only half true)?
“we have a background PHP process that listens for events and then passes them to the appropriate projection whenever they are received”
Could you speak a bit more about this? Is the process taking events from your local database and pushing them onto a queue, or do you mean it's receiving from your queue and routing them to projections?
If it's the former, is this something similar to Laravel's queue worker?
If the latter, why does that need to be a process that 'listens' — couldn't you just have your queues invoke the application for each event?
Lead Developer and Solutions Architect, I specialise in Event Sourcing, DDD and Event Driven systems. PHP and GoLang developer. Enjoys being a smart ass and having a nice whiskey.
Location
Ireland
Education
MSc in Computer Science, Trinity College, Dublin
Work
Lead Developer and Solutions Architect at Contractor
"When an aggregate is stored, it's events are pushed to a queue”
"Do you also store those events in a local database?"
Yes, this is a two phase operation. First the events are written to the database (or whatever storage you use for events), then they're pushed to the queue.
Now, you raise a valid point, what happens if the event is pushed to datastore, but not the queue, or vice versa (equally possible)? This is a long standing problem in any event driven system, and there are a couple of solutions.
In our implementation, writing to the datastore is transactional, once that completes the messages are sent to the queue. The messages are used to tell other systems that something has happened, they're just there to broadcast that a change has occurred, other system will read this message, then query the datastore to see what's actually happened. In other words, the "projectors" don't trust the events on the queue, they just use them as triggers for them to read events from the source of truth, the DB.
This still has the problem of "What if messages don't appear on the queue", but it becomes a problem that sorts itself out once another event appears on the queue, it'll trigger the projectors and they'll update normally.
BTW, we've never had this kind of failure. It's possible, but very unlikely. And even if it did happen, the system will handle it.
“we have a background PHP process that listens for events and then passes them to the appropriate projection whenever they are received”
"Could you speak a bit more about this?"
Yeah, you pretty got this in your exploration of an answer. It's receiving these events from a queue. In our current implementation, we have a single queue per service. Each service has a process that pulls events of the queue, in the same way as Laravel's queue workers. When an event is received, it queries the event log for the latest events. It then takes the new events, and play them into each projector.
If you'd like to discuss this further, drop me a DM on Twitter, we could arrange a skype call or something. Always glad to discuss.
Great article, Barry. Thank you.
I've been looking at ES for a while now. We tend to do DDD in PHP and actually have our aggregates release events to act as our audit log and for event-driven messaging between bounded contexts. However, your article raised a couple of questions for me.
First, re. projections you said:
How do you do this in your PHP applications? The only two options I can think of are:
Do you do either of these or do you do something else (maybe involving a message bus of some kind)?
Second, you said:
What do you mean by "data set"? Do you mean an instance of a PHP class (i.e. take a set of events for a given aggregate, pass them all to the
apply(Event $e)(or similar) method on that aggregate's class and then work with that object) or do you mean something else?Hi Harrison,
Good questions, glad to answer.
On the first one, we have a background PHP process that listens for events and then passes them to the appropriate projection whenever they are received. When an aggregate is stored, it's events are pushed to a queue (Beanstalkd for local env, SQS for staging/production). The queue's PHP client waits for new messages, so it doesn't have to constantly poll. It will timeout eventually, but then you just reconnect and try again.
We use Supervisord to keep the process alive and ensure there's only one instance running.
For queues, we're planning to switch to Kafka in the near future, as it allows each projection to listen to the event queue, and keep track of it's own position, allowing them to update independently.
You could easily make these projections immediately consistent, and I'd actually recommend that for the start, while it's a single monolith, easier to manage.
On the second, that's exactly what I mean, you get the aggregate to replay it's events, building up it's internal data set, you then use this dataset to ensure you're aggregate is making valid state transitions. Eg. Can't login a user is they were never registered in the first place.
Hope the above is useful.
Thanks for your detailed reply, Barry. A few follow-up questions (my apologies for the lack of basic understanding they betray!) ↓
Do you also store those events in a local database? If so, how do you ensure that events are persisted and received by the queue? I could imagine it's problematic if events end up in your database but not in the queue, and it could show the user an error in the UI after the events were persisted locally but before they got onto the queue (leading the user to believe their operation failed despite that being only half true)?
Could you speak a bit more about this? Is the process taking events from your local database and pushing them onto a queue, or do you mean it's receiving from your queue and routing them to projections?
If it's the former, is this something similar to Laravel's queue worker?
If the latter, why does that need to be a process that 'listens' — couldn't you just have your queues invoke the application for each event?
Hey Harrison,
Glad to answer, let's give this shot.
Yes, this is a two phase operation. First the events are written to the database (or whatever storage you use for events), then they're pushed to the queue.
Now, you raise a valid point, what happens if the event is pushed to datastore, but not the queue, or vice versa (equally possible)? This is a long standing problem in any event driven system, and there are a couple of solutions.
In our implementation, writing to the datastore is transactional, once that completes the messages are sent to the queue. The messages are used to tell other systems that something has happened, they're just there to broadcast that a change has occurred, other system will read this message, then query the datastore to see what's actually happened. In other words, the "projectors" don't trust the events on the queue, they just use them as triggers for them to read events from the source of truth, the DB.
This still has the problem of "What if messages don't appear on the queue", but it becomes a problem that sorts itself out once another event appears on the queue, it'll trigger the projectors and they'll update normally.
BTW, we've never had this kind of failure. It's possible, but very unlikely. And even if it did happen, the system will handle it.
Yeah, you pretty got this in your exploration of an answer. It's receiving these events from a queue. In our current implementation, we have a single queue per service. Each service has a process that pulls events of the queue, in the same way as Laravel's queue workers. When an event is received, it queries the event log for the latest events. It then takes the new events, and play them into each projector.
If you'd like to discuss this further, drop me a DM on Twitter, we could arrange a skype call or something. Always glad to discuss.
Thanks. I've sent you a DM.