BabelQueue
27 GitHub Get Started

Consuming messages with the PHP core

Last updated:

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

Decode warnings (forbidden keys)

decode() is tolerant, as the wire contract requires: a message that carries a forbidden key (timestamp, meta.max_retries, meta.attempts, meta.source, meta.ts) still decodes, the key is dropped from the returned array (so a re-encode never carries it), and a warning naming its JSON pointer is reported. A data value that is a non-empty JSON array (a list, not an object) is rejected instead: the result is [], like malformed JSON, with a warning for /data.

By default the warning goes to error_log(). Pass a per-call sink as the second argument (fn (string $message, string $pointer)), or install a process-wide one — for example a PSR-3 logger:

EnvelopeCodec::setWarningHandler(
    fn (string $message, string $pointer) => $logger->warning($message, ['pointer' => $pointer]),
);

$envelope = EnvelopeCodec::decode($rawBody, fn (string $message, string $pointer) => /* per call */ null);

EnvelopeCodec::setWarningHandler(null) restores the error_log() fallback. The keys themselves are listed in EnvelopeCodec::FORBIDDEN_TOP_LEVEL_KEYS and EnvelopeCodec::FORBIDDEN_META_KEYS.

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), and a consume() loop. attempts is reconciled to max(body.attempts, redeliveryCount). A failing handler is retried by re-publishing the message to the same topic with bq-attempts + 1 and acknowledging the original — not by a negative acknowledgement, which does not advance Pulsar’s redelivery count on Exclusive / Failover subscriptions. See Pulsar consumer: retry and the DLQ below.

Both decouple from their broker client behind a small client seam (KafkaConsumerClient / PulsarWebSocketConsumerClient) that you can fake in tests, 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 → retry (re-publish with bq-attempts + 1)
}, 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 a max-attempts cap with dead-lettering — 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: 3,                                          // the default: 3 deliveries in total
    deadLetters: new DeadLetterPublisher($pulsarProducer),   // failed / unknown-URN / poison records go 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 is retried (at-least-once). On an unknown URN the strategy applies — delete drops it, dead_letter routes it to <queue>.dlq, and fail / release throw to retry. The retry is bounded by maxAttempts, the total number of deliveries a message gets, the first included. The default is 3; 0 is still accepted and means no cap. The same cap bounds a failing handler and fail / release of an unknown URN.

  • With a DeadLetterPublisher, a message that reaches the cap is written to <queue>.dlq (reason failed or unknown_urn, dead_letter.attempts = the counted value, which is 3 with the default). The queue is taken from the consumer’s own topic / queue, not from the envelope’s meta.queue, and default.dlq is never written. When the consumer’s queue is unknown the message cannot be addressed: it is acknowledged, dropped and logged.
  • Without a publisher, a message that reaches the cap is acknowledged, dropped and logged on Pulsar (the $onDrop argument replaces the default error_log, and the body is never logged). on_unknown_urn: dead_letter without a publisher degrades to delete the same way, and the drop is now logged instead of silent.
  • Poison never reaches a handler. A body that does not decode into an envelope, or has neither a valid job nor a valid urn, is not an unknown URN. With a publisher it is written to <queue>.dlq at once, without retry, as a poison record (reason: "poison", the original body kept); delete drops it; with no publisher fail / release take the bounded path above and dead_letter drops it.
  • A version-gated Pulsar message is settled on its first delivery. A delivery whose bq-schema-version property is present, not blank and not "1" never reaches a handler, an unknown-URN decision or a retry, whatever bq-attempts, the redelivery count or maxAttempts say. With no DeadLetterPublisher (or a queue that cannot be named) it is acknowledged, dropped and logged through $onDrop. delete drops it too, and the dead_letter strategy degrades to delete the same way. Otherwise the raw body (byte for byte, with no dead_letter block and no poison wrapping; the incoming routing properties ride on it except bq-attempts and bq-schema-version) is written to <queue>.dlq with DeadLetterPublisher::publishRaw() and the delivery is acknowledged. fail and release behave alike here: a version-gated delivery is never released. A failed write throws Consume\DeadLetterWriteFailed, so PulsarConsumer::consume() negatively acknowledges it: nothing is acknowledged or retried, the gate rejects the redelivery again, and the write is tried again. The gate reads only the strategy you configured. Kafka is out of scope: a version-gated Kafka record keeps the unknown-URN path.
  • On Kafka the Dispatcher never drops. Over KafkaConsumer, a message that reaches the cap without a publisher is rethrown, so a KafkaRetryRouter (or the consumer’s own rules) takes the terminal decision. With a publisher it dead-letters to <topic>.dlq, <topic> being the record’s source topic (else the topic the consumer is subscribed to; when that is ambiguous the record is committed, dropped and logged). Because the default cap is now 3, a Dispatcher with a publisher dead-letters before a router with its default of 5: pass maxAttempts: 0 to leave the cap to the router.

Behaviour change. The default maxAttempts moved from 0 (no cap) to 3. A deployment that relied on no cap must now pass maxAttempts: 0 explicitly.

Behaviour change (ADR-0037, since 1.27.0). A version-gated Pulsar message used to take the unknown-URN path, so under fail / release it was retried through the bounded re-publish before the terminal decision, and its record was the {"raw": ...} envelope with a dead_letter block (reason: "unknown_urn"). It is now settled on its first delivery and written as the raw body, as the Pulsar binding describes. A reader of <queue>.dlq recognises that record by the absence of a dead_letter block.

Pulsar consumer: retry and the DLQ

PulsarConsumer and the Dispatcher bound every message the same way on every subscription type: a retry re-publishes the message to the same topic with bq-attempts + 1 (the envelope’s attempts kept in sync) and acknowledges the original. A failed re-publish negatively acknowledges the original instead, and a failed dead-letter write does too (Consume\DeadLetterWriteFailed), so nothing is lost.

The consumer needs a producer for that re-publish, and it derives one from your client; a value you pass wins. The SDK’s own PulsarWebSocketSubscription supplies everything, so this is the whole setup, and the cap holds on Exclusive and Failover subscriptions:

use BabelQueue\Transport\PulsarConsumer;
use BabelQueue\Transport\PulsarWebSocketSubscription;

$consumer = new PulsarConsumer(
    new PulsarWebSocketSubscription('http://pulsar:8080', 'persistent://public/default/orders', 'my-sub'),
);

Your own PulsarWebSocketConsumerClient adapter works too: implement PulsarServiceEndpoint (or pass serviceUrl:, plus serviceAuthToken: on an authenticated cluster) or retryProducer: for the producer, and PulsarConsumerDescriptor (or queue: with tenant: and namespace:) so the consumer knows its full topic.

The constructor fails closed. If no producer can be derived, it throws a BabelQueueException unless the subscription is known to be Shared or Key_Shared, where a negative acknowledgement does advance the redelivery count. An unknown subscription type counts as Exclusive, Pulsar’s own default. To keep a nack-only consumer, declare it with subscriptionType: 'Shared'; otherwise give it a producer.

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> (n is the 1-based tier index in ascending-delay order, e.g. orders.retry.1; 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.