PHP client

The Composer package is smartpricing/queen-mq, PSR-4 namespace Queen\. It requires PHP 8.3 or newer plus Guzzle 7 and ramsey/uuid. The client is synchronous: there is no promise or async variant. Laravel integration (service provider, facade, artisan command) ships in the same package and is optional. Version 1.0.0, aligned with the broker.

use Queen\Queen;

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

Constructor

new Queen(string|array $config = [])

A string is a single URL. An array with a numeric key 0 is a list of URLs. Anything else is a config array merged over Defaults::CLIENT_DEFAULTS; without url or urls it throws InvalidArgumentException('Must provide urls or url in configuration').

The load-balancing keys are inert with a single URL: the LoadBalancer is only constructed when urls holds more than one entry.

retry429

Separate from retryAttempts, which covers 5xx and network faults. Constants live on Queen\Http\Retry429Policy.

A Retry-After header (seconds) wins over the computed delay. A 429 is retried against the same backend and never triggers failover.

Queen

ack

$message is a message array, a transaction-id string, or an array of messages. $status is true → completed, false → failed, or a status string passed through verbatim. $context accepts group and error.

partitionId is mandatory on every message array; without it the builder throws InvalidArgumentException. A leaseId is forwarded when present. Per-message statuses inside a batch work by tagging items with _status and _error.

One message posts to /api/v1/ack, more than one to /api/v1/ack/batch. Both respond with one result per ack in request order.

renew

Accepts a lease-id string, a message array, or an array of either. Posts to /api/v1/lease/{leaseId}/extend once per distinct lease id.

QueueBuilder

$queen->queue($name). Configuration methods return static; terminal methods are push(), pop(), consume(), getConsumer(), create(), delete(), dlq() and flushBuffer().

Addressing

Queue lifecycle

OperationBuilder exposes onSuccess(Closure), onError(Closure) and execute(): mixed.

Defaults::QUEUE_DEFAULTS:

Extra keys you pass to config() are sent verbatim, so any option /configure accepts is reachable. /configure is a full replace: keys absent from the body reset to the broker’s defaults, so send the complete intended configuration every time. The option list is in Reference.

Producing

Each item may carry data, payload, or be the payload itself. The client mints a UUIDv7 transactionId via Queen\Support\Uuid::v7() when the item has none.

PushBuilder exposes onSuccess(Closure), onError(Closure), onDuplicate(Closure) and execute(): mixed. execute() returns the broker’s per-item array for a direct push; a buffered push returns immediately without sending.

$queen->queue('orders')
    ->partition('customer-42')
    ->push([['data' => ['orderId' => 8891]]])
    ->execute();

Consuming

pop(): array

Returns a list of message arrays, [] when there is nothing to return. Route selection: queue plus non-Default partition uses /api/v1/pop/queue/:queue/partition/:partition; queue alone uses /api/v1/pop/queue/:queue; namespace or task alone uses /api/v1/pop.

consume(Closure $handler): ConsumeBuilder

ConsumeBuilder exposes onSuccess(Closure), onError(Closure) and execute(): void.

getConsumer(): HighLevelConsumer

A pull-loop consumer modelled on php-rdkafka’s KafkaConsumer, for code that wants to own its own loop instead of handing over a closure.

consume() and consumeBatch() call pcntl_signal_dispatch() when the extension is available, so a SIGTERM handler you installed runs between polls. Long-poll timeouts and connection refusals are absorbed into null / []; every other exception propagates.

$message['trace']

Messages returned by HighLevelConsumer carry a trace closure:

$message['trace'](['traceName' => 'tenant-acme', 'eventType' => 'info', 'data' => ['step' => 1]]);

data is required; traceName accepts a string or an array of strings; eventType defaults to 'info'. It never throws: failures return ['success' => false, 'error' => ...].

Buffering

flushBuffer(): void flushes this builder’s queue/partition buffer; throws RuntimeException with no queue name. Buffers are keyed on "<queue>/<partition>", so builders addressing the same pair share one. A buffer flushes at messageCount messages or timeMillis after its first message.

Dead-letter queue

dlq(?string $consumerGroup = null): DLQBuilder. Throws RuntimeException with no queue name.

Read-only. Replay lives on the admin façade: retryMessage(string $partitionId, string $transactionId).

TransactionBuilder

$queen->transaction(). Pushes and acks in one PostgreSQL transaction, all-or-nothing.

$queen->transaction()
    ->ack($message)
    ->queue('orders.enriched')->push([['data' => ['id' => 1]]])
    ->commit();

Admin

$queen->admin(). Every method returns the broker’s decoded body. Methods taking array $params turn it into a query string.

Resources

Messages and traces

Status and analytics

Consumer groups

seekConsumerGroup posts $options verbatim: ['toEnd' => true] or ['timestamp' => '<RFC3339>'].

System

Errors

Non-2xx responses throw Queen\Exceptions\HttpException, with isClusterSuspended() and isRateLimited() helpers. The machine-readable codes are constants on Queen\Exceptions\ErrorCode:

A 204 response yields null, because the broker sends no body at all on 204.

Laravel integration

The package auto-registers Queen\Laravel\QueenServiceProvider and the Queen facade alias. Publish the config with the queen-config tag; it lands at config/queen.php. The provider binds Queen\Queen as a singleton and aliases it to queen.

Config keys and env vars

Facade

use Queen\Laravel\QueenFacade as Queen;

Queen::queue('orders')->push([['data' => ['id' => 1]]])->execute();

The facade proxies queue(), transaction(), admin(), ack(), renew(), flushAllBuffers(), getBufferStats() and close().

Artisan command

Registered only when running in console.

php artisan queen:consume orders "App\\Handlers\\OrderHandler"

The command errors out when the handler class does not exist.

Not in this client

There is no streaming SDK for PHP. The Stream builder, windows, gates and the /streams/v1/* runtime exist in the JavaScript, Python and Go clients only.

Same surface, other languages