---
title: "Consuming messages"
description: "Consuming with the PHP core — decode the envelope, validate it (quarantine unsupported versions), route by URN, and (for Kafka/Pulsar) drive the framework-less consumers and consume runtime."
source: https://babelqueue.com/docs/php-sdk/1.x/consuming-messages/
updated: 2026-06-15T00:00:00.000Z
---

# Consuming messages

The core gives you decode + validation + URN routing, and — for the brokers PHP can
consume framework-less (Kafka and Pulsar) — ready-made `consume()` loops plus a
[consume runtime](#the-consume-runtime). The loop that pulls bytes off a broker without a
core consumer is yours (or your framework adapter's).

## Decode

```php
use BabelQueue\Codec\EnvelopeCodec;

$envelope = EnvelopeCodec::decode($rawBody); // plain PHP array
```

## Validate before you dispatch

`EnvelopeValidator::check()` returns a **reason** (or `null` if valid), so you can
quarantine a message you don't understand instead of silently dropping it:

```php
use BabelQueue\Validation\EnvelopeValidator;

if ($reason = EnvelopeValidator::check($envelope)) {
    // e.g. EnvelopeValidator::REASON_UNSUPPORTED_SCHEMA_VERSION
    // → dead-letter / quarantine, don't drop.
    return;
}
```

Reasons include `REASON_MISSING_URN`, `REASON_MISSING_META`,
`REASON_UNSUPPORTED_SCHEMA_VERSION`, `REASON_INVALID_DATA`,
`REASON_MISSING_TRACE_ID`, `REASON_INVALID_ATTEMPTS`. There is also
`EnvelopeValidator::isValid($envelope): bool` and `validate($envelope): void` (which
throws `InvalidEnvelopeException` carrying the reason + envelope).

> `EnvelopeCodec::accepts($envelope)` is a quick boolean check for the same
> consumer-side rules when you don't need the reason.

For a **full structural check** against the bundled canonical JSON Schema — every field,
type and constraint, not just the consumer-side rules — use the offline `SchemaValidator`:

```php
use BabelQueue\Validation\SchemaValidator;

SchemaValidator::isValid($envelope);          // bool
$reason = SchemaValidator::check($envelope);  // "<json-pointer>: <reason>" | null
SchemaValidator::validate($envelope);         // throws on the first violation
```

It's offline and dependency-free (the schema ships in the package), so it's also handy in
tests and CI to assert an envelope you produce is conformant.

## Route by URN

```php
$urn = EnvelopeCodec::urn($envelope); // reads `job`, accepting `urn` as an alias

match ($urn) {
    'urn:babel:orders:created' => $handler->handle($envelope['data'], $envelope['meta']),
    default                    => /* apply your unknown-URN strategy */ null,
};
```

For unmapped URNs, the `BabelQueue\Routing\UnknownUrnStrategy` constants
(`FAIL` / `DELETE` / `RELEASE` / `DEAD_LETTER`) name the standard choices; see
[error handling in the wire contract](/docs/spec/1.x/envelope/#consumer-rules).

## Framework-less consumers (Kafka & Pulsar)

For Redis, RabbitMQ, SQS and Artemis, the broker loop is your framework worker's (the
Laravel drop-in drivers, Symfony Messenger) — the core stays a codec. For **Kafka** and
**Pulsar**, the core ships **complete framework-less consumers** with `receive` / ack /
release primitives and a `consume()` loop:

- **`KafkaConsumer`** (§6, over `ext-rdkafka`) — **process-then-commit** (at-least-once):
  `receive()` polls and decodes a record with `attempts` reconciled (the `bq-attempts`
  header wins, else the body), `commit()` advances the offset, and `consume($handler,
  $shouldStop)` runs the loop, committing only on a clean return.
- **`PulsarConsumer`** (§5, over Pulsar's WebSocket API) — `receive()`, `acknowledge()`
  (the §5 "delete"), `release()` (`negativeAcknowledge`, redeliver), and a `consume()`
  loop. `attempts` is reconciled to `max(body.attempts, redeliveryCount)`.

Both decouple from their broker client behind a one-method seam (`KafkaConsumerClient` /
`PulsarWebSocketConsumerClient`), so the consumer is dependency-free and testable:

```php
use BabelQueue\Transport\PulsarConsumer;

$consumer = new PulsarConsumer($pulsarWebSocketConsumerClient);

$consumer->consume(function ($message) {
    // $message->getUrn(), $message->getData(), $message->attempts()
    // return → acknowledge; throw → release (redeliver)
}, fn () => $shouldStop);
```

## The consume runtime

`consume()` takes any callable. Pass a `Consume\Dispatcher` and you get URN → handler
routing, the four `on_unknown_urn` strategies, and an optional max-attempts dead-letter
cap — composed from pieces the core already ships:

```php
use BabelQueue\Consume\Dispatcher;
use BabelQueue\Consume\DeadLetterPublisher;
use BabelQueue\Contracts\ConsumedMessage;
use BabelQueue\Routing\UnknownUrnStrategy;

$dispatch = (new Dispatcher(
    onUnknownUrn: UnknownUrnStrategy::DEAD_LETTER,
    maxAttempts: 5,
    deadLetters: new DeadLetterPublisher($pulsarProducer), // routes poison to <queue>.dlq
))->on('urn:babel:orders:created', fn (ConsumedMessage $m) => handle($m->getData(), $m->getTraceId()));

$consumer->consume($dispatch, fn () => $shouldStop);
```

A handler that returns acks the message; one that throws redelivers it (at-least-once).
On an unknown URN the strategy applies — `delete` drops it, `dead_letter` routes it to
`<queue>.dlq` (degrading to `delete` when no publisher is set), and `fail` / `release`
throw to redeliver.

### Kafka retry topics

Kafka has no native per-message retry or delay, so the core implements §6.4/§6.5 with the
tiered **retry-topic** pattern: `KafkaRetryRouter::route()` sends a failed record to
`<topic>.retry.<n>` (with `bq-attempts + 1`) or, past the cap, to `<topic>.dlq`; a
`KafkaRetryConsumer` then waits the tier delay and re-injects it into the work topic. See
the [Kafka binding](/docs/spec/1.x/broker-bindings/#apache-kafka).

## Cross-language

Because you decode the [canonical envelope](/docs/spec/1.x/envelope/), the message may
have been produced by **any** BabelQueue SDK — a Go service, a Python worker, a
Node app — and the URN is the only shared contract. See the
[cross-language example](/docs/examples/1.x/cross-language-redis/).