PHP code example of snipershady / queue-bridge

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

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

    

snipershady / queue-bridge example snippets

$queue = QueueBridge::factory()->create(StrategyDtoRedis::fromPhpRedis($redis), new QueueConfig(name: 'orders'));

$queue->producer()->publish('order.created', '{"id":1001,"total":"12.90"}');

$queue->worker()->on('order.created', static function (Message $message, DecodedText $text): void {
    // $text->value is ['id' => 1001, 'total' => '12.90']
})->run();

use QueueBridge\Config\QueueConfig;
use QueueBridge\QueueBridge;
use QueueBridge\Transport\Redis\RedisIdempotencyStore;
use QueueBridge\Transport\Redis\StrategyDtoRedis;

// ext-redis: a factory (instead of an instance) lets the client reconnect after a network failure.
$strategy = StrategyDtoRedis::fromPhpRedis(static function (): Redis {
    $redis = new Redis();
    $redis->connect('127.0.0.1', 6379, 2.0);
    $redis->auth(['app', 'secret']); // ACL user, optional

    return $redis;
});

// or Predis, which reconnects on its own
$strategy = StrategyDtoRedis::fromPredis(new Predis\Client('tcp://127.0.0.1:6379'));

// In production give each process a stable consumer name (host or pod name).
$strategy = new StrategyDtoRedis($strategy->client, consumer: gethostname());

$config = new QueueConfig(name: 'orders', claimIdleMs: 60_000);
$queue = QueueBridge::factory(logger: $logger)->create($strategy, $config);
$queue->setup(); // creates the consumer group; idempotent

// Idempotency store on the same Redis, for IdempotencyMiddleware.
$store = RedisIdempotencyStore::create($strategy->client, $config);

use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use QueueBridge\Transport\RabbitMq\Client\PhpAmqpLibConnector;
use QueueBridge\Transport\RabbitMq\RetryTiers;
use QueueBridge\Transport\RabbitMq\StrategyDtoRabbitMq;

// Connection factory: the connector reopens the channel after every AMQP error and the connection after a drop.
$strategy = StrategyDtoRabbitMq::fromConnectionFactory(
    static fn (): AbstractConnection => new AMQPStreamConnection('rabbitmq', 5672, 'app', 'secret'),
);

// Custom retry tiers and prefetch.
$strategy = new StrategyDtoRabbitMq(
    new PhpAmqpLibConnector(static fn (): AbstractConnection => new AMQPStreamConnection('rabbitmq', 5672, 'app', 'secret')),
    retryTiers: new RetryTiers([1_000, 10_000, 60_000, 600_000]),
    prefetch: 50,
);

$queue = QueueBridge::factory()->create($strategy, new QueueConfig(name: 'orders'));
$queue->setup(); // exchange, quorum queue, wait queues and DLQ; idempotent

use QueueBridge\Transport\InMemory\StrategyDtoInMemory;

$clock = new MyTestClock(); // any PSR-20 clock you can move forward
$queue = QueueBridge::factory(clock: $clock)->create(new StrategyDtoInMemory($clock), new QueueConfig(name: 'orders'));

$queue->producer()->publish('mail', 'welcome', delayMs: 60_000);
$clock->advance(60_000); // the delayed message is now due, without waiting

use QueueBridge\Dsn\StrategyDtoFactory;

$queue = QueueBridge::factory()->create((new StrategyDtoFactory())->fromDsn(getenv('QUEUE_DSN')), new QueueConfig(name: 'orders'));

$producer = $queue->producer();

$id = $producer->publish('orders', $json);                           // returns the MessageId (UUIDv7)
$producer->publish('emails', $text, headers: ['tenant' => 'acme']);
$producer->publish('reminders', $text, delayMs: 60_000);            // not delivered before one minute
$producer->publish('quotes', $text, ttlMs: 30_000);                 // not handled if older than 30 s

$ids = $producer->publishBatch([
    new OutgoingMessage('orders', $first),
    new OutgoingMessage('orders', $second, delayMs: 5_000),
]);

$worker = $queue->worker(new WorkerConfig(
    batchSize: 10,
    receiveTimeoutMs: 1000,
    maxMessages: 10_000,        // exit after N messages,
    maxRuntimeSeconds: 3600,    // or after one hour,
    memoryLimitMb: 256,         // or above this memory: the supervisor restarts the process
    retryStrategy: new ExponentialBackoff(),
    middlewares: [new LoggingMiddleware($logger), new IdempotencyMiddleware($store)],
    listeners: [$metrics],      // WorkerEventListenerInterface
));

$worker
    ->on('orders', new OrderHandler())                       // HandlerInterface
    ->on('emails', static function (Message $message, DecodedText $text): void { /* ... */ })
    ->on('*', $fallback);                                    // channels without a dedicated handler

$reason = $worker->run(); // WorkerExitReason

$multi = new MultiQueueWorker(
    [$ordersWorker, $emailsWorker],
    new WorkerConfig(maxRuntimeSeconds: 3600),
    signalHandler: new PcntlSignalHandler(),
);
$multi->run();

$queue->ping();                      // throws ConnectionException: use it as a readiness probe
$stats = $queue->stats();            // ready, inFlight, delayed, deadLetters, lag, consumers
$dlq = $queue->deadLetters();        // DeadLetterAwareInterface
$dlq->deadLetters(20);               // list<DeadLetter>: message + FailureReason (kind, error class, message, date)
$dlq->requeueDeadLetter($id);        // back to the queue with attempts = 0
$dlq->requeueDeadLetters(100);
$dlq->purgeDeadLetters();

foreach ($queue->browse(50) ?? [] as $queued) {   // null if the backend cannot list messages
    // QueuedMessage: message, state (InFlight | Ready | Delayed), availableAt (delayed messages only)
    echo $queued->message->id->value, ' ', $queued->state->value, PHP_EOL;
}

$removed = $queue->purge();                      // ready + delayed; null if the backend cannot purge

use QueueBridge\Message\FailureKind;
use QueueBridge\Message\FailureReason;
use QueueBridge\Text\TextDecoder;
use QueueBridge\Transport\Capability\MaintenanceAwareInterface;

$transport = $queue->transport();
$decoder = new TextDecoder();

while ($running) {
    if ($transport instanceof MaintenanceAwareInterface) {
        $transport->maintain(); // Redis: promotes delayed messages, recovers orphans
    }

    foreach ($transport->receive(max: 10, timeoutMs: 1000) as $delivery) {
        $decoded = $decoder->decode($delivery->message->text); // never throws

        if (!$decoded->isJson() || !is_array($decoded->value)) {
            $transport->reject($delivery, new FailureReason(
                FailureKind::Unrecoverable, $delivery->message->id, new DateTimeImmutable(), 'Expected a JSON object.',
            ));

            continue;
        }

        try {
            process($decoded->value);
            $transport->ack($delivery);
        } catch (TemporaryFailure) {
            $transport->retry($delivery, 5_000); // new copy with attempts + 1, delivered in 5 s
        }
    }
}

$factory = new QueueFactory(builders: [new SqsTransportBuilder(), new RedisTransportBuilder(), new InMemoryTransportBuilder()]);

bash
docker compose up -d --wait
# arguments: DSN, messages (default 20000), batch size (default 100); disable Xdebug for meaningful numbers
php -d xdebug.mode=off tests/Benchmark/benchmark.php 'redis://127.0.0.1:16379/0' 20000 100
php -d xdebug.mode=off tests/Benchmark/benchmark.php 'redis://127.0.0.1:16379/0?version=6.0'   # 6.0 strategies
php -d xdebug.mode=off tests/Benchmark/benchmark.php 'amqp://guest:[email protected]:35672'