Redis Queue Client
A simple and efficient abstraction layer for publishing and consuming messages from Redis queues using the byjg/message-queue-client framework.
Installation
composer require "byjg/redis-queue-client"
Documentation
Usage
Publish
<?php
// Register the connector and associate with a scheme
ConnectorFactory::registerConnector(RedisQueueConnector::class);
// Create a connector
$connector = ConnectorFactory::create(new Uri("redis://$user:$pass@$host:$port"));
// Create a queue
$pipe = new Pipe("test");
$pipe->withDeadLetter(new Pipe("dlq_test"));
// Create a message
$message = new Message("Hello World");
// Publish the message into the queue
$connector->publish(new Envelope($pipe, $message));
Consume
<?php
// Register the connector and associate with a scheme
ConnectorFactory::registerConnector(RedisQueueConnector::class);
// Create a connector
$connector = ConnectorFactory::create(new Uri("redis://$user:$pass@$host:$port"));
// Create a queue
$pipe = new Pipe("test");
$pipe->withDeadLetter(new Pipe("dlq_test"));
// Connect to the queue and wait to consume the message
$connector->consume(
$pipe, // Queue name
function (Envelope $envelope) { // Callback function to process the message
echo "Process the message";
echo $envelope->getMessage()->getBody();
return Message::ACK;
},
function (Envelope $envelope, $ex) { // Callback function to process the failed message
echo "Process the failed message";
echo $ex->getMessage();
return Message::REQUEUE;
}
);
The consume method will wait for a message and call the callback function to process the message. If there is no message in the queue, the method will wait until a message arrives.
If you want to exit the consume method, just return Message::ACK | Message::EXIT from the callback function.
Possible return values from the callback function:
Message::ACK- Acknowledge the message and remove from the queueMessage::NACK- Not acknowledge the message and remove from the queue. If the queue has a dead letter queue, the message will be sent to the dead letter queue.Message::REQUEUE- Requeue the messageMessage::EXIT- Exit the consume method
Messages are taken with BLMOVE into a processing list (<queue>:processing) and removed from it only
after the outcome is done, so a consumer that dies mid-processing does not lose the message: the next
consume() puts it back on the queue. Delivery is at-least-once, and Redis 6.2 or later is required.
When several consumers read the same queue, give each one a stable identification
(consume($pipe, $onReceive, $onError, "worker-1")). See Reliable Delivery.
Protocols
| Protocol | URI Example | Notes |
|---|---|---|
| Redis | redis://user:pass@host:port | Default port: 6379 |
Running Tests
docker compose up -d
vendor/bin/phpunit