Dispatch and projections

Subscribers and dispatcher

An EventSubscriberInterface declares the event classes, base classes, or interfaces it handles:

use DomainFlow\EventSourcing\Interface\EventSubscriberInterface;

final class SearchIndexer implements EventSubscriberInterface
{
    public static function getSubscribedTo(): array
    {
        return [ProductCreated::class];
    }

    public function handle(DomainEventInterface $event): void
    {
        // Update an external read model or perform another reaction.
    }
}

Register subscribers with EventDispatcher and pass the dispatcher to the facade. persist() then dispatches the events that were actually stored:

$dispatcher = new EventDispatcher();
$dispatcher->register(new SearchIndexer());

$facade = new EventSourcingFacade($storage, dispatcher: $dispatcher);

The dispatcher invokes every matching subscriber. It collects subscriber failures and raises SubscriberDispatchException after all matching subscribers have had their turn. A subscriber registered through multiple matching types still receives one call per event.

Projectors

ProjectorInterface extends the subscriber contract with reset(), replay(), supports(), and getName(). ProjectorRegistry collects projectors and registers them with a dispatcher.

Live dispatch and rebuilding are separate concerns: live subscribers receive events as they are persisted, while a rebuild resets a read model and replays the global event stream.

Rebuilding a projection

RebuildProjection uses CatchUpReader, starts at the beginning of the global stream, calls reset(), and hands supported events to replay():

use DomainFlow\EventSourcing\Operation\RebuildProjection;

$result = (new RebuildProjection($storage))($projector);
$result->getEventsReplayed();
$result->getPosition();

For a long-lived or concurrent global reader, use CatchUpReader directly. It keeps a safe resumable position, tolerates a late commit behind the visible frontier, and avoids handing the same aggregate/version identity to the handler twice while it catches up. Persist getSafePosition() after a successful processing cycle.

Inline dispatch versus outbox

Choose one delivery path for a given write. A facade with an inline dispatcher must not also use a storage configured to deliver through an outbox; the facade raises DoubleDeliveryException for that combination when the storage exposes OutboxBackedStorageInterface.