PHP and Laravel

The PHP client is a plain PSR-4 library (Queen\ mapped onto src/) with an optional Laravel layer on top. It is synchronous: every builder chain ends in an explicit execute(), pop(), get() or commit(). It requires PHP 8.3 and depends on guzzlehttp/guzzle and ramsey/uuid.

composer require smartpricing/queen-mq

Plain PHP

use Queen\Queen;

$client = new Queen('http://localhost:6632');

The constructor takes a URL string, a list of URLs (a sequentially keyed array), or an associative config array. One URL is a direct client; several build a load balancer with failover. Passing neither url nor urls throws InvalidArgumentException.

$client = new Queen([
    'urls' => ['http://broker-a:6632', 'http://broker-b:6632'],
    'bearerToken' => getenv('QUEEN_TOKEN'),
    'timeoutMillis' => 30000,
    'loadBalancingStrategy' => 'affinity',
]);

Client options

Defaults come from Queen\Support\Defaults::CLIENT_DEFAULTS.

Push

$results = $client->queue('orders')
    ->partition('acct-42')
    ->push([['transactionId' => 'order-1-created', 'data' => ['id' => 1]]])
    ->execute();

push() takes one associative array or a list of them, and returns a PushBuilder that does nothing until execute(). The body is data, or payload, or the array itself; a missing transactionId is minted as a UUIDv7. execute() returns the broker’s per-item array with status set to queued, duplicate, buffered or failed.

onSuccess, onDuplicate and onError can be chained before execute() to bucket the response by status. Without an onError handler, a failed item makes execute() throw.

Pop

$messages = $client->queue('orders')->group('billing')->batch(10)->pop();

pop() returns an array of message arrays. wait starts at true (it is initialised from the consume defaults), so a bare pop() is a 30-second long poll; use ->wait(false) for an immediate return and ->timeoutMillis($ms) to change the window. ->batch($n) caps the messages returned, ->partitions($n) claims up to n partitions in one call with batch as the global cap, and ->autoAck(true) makes the server commit the cursor inside the pop transaction (at-most-once).

Consume with a closure

$client->queue('jobs')
    ->group('workers')
    ->batch(10)
    ->each()
    ->consume(function (array $message) {
        process($message['data']);
    })
    ->execute();

consume() returns a ConsumeBuilder; execute() runs the worker loop and blocks. The builder methods are group, concurrency, batch, partitions, each, limit, idleMillis, autoAck, wait, timeoutMillis, renewLease, subscriptionMode and subscriptionFrom (the same set as the other SDKs, in camelCase).

With concurrency(1) a single loop runs synchronously. Above 1, the client uses Guzzle’s async pool so the long polls overlap on one cURL multi-handle; message processing stays sequential per worker. A 429 in the concurrent path backs off once for the whole round rather than once per worker, because all the workers share the same tenant bucket.

Without each(), the closure receives the array of popped messages, even with batch(1). Only each() makes it receive one message at a time. autoAck is client-side: the loop calls POST /api/v1/ack after your closure returns (status completed) or throws (status failed).

ConsumerManager::start() installs SIGINT and SIGTERM handlers that stop the loop, and restores the previous handlers in a finally block, so it is safe inside a larger application. Signal handling needs the pcntl extension; without it the loop has no signal handling at all and only limit or idleMillis will stop it.

The high-level consumer

The rdkafka-shaped alternative gives you the loop instead of a callback:

$consumer = $client->queue('orders')->group('processors')->batch(10)->getConsumer();
$consumer->subscribe();

while (!$consumer->isClosed()) {
    $message = $consumer->consume(1000);
    if ($message === null) {
        continue;
    }
    process($message);
    $consumer->ack($message);
}

$consumer->close();

subscribe() builds the pop path and installs the signal handlers. consume($timeoutMs) returns one message or null; consumeBatch($timeoutMs, $max) returns an array. Both force wait=true and override the batch size and timeout on every call, so the long-poll window is whatever you pass, not what the builder held. ack, nack and renewLease delegate to the client; isClosed() dispatches pending signals each time it is called, which is what makes the while loop interruptible. close() only flips the flag. It does not flush push buffers, so call $client->close() too if you buffered.

consume() and consumeBatch() return null / [] on a timeout or a connection error but rethrow everything else, including a terminal 403.

Acknowledge manually

$client->ack($messages, true, ['group' => 'billing']);

ack() takes one message array, a list of them, or a bare transactionId string. $status is true (completed), false (failed), or the string completed, failed, retry or dlq. $context['group'] names the consumer group; $context['error'] records the failure reason that lands on the DLQ row if the nack exhausts the retry budget. Set _status and _error on individual messages to give them different outcomes in one batch call.

partitionId is mandatory: a batch ack without it throws InvalidArgumentException, a single ack returns ['success' => false, 'error' => …].

Transactions

$client->transaction()
    ->queue('stage-2')->partition('acct-42')->push([['data' => ['value' => 2]]])
    ->ack($message)
    ->commit();

commit() sends every collected push and ack as one PostgreSQL transaction, all or nothing, together with the acked messages’ leaseIds as requiredLeases, so the commit fails if a lease expired. Committing with no operations throws.

Buffering, admin, DLQ

$client->queue('events')
    ->buffer(['messageCount' => 500, 'timeMillis' => 200])
    ->push([['data' => ['seen' => time()]]])
    ->execute();

Buffered pushes return ['buffered' => true, 'count' => N] and flush on a count or time trigger (defaults 100 and 1000 ms), keyed per queue/partition. The buffer is in process memory: $client->flushAllBuffers() or $client->close() before exiting, or lose it. $client->getBufferStats() reports the counters.

$client->admin() mirrors the JavaScript admin surface in camelCase: resources, messages, traces, status and analytics, consumer groups, health, metrics and the maintenance switches. The route list with access levels is in Reference.

$dlq = $client->queue('orders')->dlq('billing')->limit(50)->get();

429 and 403

Retry429Policy implements the same policy as the other SDKs: a 429 is retried in place against the same backend, honouring Retry-After (seconds) when present and otherwise backing off baseMs * 2^attempt capped at capMs, both jittered. Ordinary requests get ten attempts; a long-poll pop retries unboundedly unless maxAttempts is set, in which case it applies to both.

HttpException carries $statusCode, $errorCode and $retryAfterSeconds. Branch on the constants in Queen\Exceptions\ErrorCode rather than on message text:

use Queen\Exceptions\ErrorCode;
use Queen\Exceptions\HttpException;

try {
    $client->queue('orders')->push($items)->execute();
} catch (HttpException $e) {
    if ($e->errorCode === ErrorCode::CLUSTER_SUSPENDED) {
        // terminal: operator action required, never retry
    }
}

429 maps to rate_limited or quota_exceeded and is retryable. 403 maps to cluster_suspended, storage_quota_exceeded, feature_gated or forbidden and is terminal. $errorCode stays null when the response body carries no code.

Laravel

The package declares the provider and the facade alias in composer.json under extra.laravel, so Laravel auto-discovery wires them without any manual registration:

• Queen\Laravel\QueenServiceProvider binds Queen\Queen as a singleton and aliases it as queen.
• Queen\Laravel\QueenFacade is aliased as Queen.

Publish the config to override it:

php artisan vendor:publish --tag=queen-config

Configuration

config/queen.php reads these environment variables:

Unset retry_429 variables arrive as null and are filtered out by the provider, so the per-request-kind defaults apply. Note that the Laravel defaults for the affinity ring (150) and the unhealthy-backend cooldown (30000 ms) differ from the library’s own defaults (128 and 5000 ms).

Injecting the client

use Queen\Queen;

class OrderProducer
{
    public function __construct(private Queen $queen) {}

    public function emit(array $order): void
    {
        $this->queen->queue('orders')
            ->partition("acct-{$order['account']}")
            ->push([['transactionId' => "order-{$order['id']}", 'data' => $order]])
            ->execute();
    }
}

Or through the facade: Queen::queue('orders')->push(...)->execute();.

The artisan consume command

php artisan queen:consume orders "App\\Queen\\OrderHandler" --group=processors --batch=10 --auto-ack

The handler argument is a fully qualified class name resolved out of the container; it must have a handle() method. The command builds a HighLevelConsumer, subscribes, and loops until a signal closes it.

handle() receives one message array when --batch is 1, and an array of messages when it is greater than 1.