Write event handlers with a single attribute. Vortos discovers, wires, and runs them automatically through the full consumer pipeline.

Consuming Events

Vortos' consumer pipeline runs as a long-lived worker process. It polls Kafka continuously, deserializes incoming messages into typed domain event objects, resolves all registered handlers, and runs them through the middleware stack. You write handlers — the framework does everything else.

Writing a Handler

The simplest handler is a class with __invoke and a single #[AsEventHandler] attribute:

use Vortos\Messaging\Attribute\AsEventHandler;

#[AsEventHandler(
    handlerId: 'user.registered.send-welcome-email',
    consumer: 'user.events',
)]
final class SendWelcomeEmailHandler
{
    public function __construct(
        private readonly MailerInterface $mailer,
    ) {}

    public function __invoke(UserRegistered $event): void
    {
        $this->mailer->send(
            to: $event->email,
            subject: 'Welcome to the platform',
        );
    }
}

That is the entire integration. No registration, no YAML, no service tags. Vortos discovers this class at container compile time via the #[AsEventHandler] attribute and wires it into the consumer pipeline automatically.

How Discovery Works

At container compile time, HandlerDiscoveryCompilerPass runs and:

Finds all services tagged vortos.event_handler (set automatically by the attribute autoconfiguration in MessagingExtension)
Reflects the class to find the handler method and its first parameter type
Validates that the first parameter is a final class (the POPO event payload) — throws at compile time if not
Builds a descriptor array containing the service ID, method name, priority, idempotency flag, payload type, and optional EventEnvelope injection flag
Stores descriptors in the vortos.handlers container parameter, keyed by [consumerName][payloadType]
Populates vortos.handler_locator (a Symfony ServiceLocator) with all handler services for lazy resolution at runtime

Compile time:
HandlerDiscoveryCompilerPass
    └── Scans tagged services
        └── Builds handler descriptor:
            {
                handlerId:       'user.registered.send-welcome-email',
                serviceId:       'App\...\SendWelcomeEmailHandler',
                method:          '__invoke',
                consumer:        'user.events',
                payloadType:     'App\User\Domain\Event\UserRegistered',
                injectEnvelope:  true,   // true if EventEnvelope is second param
                priority:        0,
                idempotent:      false,
            }

Runtime:
ConsumerRunner::handleMessage()
    └── HandlerRegistry::getHandlers('user.events', 'App\...\UserRegistered')
        └── Returns [descriptor1, descriptor2, ...]
            └── For each descriptor: handlerLocator->get(serviceId)
                └── calls handler($payload) or handler($payload, $envelope)

Attribute Reference

`#[AsEventHandler]`

Can be placed on a class (uses __invoke) or on a method:

// Class-level — __invoke is the handler
#[AsEventHandler(handlerId: 'user.registered.notify', consumer: 'user.events')]
final class NotifyAdminHandler
{
    public function __invoke(UserRegisteredEvent $event): void {}
}

// Method-level — multiple handlers in one class
final class UserEventHandlers
{
    #[AsEventHandler(handlerId: 'user.registered.email', consumer: 'user.events')]
    public function onRegistered(UserRegisteredEvent $event): void {}

    #[AsEventHandler(handlerId: 'user.deleted.cleanup', consumer: 'user.events')]
    public function onDeleted(UserDeletedEvent $event): void {}
}

Parameters

Parameter	Type	Required	Description
`handlerId`	`string`	✓	Unique ID for this handler. Used for idempotency tracking and logging.
`consumer`	`string`	✓	Consumer pipeline name. Must match a consumer registered with `#[RegisterConsumer]`.
`priority`	`int`	—	Execution order when multiple handlers handle the same event. Higher runs first. Default: `0`.
`idempotent`	`bool`	—	If `true`, the handler opts out of deduplication — it always runs, even for duplicate `event_id` values. Set this when the handler is genuinely idempotent (e.g. an upsert that always produces the same result). Default: `false` — duplicates are skipped.
`version`	`int\|null`	—	Optional schema version filter. Only processes messages with a matching version header.

Multiple Handlers for the Same Event

Multiple handlers can be registered to the same consumer and event class. They run sequentially in priority order within a single message processing cycle. The Kafka offset is committed only after all handlers succeed:

// Runs first (priority: 100)
#[AsEventHandler(handlerId: 'order.placed.reserve-inventory', consumer: 'orders', priority: 100)]
final class ReserveInventoryHandler
{
    public function __invoke(OrderPlacedEvent $event): void {}
}

// Runs second (priority: 50)
#[AsEventHandler(handlerId: 'order.placed.send-confirmation', consumer: 'orders', priority: 50)]
final class SendOrderConfirmationHandler
{
    public function __invoke(OrderPlacedEvent $event): void {}
}

// Runs third (priority: 0 — default)
#[AsEventHandler(handlerId: 'order.placed.update-analytics', consumer: 'orders', priority: 0)]
final class UpdateAnalyticsHandler
{
    public function __invoke(OrderPlacedEvent $event): void {}
}

All or Nothing

If any handler in the chain throws an exception, the entire message fails — none of the subsequent handlers run, and the offset is not committed. The message is retried from the beginning of the handler chain. By default (idempotent: false), handlers that already succeeded before the failure are skipped on retry — the cache key set after their first success prevents them running again. If your handler must always run even on retry (e.g. a pure upsert), set idempotent: true.

EventEnvelope — Metadata in Handlers

Declare EventEnvelope as a second parameter to access aggregate identity, timing, and cross-cutting metadata without any header attributes:

use Vortos\Messaging\Contract\EventEnvelope;

#[AsEventHandler(handlerId: 'user.registered.audit', consumer: 'user.events')]
final class AuditUserRegistrationHandler
{
    public function __invoke(UserRegistered $event, EventEnvelope $envelope): void
    {
        $aggregateId      = $envelope->aggregateId;
        $occurredAt       = $envelope->occurredAt;
        $correlationId    = $envelope->metadata->correlationId;
        $tenantId         = $envelope->metadata->tenantId;
    }
}

EventEnvelope is the preferred way to access metadata — it is type-safe and gives you everything in one place.

Header Injection

For cases where you need individual values as typed parameters (useful for handlers that only need one or two fields), use injection attributes. There are two kinds: built-in attributes for standard Vortos fields, and a generic #[Header] attribute for any raw Kafka header.

Built-in header attributes

use Vortos\Messaging\Attribute\Header\MessageId;
use Vortos\Messaging\Attribute\Header\CorrelationId;
use Vortos\Messaging\Attribute\Header\Timestamp;
use Vortos\Messaging\Attribute\Header\CausationId;
use Vortos\Messaging\Attribute\Header\TraceId;
use Vortos\Messaging\Attribute\Header\TenantId;
use Vortos\Messaging\Attribute\Header\UserId;

#[AsEventHandler(handlerId: 'user.registered.audit', consumer: 'user.events')]
final class AuditUserRegistrationHandler
{
    public function __invoke(
        UserRegistered $event,
        #[MessageId]     string            $messageId,
        #[CorrelationId] string            $correlationId,
        #[Timestamp]     DateTimeImmutable $occurredAt,
        #[TenantId]      ?string           $tenantId,
        #[UserId]        ?string           $userId,
        #[CausationId]   ?string           $causationId,
        #[TraceId]       ?string           $traceId,
    ): void {}
}

Attribute	Source	Type
`#[MessageId]`	`EventEnvelope::$eventId`	`string`
`#[CorrelationId]`	`EventEnvelope::$metadata->correlationId`	`string`
`#[Timestamp]`	`EventEnvelope::$occurredAt`	`DateTimeImmutable`
`#[CausationId]`	`EventEnvelope::$metadata->causationId`	`?string`
`#[TraceId]`	`EventEnvelope::$metadata->traceId`	`?string`
`#[TenantId]`	`EventEnvelope::$metadata->tenantId`	`?string`
`#[UserId]`	`EventEnvelope::$metadata->userId`	`?string`

Use #[Header('header-name')] to inject any raw Kafka message header directly into a handler parameter. The value is extracted from the raw headers array carried by HeadersStamp on the envelope:

use Vortos\Messaging\Attribute\Header\Header;

#[AsEventHandler(handlerId: 'order.placed.process', consumer: 'orders')]
final class ProcessOrderHandler
{
    public function __invoke(
        OrderPlacedEvent $event,
        #[Header('x-tenant-id')] ?string $tenantId,
        #[Header('x-source-region')] ?string $region,
    ): void {
        // $tenantId — from the raw 'x-tenant-id' Kafka header
        // $region   — from the raw 'x-source-region' Kafka header
    }
}

If the header is not present on the message, the injected value is null. Declare the parameter as ?string to make this explicit.

#[Header] is resolved at compile time by HandlerDiscoveryCompilerPass — the parameter descriptor is stored alongside the event type, with no runtime reflection.

Registering a Consumer

Consumers are defined in your #[MessagingConfig] class alongside your transport:

use Vortos\Messaging\Retry\RetryPolicy;

#[RegisterConsumer]
public function userConsumer(): KafkaConsumerDefinition
{
    return KafkaConsumerDefinition::create('user.events')
        ->groupId('user-service')         // Kafka consumer group ID
        ->parallelism(1)                  // message-level parallelism
        ->batchSize(1)                    // messages per commit cycle
        ->retry(RetryPolicy::exponential(
            attempts: 3,
            initialDelayMs: 500,
        ))
        ->dlq('user.events.dlq')          // Kafka transport to produce failed messages to
        ->idempotencyTtl(86400)           // dedup window in seconds — overrides global default
        ->offsetReset('earliest');        // where to start if no offset committed
}

The consumer name ('user.events') must match the consumer: parameter in your #[AsEventHandler] attributes.

`idempotencyTtl`

Each consumer can override the global idempotency TTL. The global default (86400 seconds) comes from VortosMessagingConfig::consumerDefaults() — see Idempotency. Calling ->idempotencyTtl() on a consumer definition overrides it for that consumer only:

->idempotencyTtl(3600)   // 1-hour dedup window for this consumer

`dlq`

->dlq('transport.name') names a registered Kafka transport. After retry exhaustion, the consumer writes the message to vortos_failed_messages and produces it to this Kafka topic. If not set, the message is only written to vortos_failed_messages. See Dead Letter for details.

Running the Worker

Each consumer runs as a separate long-lived process managed by supervisord:

docker/worker/supervisord.conf

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

To add a new consumer, add a new [program:...] block — Docker Compose never needs to change.

The vortos:consume command accepts a --max-messages option to exit cleanly after processing a fixed number of messages. This is useful for one-shot batch runs or cron-driven processing:

php bin/console vortos:consume user.events --max-messages=500

When --max-messages is set, the process exits with code 0 after processing that many messages (or when SIGTERM is received, whichever comes first). Supervisord will restart it unless you set autorestart=false for that program.

Graceful Shutdown

The vortos:consume command installs signal handlers for SIGTERM and SIGINT. When supervisord stops the process, the consumer finishes processing the current message before exiting — the offset is always committed cleanly.

Listing Registered Consumers

php bin/console vortos:consumers:list

Example output:

Found 2 consumer(s).

▶ user.events  [kafka]
  Group:       user-service
  Transport:   user.events
  Config:      parallelism: 1   batch: 10   ttl: 86400s
  Retry:       exponential · 3 attempts · 500ms initial · 30000ms cap
  DLQ:         user.events.dlq

  Handlers (3):
    UserRegistered  (App\User\Domain\Event\UserRegistered)
      ├ user.registered.send-welcome-email    priority:   0   idempotent: no   (dedup on — duplicates skipped)
      └ user.registered.audit                priority:  10   idempotent: no   (dedup on — duplicates skipped)

    UserDeleted  (App\User\Domain\Event\UserDeleted)
      └ user.deleted.cleanup                  priority:   0   idempotent: yes  (no dedup — always runs)

▶ notifications.internal  [in-process]
  Group:       —
  Transport:   —
  Config:      parallelism: 1   batch: 1    ttl: —
  Retry:       —
  DLQ:         —

  Handlers (1):
    OrderPlaced  (App\Order\Domain\Event\OrderPlaced)
      └ order.placed.push-notification        priority:   0   idempotent: no   (dedup on — duplicates skipped)

Consumer Pipeline Internals

When a message arrives from Kafka, ConsumerRunner::handleMessage() runs the following sequence:

Extract payload type

Read the payload_type header from the message (the FQCN of the POPO event class). If missing, reject the message immediately — it cannot be routed.

Resolve handlers

Call HandlerRegistry::getHandlers($consumerName, $payloadType) to look up all handler descriptors for this consumer and payload type, sorted by priority descending.

If no handlers are registered, the message is acknowledged and discarded — it is not an error (the consumer simply doesn't handle this payload type). This step also acts as a security gate: messages claiming a payload_type that has no registered handlers are dropped before any deserialization or object construction occurs.

Deserialize payload

Call JsonSerializer::deserialize($payload, $eventClass) to reconstruct the typed domain event object from the JSON payload. Deserialization only runs for event classes that passed the handler registry check in the previous step.

Process each handler

For each descriptor, fetch the handler service from handlerLocator, build the callable, and run it through MiddlewareStack::process(). The middleware stack wraps the handler with tracing, logging, hooks, and transaction management.

Acknowledge or reject

If all handlers succeeded, call $consumer->acknowledge($message) — Kafka offset is committed. If any handler failed after all retries, call $consumer->reject($message, false) — offset is committed and the message is written to the dead letter store.

Dev: Live Message Inspection

vortos:kafka:tail streams Vortos events from a Kafka transport to the terminal in real time. It uses a temporary consumer group so it never interferes with your application's consumer offsets.

# tail latest messages on the user.events transport
php bin/console vortos:kafka:tail user.events

# read from the beginning of the topic
php bin/console vortos:kafka:tail user.events --from-beginning

# stop after 20 messages
php bin/console vortos:kafka:tail user.events --limit=20

# override broker address (otherwise resolved from the transport DSN)
php bin/console vortos:kafka:tail user.events --brokers=localhost:9092

Default output — one compact line per message:

Tailing transport: user.events  (topic: user-events, group: vortos-debug-a3f9c2)
Waiting for messages... Ctrl+C to stop.

14:32:01  UserRegistered     agg: usr_019d4784-7b26...  schema: 1  corr: abc-123-def
          {"email":"alice@example.com","name":"Alice"}

14:32:04  UserDeleted        agg: usr_019d4785-93cf...  schema: 1  corr: def-456-ghi
          {"reason":"account_closed"}

^C  Stopped. 2 message(s) read.

Add -v for partition, offset, Kafka timestamp, and all headers formatted line by line:

php bin/console vortos:kafka:tail user.events -v

14:32:01  UserRegistered     agg: usr_019d4784-7b26...  schema: 1  corr: abc-123-def
          partition: 2   offset: 4821   timestamp: 2026-05-22T14:32:01Z
          headers:
            payload_type:      App\User\Domain\Event\UserRegistered
            schema_version:    1
            correlation_id:    abc-123-def
            causation_id:      cmd_019d4784-7b26...
            trace_id:          trace_abc123
          payload:
            {"email":"alice@example.com","name":"Alice"}

The command requires the rdkafka PHP extension. Broker address is resolved from --brokers, then the registered transport DSN, then the KAFKA_BROKERS env var.

Payloads are passed through PayloadSanitizerInterface before display, so any PII masking rules you have configured for the dead letter queue are applied automatically.

Dev: Live Consumer Activity

There are two ways to observe live per-handler activity for a consumer. Both are non-production only.

Option A — `--tail` flag (same process)

Pass --tail to vortos:consume to print activity directly to the terminal from the same process:

php bin/console vortos:consume user-consumer --tail

No Redis required. The worker runs normally and prints each message as it is processed. Ctrl+C stops the worker.

Option B — `vortos:consumer:tail` (observe a running worker)

vortos:consumer:tail attaches to an already-running vortos:consume worker without starting a second consumer or competing for Kafka partitions.

This command requires the Redis cache driver and is not registered in production (kernel.env = prod).

How it works:

vortos:consumer:tail sets a Redis key vortos:tail-ctrl:{name} (TTL 300s) and subscribes to vortos:tail:{name}
The running worker's ConsumerTailControlHook polls for that key before each handler — when it finds it, it activates tail mode
While active, the worker publishes per-handler events to the vortos:tail:{name} Redis pub/sub channel (ephemeral — nothing stored)
vortos:consumer:tail renders events in real time as they arrive
Ctrl+C deletes the Redis key — the worker deactivates on the next handler and publishes stream_end

# Terminal 1 — start the worker (keeps running normally)
php bin/console vortos:consume user-consumer

# Terminal 2 — observe it
php bin/console vortos:consumer:tail user-consumer

Tailing consumer: user-consumer (PID 12345)
Waiting for messages... Ctrl+C to stop.

14:32:01  UserRegistered     agg: usr_019d4784-7b26...  corr: abc-123-def
  ├ SendWelcomeEmailHandler::__invoke    OK      22.0ms
  └ SyncCrmHandler::__invoke             OK       8.3ms

14:32:04  UserDeleted        agg: usr_019d4785-93cf...  corr: def-456-ghi
  └ DeleteUserHandler::__invoke          OK       5.1ms

14:32:09  OrderPlaced        agg: ord_019d4785-a1b2...  corr: xyz-789-abc
  ├ ReserveInventoryHandler::__invoke    OK      14.2ms
  ├ SendConfirmationEmailHandler::__invoke  SKIP     —    (idempotent)
  └ NotifyFulfillmentHandler::__invoke   FAIL    31.7ms  Connection refused

Each message prints a header line (event type, aggregate ID, correlation ID) followed by one indented row per handler showing the handler name, outcome, and latency. The tree characters (├/└) mark whether more handlers follow.

Possible outcomes per handler:

Outcome	Meaning
`OK`	Handler executed and committed on the first attempt
`OK (retried)`	Handler eventually succeeded after one or more retry attempts
`SKIP`	Duplicate message — idempotency key already set, handler skipped
`DISCARD`	Replay limit exceeded — message discarded without running
`DLQ`	All retries exhausted — written to dead letter store
`RETRY`	Intermediate retry attempt failed — handler will be retried

Redis pub/sub is ephemeral — messages published while no subscriber is listening are lost. There is no backlog or persistence.

Unlike vortos:kafka:tail (raw Kafka bytes), both tail modes run the full consumer pipeline — deserialization, middleware, handlers, retry — and report the outcome per handler. Use vortos:kafka:tail to inspect what's on the topic; use the tail options to watch your handlers run.

Kafka and Requeue

Kafka does not support server-side message requeue. The reject($message, requeue: true) call is not available on the Kafka driver — calling it throws a \LogicException:

\LogicException: Kafka does not support server-side requeue. To redeliver,
do not commit the offset — simply omit the acknowledge() call instead of
calling reject(requeue: true).

If you want a message to be redelivered by Kafka, do not call acknowledge() or reject() — the offset will remain uncommitted and Kafka will redeliver when the worker restarts or the partition is rebalanced. The at-least-once guarantee handles this automatically.

Consuming Events

Consuming Events

Writing a Handler

How Discovery Works

Attribute Reference

`#[AsEventHandler]`

Parameters

Multiple Handlers for the Same Event

EventEnvelope — Metadata in Handlers

Header Injection

Built-in header attributes

`#[Header]` — arbitrary header injection

Registering a Consumer

`idempotencyTtl`

`dlq`

Running the Worker

Listing Registered Consumers

Consumer Pipeline Internals

Extract payload type

Resolve handlers

Deserialize payload

Process each handler

Acknowledge or reject

Dev: Live Message Inspection

Dev: Live Consumer Activity

Option A — `--tail` flag (same process)

Option B — `vortos:consumer:tail` (observe a running worker)

Kafka and Requeue

On this page