Download the PHP package lingoda/domain-events without Composer

Domain Events Bundle

Installation

Bundle configuration

Records currently claimed by the messenger transport are skipped by this publisher and by ReplaceableDomainEvent replacement, so the two mechanisms cannot publish the same event twice. One consequence: if a worker dies holding a claim, a replacement event will not supersede that record until its lease expires, and both may end up relayed.

Usage

IMPORTANT NOTE: Never record domain events in doctrine lifecycle hooks!

Example of simple User entity that triggers a Domain Event.

Sample Domain Event

Sample User entity that records the event

In action

Dispatching domain events with Messenger Worker

First configure the outbox messenger transport

Skip Locked

When running multiple consumers concurrently, you can enable SKIP LOCKED to avoid row contention. Instead of consumers blocking each other on locked rows, each consumer will skip already-locked rows and pick the next available one.

This requires MySQL 8.0+ or PostgreSQL 9.5+.

Enable it via the DSN:

Or via transport options:

After that we can consume the Outbox table and dispatch domain events from it with the below command

Batching

By default the transport claims one record per database transaction, so throughput is bound by the commit fsync and round-trip of every single event. Raising batch_size claims that many records in one transaction instead, cutting the number of commits by roughly batch_size.

All options can equally be given as transport options:

Delivery guarantees

A record is only recorded as published once the worker confirms the domain event was handled - which, for an asynchronously routed event, means it reached the broker. Claiming is separate and reversible: claimedAt hides a record from other workers without asserting anything about publication.

That gives three distinct outcomes:

So delivery is at least once: your handlers must be idempotent. There is no way around this - the database and the broker cannot share a transaction, so a crash between publishing an event and recording that fact must either lose the event or replay it, and this transport chooses to replay.

A replay is byte-identical to the original, and the envelope carries an OutboxRecordIdStamp whose id is stable across replays, so consumers can deduplicate on it.

Note that a permanently failing event is retried every lease window indefinitely and keeps showing up in getMessageCount(). That is deliberate - it is visible rather than silently discarded - but it does mean a poison event shows as a backlog that never drains.

prune: true only changes what confirming a record does - delete it rather than stamp publishedOn. It is safe at any batch_size, and the trade is that a confirmed event leaves no row behind to inspect.

Upgrading from 2.x

1. Run doctrine:migrations:diff. OutboxRecord gains a nullable claimedAt column and an idx_unpublished_claimed_occurred index. publishedOn keeps its meaning, so OutboxStore::publish(), purgePublishedEvents() and LockableEventPublisher are unaffected.
2. Make your handlers idempotent. Delivery is at least once now (see above). Deduplicate on OutboxRecordIdStamp, whose id is stable across replays.
3. A failing event no longer disappears. reject() used to delete the record; it now keeps its claim and retries every lease window, so a poison event shows up as a backlog that never drains rather than as silent data loss.

4. Retune for throughput. Confirming publication is a second write, so at the default batch_size: 1 a message costs two commits where 2.x cost one. Batching pays that back, and ack_flush caps replays independently of batch size:

   config	commits per message	max replayed on a hard kill
   2.x	1.0	none - but up to 1 event lost per crash
   batch_size=1 (default)	2.0	1
   batch_size=10&ack_flush=1	1.1	1
   batch_size=100&ack_flush=1	1.01	1
   batch_size=100&ack_flush=10	0.11	10

   ?skip_locked=true&batch_size=100&ack_flush=10 is a good starting point; use ack_flush=1 if you would rather hold replays to a single record. The default is deliberately the slowest row - raising batch_size widens FOR UPDATE to that many rows, which needs skip_locked (MySQL 8+ / PostgreSQL 9.5+) to avoid contention between workers.

5. Check for direct repository calls. fetchNextRecordForUpdate() and deleteRecord() are gone; getRecordCount() and createAvailableMessagesQueryBuilder() now take the lease cut-off.

Scheduling events

We can schedule Domain Events to be published in the future

Replacing/Re-scheduling events in the event_store

We can replace/re-schedule unpublished events by implementing the ReplaceableEventInterface for the Domain Event If you implement this interface, before the OutboxRecord persister stores a new domain event, it will check if there is any previously stored but unpublished events from the same entity id, if yes it will delete them and add the new one only.

Enriching Domain Events

While domain events should be immutable, sometimes it's inevitable that you need to enrich with additional information but you don't want to assign at creation time because the service is not accessible inside the entity.

You can listen to the PreAppendEvent in a subscriber/listener that is dispatched right before the Domain Event gets persisted. At this point you can enrich with additional information.

Simple example would be injecting and actorId which corresponds to the user id that is currently interacting with the app.

Testing

Install dev dependencies

Run tests

TODO

• Add functional tests
• Add instructions for doctrine mapping and routing DomainEvent
• Fix issues around Carbon serialization