PHPackages babelqueue/php-sdk - PHPackages - PHPackages [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register) 1. [Directory](/) 2. / 3. [Caching](/categories/caching) 4. / 5. babelqueue/php-sdk ActiveLibrary[Caching](/categories/caching) babelqueue/php-sdk ================== Framework-agnostic core for BabelQueue: the canonical polyglot queue envelope codec, contracts and dead-letter helpers. Framework adapters (Laravel, Symfony, ...) are built on top. v1.16.0(1mo ago)1281↓87.5%3MITPHPPHP ^8.2CI passing Since Jun 6Pushed 1mo agoCompare [ Source](https://github.com/BabelQueue/php-sdk)[ Packagist](https://packagist.org/packages/babelqueue/php-sdk)[ Docs](https://babelqueue.com)[ GitHub Sponsors](https://github.com/muhammetsafak)[ RSS](/packages/babelqueue-php-sdk/feed)WikiDiscussions main Synced 1w ago READMEChangelog (10)Dependencies (11)Versions (21)Used By (3) BabelQueue PHP SDK (core) ========================= [](#babelqueue-php-sdk-core) [![License: MIT](https://camo.githubusercontent.com/8bb50fd2278f18fc326bf71f6e88ca8f884f72f179d3e555e20ed30157190d0d/68747470733a2f2f696d672e736869656c64732e696f2f62616467652f6c6963656e73652d4d49542d677265656e2e737667)](LICENSE) > The **framework-agnostic core** of BabelQueue for PHP — the canonical polyglot queue envelope codec, contracts and dead-letter helpers. Framework adapters (`babelqueue/laravel`, `babelqueue/symfony`, …) are built on top of this. You usually don't install this directly — you install an adapter: ``` composer require babelqueue/laravel # Laravel composer require babelqueue/symfony # Symfony (Messenger) ``` …and the adapter pulls this core in. Install it directly only for a framework-less PHP app, or to build a new adapter. ``` composer require babelqueue/php-sdk ``` What's in here -------------- [](#whats-in-here) AreaClassRoleCodec`BabelQueue\Codec\EnvelopeCodec`Build / encode / decode the canonical `{job, trace_id, data, meta, attempts}` envelope (`schema_version` 1). **The single PHP implementation of the wire format** — every adapter reuses it, so Laravel and Symfony can't drift.Contracts`BabelQueue\Contracts\PolyglotJob`Producible message: `getBabelUrn()` + `toPayload()`.`BabelQueue\Contracts\HasBabelUrn` / `HasTraceId`URN identity / optional trace-id propagation.`BabelQueue\Contracts\InboundMessage`Read-only decoded view of a consumed envelope.`BabelQueue\Contracts\Transport`Minimal publish seam (framework-less / adapter use).Validation`BabelQueue\Validation\EnvelopeValidator`Consumer-side validation **with a reason** — quarantine an unsupported `meta.schema_version` instead of dropping it.Transports`BabelQueue\Transport\RedisTransport` / `AmqpTransport`Optional framework-less reference `Transport` impls (Redis `RPUSH`; RabbitMQ durable + contract AMQP properties).Dead-letter`BabelQueue\DeadLetter\DeadLetter`Annotate an envelope with the additive `dead_letter` block (ADR-0009).Idempotency`BabelQueue\Idempotency\Idempotent` / `IdempotencyStore` / `InMemoryStore`Dedupe on `meta.id` (ADR-0022): `Idempotent::wrap($store, $handler)` skips an already-processed message.`BabelQueue\Idempotency\ClaimingStore` / `PdoStore` / `RedisStore` / `ClaimingDispatch`**Persistent** stores for a fleet: an **atomic claim** (PDO unique-INSERT / Redis `SET NX PX`) serializes concurrent deliveries of one id. `PdoStore` is `ext-pdo`-only; `RedisStore` reuses the optional `predis` client.Redrive`BabelQueue\Redrive\Redrive` / `RedriveIO` / `RedriveOptions`Safe DLQ replay (ADR-0026): reset + dry-run + sandbox + select, driven through a `RedriveIO` you bind to your broker.`BabelQueue\Redrive\ReplayBypass` / `HeaderRedriveIO`Replay-bypass (ADR-0027): a redrive can stamp `bq-replay-bypass` so a handler skips already-fired external effects (`bypassExternalEffects`).Outbox`BabelQueue\Outbox\Outbox` / `OutboxRelay` / `OutboxStore`Transactional outbox (ADR-0029): persist the message **atomically with the business write**, relay it later. Dependency-free — `OutboxStore` is an interface you bind to your DB.GDPR`BabelQueue\Gdpr\Gdpr` / `Cipher` / `OpenSslCipher`Opt-in field-level encryption (ADR-0030) of the `data` fields a schema marked `x-gdpr-sensitive`: `Gdpr::protect()` before publish, `Gdpr::unprotect()` after decode. The envelope stays frozen — a sensitive value becomes a ciphertext **string**. The `Cipher` is a caller-provided seam (bind a KMS/Vault); the reference `OpenSslCipher` (AES-256-GCM) is `ext-openssl`-only and a `suggest`.Tracing`BabelQueue\Otel\Tracing`Optional OpenTelemetry produce/consume spans (ADR-0025/0028): correlate across hops via `trace_id`, and — when the transport carries headers — link spans across hops via a W3C `traceparent`. Opt-in; `open-telemetry/api` is a `suggest`.Headers`BabelQueue\Contracts\HeaderPublisher` / `HasHeaders`The out-of-band transport-header seam (ADR-0027/0028): publish headers **beside** the frozen envelope, and surface them on a consumed message.Routing`BabelQueue\Routing\UnknownUrnStrategy``fail` / `delete` / `release` / `dead_letter` constants.Support`BabelQueue\Support\Uuid`Dependency-free UUIDv4 (no ramsey/symfony-uid needed).Errors`BabelQueue\Exceptions\BabelQueueException` / `UnknownUrnException` / `InvalidEnvelopeException`Exception hierarchy; `InvalidEnvelopeException` carries the rejection reason + envelope.The contract this core implements — the canonical envelope, URN scheme, broker bindings and versioning policy — is documented at [babelqueue.com](https://babelqueue.com). The golden conformance fixtures live in [`tests/fixtures/`](tests/fixtures/) — every PHP package must round-trip them. Framework-less use ------------------ [](#framework-less-use) Produce the canonical envelope from a plain PHP app and let any other SDK consume it. The reference transports keep the core dependency-free — install only the broker client you use: ``` composer require predis/predis # for RedisTransport composer require php-amqplib/php-amqplib # for AmqpTransport ``` ``` use BabelQueue\Codec\EnvelopeCodec; use BabelQueue\Transport\RedisTransport; use BabelQueue\Validation\EnvelopeValidator; // Produce — a Go/Python/Node consumer reads the identical envelope off "orders". $transport = new RedisTransport(new Predis\Client('redis://localhost:6379')); $transport->publish(EnvelopeCodec::encode(EnvelopeCodec::fromJob($job, 'orders')), 'orders'); // Consume (your own loop) — validate before dispatch, quarantine the unknown. $envelope = EnvelopeCodec::decode($rawBody); if ($reason = EnvelopeValidator::check($envelope)) { // $reason === 'unsupported_schema_version' → dead-letter, don't drop. return; } ``` phpredis (`ext-redis`) users can implement the one-method `Transport` directly — it is just an `rpush`. Idempotency — dedupe on `meta.id` (ADR-0022) -------------------------------------------- [](#idempotency--dedupe-on-metaid-adr-0022) BabelQueue is **at-least-once**, so handlers should be idempotent — dedupe on the envelope's `meta.id`. `Idempotent::wrap($store, $handler)` makes that a one-liner: a previously-succeeded id is skipped + acked; a throw leaves it unmarked so retry/DLQ still apply. The reference `InMemoryStore`is single-process; for a **fleet** of consumers two **persistent** stores share one dedupe record and add an **atomic claim** so two workers handed the same id never both run (GR-7 — nothing added to `require`): ``` use BabelQueue\Idempotency\ClaimingDispatch; use BabelQueue\Idempotency\PdoStore; use BabelQueue\Idempotency\RedisStore; // PDO — Postgres / MySQL / SQLite. Atomic claim = a unique-key INSERT caught as a duplicate // (every engine enforces the PRIMARY KEY atomically — no dialect-specific upsert). Ship the table: $pdo->exec(PdoStore::ddl()); // CREATE TABLE IF NOT EXISTS bq_idempotency (...) $store = new PdoStore($pdo); // Redis — atomic claim = SET key value NX PX . Reuses your predis client. $store = new RedisStore($predis); // ClaimingDispatch drives claim → run → commit; a duplicate skips, a concurrent in-flight // delivery parks (throws ClaimParkedException → redelivered, not acked), a throw releases the claim. $dispatch->on('urn:babel:orders:created', ClaimingDispatch::wrap($store, fn ($m) => handle($m))); ``` A claim's TTL bounds a crash between claim and commit, after which a redelivery may re-run — still at-least-once, **not** exactly-once (the dual-write window; the outbox below narrows the produce side). The frozen `IdempotencyStore` interface is untouched; the claim contract is the opt-in `ClaimingStore` extension. Safe DLQ replay + replay-bypass (ADR-0026 / ADR-0027) ----------------------------------------------------- [](#safe-dlq-replay--replay-bypass-adr-0026--adr-0027) `Redrive` replays dead-lettered messages back to their source (or a sandbox) — reset for reprocessing, with dry-run and `select`, driven through a `RedriveIO` you bind to your broker. Replaying into the *real* queue re-runs the handler, so its external effects (charge, email) would re-fire. **Replay-bypass** closes that: a redrive can stamp the out-of-band `bq-replay-bypass`transport header (beside the frozen envelope, never in it — GR-1), and a handler wraps its external, non-idempotent side in `ReplayBypass::bypassExternalEffects()` to skip it on a replay while the idempotent core still runs. ``` use BabelQueue\Redrive\Redrive; use BabelQueue\Redrive\RedriveOptions; use BabelQueue\Redrive\ReplayBypass; // Producer side — stamp the marker on a replay (needs a RedriveIO that also implements // HeaderRedriveIO; over a plain RedriveIO bypass is a best-effort no-op and item->bypassed = false). Redrive::run($io, 'orders.dlq', new RedriveOptions(bypass: true)); // Consumer side — the idempotent core always runs; the external effect is skipped on a replay. $dispatch->on('urn:babel:orders:created', static function ($m): void { saveOrder($m); // idempotent core ReplayBypass::bypassExternalEffects($m, static fn () => sendConfirmationEmail($m)); }); ``` The marker `bq-replay-bypass` is identical across SDKs (Go's `HeaderReplayBypass`), so a Go-produced replay is recognised by a PHP consumer. It rides the same `HeaderPublisher` / `HasHeaders` seam as the OTel `traceparent`; `schema_version` stays **1** and `trace_id` is preserved. Transactional outbox (ADR-0029) ------------------------------- [](#transactional-outbox-adr-0029) A plain producer makes a **dual write** — commit the business row *and* publish to the broker — two systems that can disagree on a crash. The outbox removes it: the message is written into the **same database, in the same transaction** as the business data (so they commit or roll back atomically), and a separate **relay** publishes it afterwards. No distributed transaction; exactly-once *handoff* into the broker (then at-least-once on the wire, deduped on `meta.id` by the consumer-side `Idempotent::wrap`, ADR-0022). The helper is dependency-free (GR-7): the core defines `OutboxStore` and you bind it to your DB. **The transaction boundary is yours** — `OutboxStore::save()` runs inside the transaction *you* opened around your business write. The envelope is stored **verbatim**(GR-1) and relayed byte-for-byte, so `trace_id` is preserved end-to-end (GR-4). ``` use BabelQueue\Codec\EnvelopeCodec; use BabelQueue\Outbox\Outbox; use BabelQueue\Outbox\OutboxRelay; // WRITE side — one transaction for the business row AND the message (your tx boundary). $outbox = new Outbox($store); // $store implements OutboxStore (your DB) $db->transaction(function () use ($db, $outbox, $order): void { $db->insertOrder($order); // business write $outbox->write(EnvelopeCodec::make('urn:babel:orders:created', $order, 'orders')); }); // both commit, or neither // READ side — a worker/cron drains the durable rows onto the broker. $relay = new OutboxRelay($transport, $store); // $transport is any BabelQueue Transport $relay->drain(); // publishes verbatim, marks published/failed ``` `OutboxStore` is four methods — `save()`, `fetchUnpublished()`, `markPublished()`, `markFailed()`. A reference `InMemoryOutboxStore` ships for tests; a runnable **InitORM-backed** adapter + the outbox-table DDL live in [`babelqueue-examples/outbox-initorm/`](https://github.com/BabelQueue/babelqueue-examples/tree/main/outbox-initorm), keeping this core DB-free. GDPR field encryption (ADR-0030) -------------------------------- [](#gdpr-field-encryption-adr-0030) When a schema marks a `data` field `x-gdpr-sensitive`, BabelQueue can **encrypt that field's value end-to-end**: the producer encrypts it before publish, the consumer decrypts it after decode. It is opt-in and standalone — a producer/consumer that never calls it behaves exactly as before. The **envelope stays frozen** (GR-1): only the sensitive *value* changes — it becomes a ciphertext **string**, so `data` stays pure JSON (GR-3), `meta.schema_version` stays `1`, and `trace_id` is untouched (GR-4). An SDK or hop without the key still carries the envelope byte-compatibly; it just can't read the protected fields. The crypto is a **caller-provided `Cipher`** (GR-7 — the core pulls no crypto dependency, it stays `ext-json`). Bind it to your KMS / Vault / HSM / tokenisation service, or use the bundled reference `OpenSslCipher` (AES-256-GCM) — which needs `ext-openssl`, an **optional** `suggest`: ``` # only if you use the reference cipher; otherwise bind your own Cipher onto a KMS/Vault docker-php-ext-enable openssl # ext-openssl is usually already present ``` ``` use BabelQueue\Codec\EnvelopeCodec; use BabelQueue\Gdpr\Gdpr; use BabelQueue\Gdpr\OpenSslCipher; use BabelQueue\Schema\SchemaValidated; $cipher = new OpenSslCipher($key); // 32 bytes ⇒ AES-256-GCM; key is yours to manage $schema = $provider->schemaFor('urn:babel:orders:created'); // PRODUCE — validate CLEARTEXT first, then encrypt marked leaves in place, then encode. $data = ['email' => 'alice@example.com', 'order_id' => 1042]; if ($schema !== null) { SchemaValidated::assert($provider, 'urn:babel:orders:created', $data); // cleartext Gdpr::protect($data, $schema, $cipher); // 'email' becomes a ciphertext string, in place } $body = EnvelopeCodec::encode(EnvelopeCodec::make('urn:babel:orders:created', $data, 'orders')); // CONSUME — decode, decrypt marked leaves in place, THEN read/validate cleartext. $envelope = EnvelopeCodec::decode($body); $data = $envelope['data']; if ($schema !== null) { Gdpr::unprotect($data, $schema, $cipher); // restores 'email' byte-for-byte } ``` Mark a field in the `data` schema with the `x-gdpr-sensitive` keyword — the boolean `true`, or a free-form category string for documentation. It is **validation-neutral** (it never makes a value valid or invalid), so annotating a schema is never a breaking change: ``` { "type": "object", "properties": { "email": {"type": "string", "x-gdpr-sensitive": "email"}, "profile": {"type": "object", "properties": { "full_name": {"type": "string", "x-gdpr-sensitive": true} }}, "addresses": {"type": "array", "items": {"type": "object", "properties": { "line": {"type": "string", "x-gdpr-sensitive": true} }}} } } ``` `Gdpr::protect()` / `unprotect()` walk every marked leaf — nested objects (`profile.full_name`), array items (`addresses[].line`), scalar array items, and a root mark — and rewrite each **in place**. Each value is canonically JSON-encoded then replaced by the ciphertext string (and the exact inverse on the way back), so the round-trip restores the original value **byte-for-byte**. Behaviour at the edges: - **Validate cleartext** — run schema validation *before* `protect()` and *after* `unprotect()`; a constrained sensitive field (`minLength`, `enum`, …) would reject the ciphertext string otherwise. - **Absent field** — a marked path missing from a message is skipped (not an error); schemas evolve. - **Idempotent decrypt** — a non-string leaf is left untouched, so `unprotect()` is safe to re-run on already-cleartext data. - **Wrong key / tampered** — `unprotect()` throws the typed `BabelQueue\Gdpr\DecryptException`, so the consumer fails the message (retry / dead-letter) rather than process unreadable PII. The sensitive marks are read from the same per-URN `data` schema the validation path already loads (`BabelQueue\Schema\SensitivePaths` exposes them, mirroring babelqueue-registry's inventory). The registry **declares and audits** sensitivity; this is the SDK that **enforces** it on the wire. OpenTelemetry tracing (ADR-0025 / ADR-0028) ------------------------------------------- [](#opentelemetry-tracing-adr-0025--adr-0028) `BabelQueue\Otel\Tracing` adds **optional** OpenTelemetry spans to a producer or consumer. It is opt-in and dependency-light — `open-telemetry/api` is only a `suggest`, so the core stays `ext-json` (GR-7). Cross-hop trace propagation layers two levels: - **`trace_id` ↔ TraceId** (v0.1): the envelope's `trace_id` maps 1:1 to an OTel trace id, so every hop that shares a `trace_id` shares one trace — correlation + per-hop timing with **zero**wire/transport change. Preserved end-to-end (GR-4). - **W3C `traceparent`** (v0.2): the producer also injects the active span context as a `traceparent`**transport header** — beside the frozen envelope, never in it (GR-1) — so the consumer starts its span as a true **child** of the producer span (real cross-hop parent-child linkage). With no `traceparent` present it falls back to the v0.1 behaviour — a strict, backward-compatible upgrade. ``` composer require open-telemetry/api # + open-telemetry/sdk to export ``` ``` use BabelQueue\Otel\Tracing; // Produce — a PRODUCER span "publish "; the active span's traceparent rides the transport // header when the transport carries headers (AMQP/SQS/Redis below), else it degrades to a plain // publish with the trace_id still stamped (no error). Tracing::publish($tracer, $transport, 'urn:babel:orders:created', ['order_id' => 1042], 'orders'); // Consume — a CONSUMER span "process ", started as a child of the producer span when the // delivered message surfaces a traceparent (HasHeaders); else from the trace_id (v0.1). $dispatch->on('urn:babel:orders:created', Tracing::wrap($tracer, fn ($m) => handle($m))); ``` The `traceparent` rides the out-of-band `HeaderPublisher` / `HasHeaders` seam (the same seam as the replay-bypass marker, ADR-0027). Which reference transports carry it: TransportCarries `traceparent`Carrier`RedisTransport`**Yes**A transport-owned `__bq_frame` JSON frame wrapping the bare envelope (Redis has no native per-message metadata channel). A header-less publish stays byte-for-byte bare; `unframe()` is back-compatible with bare/cross-version values.`AmqpTransport`**Yes**AMQP message headers, beside the contract `x-*` headers (contract wins a collision).`SqsTransport`**Yes**SQS `MessageAttributes` (String), beside the contract `bq-*` attributes (contract wins; bounded by the 10-attribute SQS cap).`KafkaTransport` / `PulsarTransport` / `StompTransport`DeferredThese producers already project rich `bq-*`/`bq_*` headers; wiring `traceparent` through their retry-topic / annotation machinery is a documented follow-up. Until then they degrade to v0.1 `trace_id` correlation (no error). `KafkaMessage` already surfaces its record headers via `HasHeaders`, the consume-side hook a future Kafka producer would reach.A plain `publish()` (no tracer) is unchanged on every transport. Design ------ [](#design) This core is the **contract runtime**, not a worker. It does not own a broker loop or retry — adapters bind to each framework's native queue (Laravel's drop-in driver, Symfony Messenger) and reuse that framework's worker/retry. Testing ------- [](#testing) ``` composer install vendor/bin/phpunit ``` License ------- [](#license) MIT © Muhammet Şafak. ### Health Score 47 — FairBetter than 93% of packages Maintenance93 Actively maintained with recent releases Popularity17 Limited adoption so far Community11 Small or concentrated contributor base Maturity55 Maturing project, gaining track record Bus Factor1 Top contributor holds 100% of commits — single point of failure How is this calculated?**Maintenance (25%)** — Last commit recency, latest release date, and issue-to-star ratio. Uses a 2-year decay window. **Popularity (30%)** — Total and monthly downloads, GitHub stars, and forks. Logarithmic scaling prevents top-heavy scores. **Community (15%)** — Contributors, dependents, forks, watchers, and maintainers. Measures real ecosystem engagement. **Maturity (30%)** — Project age, version count, PHP version support, and release stability. ### Release Activity Cadence Every ~1 days Total 20 Last Release 33d ago Major Versions v0.3.0 → v1.0.02026-06-07 ### Community Maintainers ![](https://www.gravatar.com/avatar/4b6b34f3ac8938d8ee52ba3bd260680855dc5715c7b2929d9380de30d15a67dd?d=identicon)[muhammetsafak](/maintainers/muhammetsafak) --- Top Contributors [![muhammetsafak](https://avatars.githubusercontent.com/u/104234499?v=4)](https://github.com/muhammetsafak "muhammetsafak (35 commits)") --- Tags babelqueuebroker-agnosticdistributed-tracinginteroperabilityjsonlaravelmessage-queuemessagingmicroservicesphpphp-sdkpolyglotqueuerabbitmqredisserializationsymfonyjsonqueuemicroservicespolyglotenvelope ### Code Quality TestsPHPUnit Static AnalysisPHPStan Type Coverage Yes ### Embed Badge ![Health badge](/badges/babelqueue-php-sdk/health.svg) ``` [![Health](https://phpackages.com/badges/babelqueue-php-sdk/health.svg)](https://phpackages.com/packages/babelqueue-php-sdk) ``` ### Alternatives [spatie/valuestore Easily store some values 7681.1M56](/packages/spatie-valuestore)[enqueue/redis Message Queue Redis Transport 445.7M32](/packages/enqueue-redis)[mjphaynes/php-resque Redis backed library for creating background jobs and processing them later. 228201.4k2](/packages/mjphaynes-php-resque)[javibravo/simpleue Php package to manage queue tasks in a simple way 131348.6k1](/packages/javibravo-simpleue)[aplus/cache Aplus Framework Cache Library 171.6M4](/packages/aplus-cache)[resquebundle/resque A Symfony 4 bundle to manage Resque job queues 51139.4k1](/packages/resquebundle-resque) PHPackages © 2026 [Directory](/)[Categories](/categories)[Trending](/trending)[Changelog](/changelog)[Analyze](/analyze)