PHP code example of babelqueue / php-sdk

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

babelqueue / php-sdk example snippets

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;
}

// Producer: extra meta keys travel next to the core ones; a collision throws.
$envelope = EnvelopeCodec::make('urn:acme:orders:created', ['order_id' => 7], 'orders',
    meta: ['ext' => ['acme' => ['region' => 'eu']]]);
EnvelopeCodec::make('urn:acme:orders:created', [], 'orders', meta: ['ts' => 1]);   // InvalidEnvelopeException

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 <ttl>. 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)));

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));
});

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

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' => '[email protected]', '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
}

use BabelQueue\Otel\Tracing;

// Produce — a PRODUCER span "publish <urn>"; 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 <urn>", 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)));

bash
composer 

bash
composer uire php-amqplib/php-amqplib    # for AmqpTransport