Enterprise-grade event-driven messaging for Vortos — Kafka, outbox pattern, retry, dead letter, and more.

Vortos Messaging

The Vortos Messaging module is a production-ready, broker-agnostic event-driven messaging layer built directly into the framework. It provides everything you need to publish, consume, and reliably process domain events across services — with zero manual wiring.

Design Philosophy

Your domain code calls one method: $eventBus->dispatch($event). Everything else — Kafka transport, serialization, middleware, retries, dead letters, tracing — happens automatically behind the scenes.

What It Does

The messaging module handles the full lifecycle of a domain event from the moment it is dispatched to the moment a handler processes it:

On the producer side, the EventBus receives your domain event, stamps it with a unique ID, correlation ID, and timestamp, then routes it to the correct Kafka topic via the producer registry. If outbox mode is enabled, the event is first written to a database table within your active transaction, guaranteeing no message is ever lost even if Kafka is temporarily unavailable.

On the consumer side, a long-running worker process polls Kafka for messages, deserializes the payload back into a typed domain event, resolves all registered handlers, and runs each one through the middleware pipeline. On success the Kafka offset is committed. On failure, the consumer retries with exponential backoff before writing the message to a dead letter store.

Architecture Overview

Your Code
    │
    ▼
EventBusInterface::dispatch(DomainEvent)
    │
    ├─── Has registered producer? ──► Outbox enabled? ──► Write to outbox table
    │                                                      (OutboxRelayWorker → Kafka)
    │                              └─► Direct produce ──► Kafka topic
    │
    └─── Has internal handlers? ──► Symfony Messenger bus (in-process)

Kafka Topic
    │
    ▼
Worker Process (supervisord)
    │
    ▼
ConsumerRunner
    ├── Resolve handlers from HandlerRegistry  ← validates event class is known
    ├── Deserialize payload → DomainEvent
    └── For each handler:
            │
            ▼
        MiddlewareStack
            ├── TracingMiddleware
            ├── LoggingMiddleware
            ├── HookMiddleware
            └── TransactionalMiddleware
                    │
                    ▼
                Handler::__invoke(DomainEvent)
                    │
            ┌───────┴───────┐
          Success          Failure
            │                │
        Commit offset    Retry (exponential backoff)
                              │
                         Exhausted → Dead Letter

use Vortos\Messaging\Attribute\MessagingConfig;
use Vortos\Messaging\Attribute\RegisterTransport;
use Vortos\Messaging\Attribute\RegisterProducer;
use Vortos\Messaging\Attribute\RegisterConsumer;
use Vortos\Messaging\Driver\Kafka\Definition\KafkaTransportDefinition;
use Vortos\Messaging\Driver\Kafka\Definition\KafkaProducerDefinition;
use Vortos\Messaging\Driver\Kafka\Definition\KafkaConsumerDefinition;

#[MessagingConfig]
final class UserMessagingConfig
{
    #[RegisterTransport]
    public function userTransport(): KafkaTransportDefinition
    {
        return KafkaTransportDefinition::create('user.events')
            ->dsn('kafka://kafka:9092')
            ->topic('user.events')
            ->partitions(3)
            ->replicationFactor(1);
    }

    #[RegisterProducer]
    public function userProducer(): KafkaProducerDefinition
    {
        return KafkaProducerDefinition::create('user.events')
            ->transport('user.events')
            ->publishes(UserCreatedEvent::class)
            ->outbox(false);
    }

    #[RegisterConsumer]
    public function userConsumer(): KafkaConsumerDefinition
    {
        return KafkaConsumerDefinition::create('user.events')
            ->groupId('user-service')
            ->parallelism(1)
            ->retry(RetryPolicy::exponential(attempts: 3, initialDelayMs: 500))
            ->dlq('user.events.dlq');
    }
}

Dispatch an Event

src/User/Application/RegisterUserHandler.php

use Vortos\Messaging\Contract\EventBusInterface;

final class RegisterUserHandler
{
    public function __construct(private EventBusInterface $eventBus) {}

    public function handle(RegisterUserCommand $command): void
    {
        // ... domain logic

        $this->eventBus->dispatch(new UserCreatedEvent(
            id: new UuidV7(),
            email: $command->email,
        ));
    }
}

Write a Handler

src/User/Application/EventHandler/SendWelcomeEmailHandler.php

use Vortos\Messaging\Attribute\AsEventHandler;

#[AsEventHandler(handlerId: 'user.created.welcome-email', consumer: 'user.events')]
final class SendWelcomeEmailHandler
{
    public function __invoke(UserCreatedEvent $event): void
    {
        // send welcome email
    }
}

Start the Worker

Add your consumer to docker/worker/supervisord.conf:

[program:consumer-user-events]
command=php /var/www/html/bin/console vortos:consume user.events
autostart=true
autorestart=true
stdout_logfile=/var/log/supervisor/user-events.out.log
stderr_logfile=/var/log/supervisor/user-events.err.log

CLI Commands

The framework ships with built-in console commands for operating and inspecting the messaging system:

Command	Description
`vortos:consume <name>`	Start a consumer worker for the named pipeline
`vortos:transports:list`	List all registered transports and producers
`vortos:consumers:list`	List all registered consumers and their handlers
`vortos:outbox:relay`	Start the outbox relay worker
`vortos:outbox:replay`	Reset failed outbox rows to pending so the relay retries them
`vortos:dlq:replay`	Replay consumer failed messages from `vortos_failed_messages` back to Kafka

Messaging

Vortos Messaging

What It Does

Architecture Overview

Key Features

Transport Registration

Event Publishing

Event Consuming

Middleware Pipeline

Outbox Pattern

Retry & Dead Letter

Quick Setup

Register a Transport

Dispatch an Event

Write a Handler

Start the Worker

CLI Commands

On this page