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 loop that pulls bytes off a broker without a core consumer is yours (or your framework adapter’s).

Decode

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:

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:

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

$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.

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:

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:

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.

Cross-language

Because you decode the canonical 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.