Download the PHP package snipershady/queue-bridge without Composer

queue-bridge

English | Italiano

Backend-agnostic message queue for PHP 8.3+: one typed API for Redis Streams and RabbitMQ, with retries, a dead letter queue, delayed messages and automatic Redis version detection. An InMemory backend with the same semantics serves tests and local development.

Switching to RabbitMQ changes one line: StrategyDtoRabbitMq::fromConnection($connection) instead of StrategyDtoRedis::fromPhpRedis($redis).

• At-least-once delivery: no message is lost when a worker crashes (exception, segfault, OOM, kill -9); at worst it is delivered twice.
• Arbitrary payloads: the text can be JSON, structured data or plain text; it is decoded safely (JSON when valid, otherwise the raw string) and never throws.
• Redis 6.0 → 8.4: the server version is detected and the best commands are used (XACKDEL, XTRIM ... ACKED, XREADGROUP ... CLAIM, DELEX) without changing the public API.
• RabbitMQ 3.13 / 4.x: quorum queues, publisher confirms, x-delivery-limit with at-least-once dead lettering.
• Worker: handlers per channel, middleware, retry strategies, poison message protection (maxDeliveries), message expiration, graceful shutdown, events for metrics, several queues in one process, and a CLI.

Contents

• Installation
• Redis, RabbitMQ or InMemory?
• DSN
• Producing messages
• Consuming with the Worker
• Consuming without the Worker
• CLI
• Examples
• Configuration
• Recommended server configuration
• Delivery guarantees and limits
• Redis version support
• Benchmark
• Extending
• Development

The full design documentation (queue concepts, UML diagrams, concurrency, worker scheduling) is in DOC.md.

Installation

Package page: packagist.org/packages/snipershady/queue-bridge · Source: github.com/snipershady/queue-bridge · Releases: CHANGELOG.md

Then install the client of the backend you use:

Requirements: PHP 8.3, 8.4 or 8.5; Redis >= 6.0 (Valkey and managed services reporting redis_version 7.2 work as Redis 7.0); RabbitMQ 3.13 or 4.x.

Redis, RabbitMQ or InMemory?

All three backends implement the same contract and pass the same contract tests: the application code does not change. They differ in how the guarantees are obtained and in what they cost to run.

Integration

QueueBridge::factory() returns a factory that receives one typed DTO per backend and a QueueConfig. Everything else (producer, worker, handlers, middleware) is the same on every backend.

Redis

The Redis version is detected from INFO server. If the ACL forbids INFO, pass it: StrategyDtoRedis::fromPhpRedis($redis, RedisVersionEnum::GE70); an override higher than the server throws VersionMismatchException in create().

RabbitMQ

StrategyDtoRabbitMq::fromConnection($connection) and fromPhpAmqpLib($channel) reuse an existing connection. Publisher confirms are always on: retry, reject and requeue acknowledge the original only after the broker has confirmed the copy, so they cannot be turned off. When you build the connection yourself, give it a channel_rpc_timeout (AMQPConnectionConfig::setChannelRPCTimeout()): without one, a network that silently drops packets makes stats() or deadLetters() wait forever (the DSN sets 10 s).

InMemory

From a DSN

Handy for configuration files and environment variables:

Unknown options are an error, so a typo (?verison=8.0) is not silently ignored; memory:// takes no host or option. A Redis user needs a password (an ACL user with nopass accepts any). Credentials must be URL-encoded. Boolean options accept 1/0, true/false, yes/no, on/off. The AMQP vhost defaults to /. Example: redis://app:secret@cache:6379/2?consumer=web-1&timeout=1.5.

Producing messages

• text is any valid UTF-8 string up to maxTextBytes (default 1 MiB): JSON, structured data or plain text.
• Channel names match ^[A-Za-z0-9._:-]+$ (max 255 characters).
• publishBatch() uses one pipeline on Redis and batched confirms on RabbitMQ. A partial failure throws BatchSendException with the sent and the failed IDs. The messages are all read and encoded before anything is sent: an error while iterating the input sends nothing. With Predis the outcome of each command is reported whatever the client's exceptions option.
• Producer middleware: CorrelationIdMiddleware adds correlation_id/causation_id from the message being handled, TraceContextMiddleware adds a W3C traceparent. Pass them to QueueBridge::factory(producerMiddlewares: [...]).

Consuming with the Worker

The handler receives the Message (id, channel, raw text, attempts, headers, timestamps) and a DecodedText: $text->isJson(), $text->value (decoded JSON with objects as arrays, or the raw string) and $text->raw. You can also ignore it and decode $message->text yourself.

• Graceful shutdown: on SIGTERM/SIGINT (with ext-pcntl) or stop() (declared in WorkerInterface) the worker finishes the current message, gives back the ones already received and returns WorkerExitReason::StopRequested. stopWhenEmpty: true exits at the first empty receive (batch jobs, cron).
• Middleware: LoggingMiddleware, TimeoutMiddleware (needs ext-pcntl; a timeout is retried), IdempotencyMiddleware (two-phase lock on the message ID with InMemoryIdempotencyStore, RedisIdempotencyStore::create() or your own IdempotencyStoreInterface, on any backend; a duplicate still being processed by another worker is retried when that worker's lock expires, never acknowledged), CorrelationContextMiddleware.
• Events: MessageReceived, MessageHandled, MessageFailed, MessageRetried, MessageDeadLettered, MessageDiscarded, MessageReleased, WorkerStopped.
• Several queues in one process: MultiQueueWorker runs one round of each worker in turn, so a slow queue does not block the others; give the inner workers a short receiveTimeoutMs (e.g. 100 ms).

Dead letters, statistics and health checks:

Listing and purging the queue itself:

• browse() never consumes: IDs and attempts do not change. Order: in flight, ready, delayed by due time. It is a snapshot: with running workers messages can change state while you read.
• purge() removes the ready and delayed messages present when it is called. Messages being processed are not touched (they are acknowledged or come back as usual), the dead letter queue has its own purgeDeadLetters(), and the queue stays usable.
• RabbitMQ: AMQP cannot read a queue without consuming it, so browse() takes the ready messages with basic.get and gives them back with nack + requeue. Messages in the wait queues and those held by other consumers are not visible; on RabbitMQ 3.13 every browse increments x-delivery-count (frequent browsing pushes messages towards x-delivery-limit and the DLQ), and messages read by a browse() that does not reach the end of the queue go back to its tail. Use it for occasional inspection, not for polling.

Consuming without the Worker

To embed the queue in an existing daemon or event loop, use the transport directly and decide yourself what to do with each message:

Without the Worker you take over what it does for you: maxAttempts ($delivery->message->attempts), maxDeliveries ($delivery->deliveryCount), expiration ($delivery->message->expiresAt), signals and limits. $transport->release($delivery) gives a message back without counting a failure.

CLI

The DSN can also come from the QUEUE_DSN environment variable; --strategy=redis|rabbitmq|in-memory is an optional check against the DSN scheme. -v/-vv print info/debug logs on stderr. An option that the command does not accept is a usage error (work --max-message=5 does not run with the default limit). work validates its options and the DSN before loading the bootstrap file; a bootstrap that throws is a usage error too. Exit codes: 0 success, 1 backend error, 2 usage error.

Run workers under a supervisor with --max-runtime and --memory-limit: the process exits cleanly and is restarted. Ready-made configurations: systemd and Supervisor.

Examples

The examples/ directory (left out of the Composer package) contains runnable scripts. Services and ports are those of docker-compose.yml (docker compose up -d --wait); REDIS_HOST, REDIS_PORT, AMQP_HOST, AMQP_PORT, AMQP_USER, AMQP_PASSWORD point them elsewhere.

Sample payloads are in examples/deploy/.

Configuration

QueueConfig is the same on every backend:

Redis layout (hash tags keep the keys of a queue in one cluster slot): rq:{orders}:stream, rq:{orders}:delayed, rq:{orders}:dlq, rq:{orders}:idem:<id>; consumer group queue-bridge.

RabbitMQ topology created by setup(): topic exchange rq.orders (routing key = channel), quorum queue rq.orders with x-delivery-limit, wait queues rq.orders.retry.<ms>, classic DLQ rq.orders.dlq with x-max-length = dlqMaxLength (overflow drop-head: the oldest dead letters are dropped, a reject never fails because the DLQ is full).

Recommended server configuration

Redis

Without AOF a restart empties the queue; with an eviction policy Redis may delete queue keys under memory pressure. Use ACL users limited to the queue keys (~rq:*); if the ACL forbids INFO, pass the version explicitly (otherwise the library falls back to the 6.0 baseline and logs a warning).

RabbitMQ

• The library declares quorum queues: run at least three nodes in production.
• consumer_timeout (default 30 minutes) closes the channel when a message stays unacknowledged longer: keep handlers well below it.
• Heartbeats: a synchronous PHP worker does not answer heartbeats while a handler runs, so long handlers make the broker close the connection. Set a heartbeat larger than the longest handler, disable it (heartbeat=0, the DSN default) or use php-amqplib's PCNTLHeartbeatSender (ext-pcntl).

Delivery guarantees and limits

• At-least-once: a message is acknowledged only after the handler returns. A crash between the handler and the ack, a network failure or an expired lock produce a duplicate: handlers must be idempotent (or use IdempotencyMiddleware). Retries and DLQ moves are atomic on Redis (Lua) and publish-confirm-then-ack on RabbitMQ: never lost, at worst duplicated.
• No ordering guarantee: retries, delays, released and recovered messages change the order (a released or recovered message goes behind the others, on InMemory as on Redis). Messages arrive in publishing order only in the absence of these events. Delayed messages become available in order of due time; on RabbitMQ only across different tiers.
• No fan-out: one QueueConfig is one queue; each message is handled by one worker. For publish/subscribe, publish to several queues.
• Scaling with processes: the worker is single-process and never forks. Run more processes, on more machines, on the same queue; two workers never handle the same message at the same time.
• claimIdleMs and handler duration (Redis): a message unacknowledged for claimIdleMs is considered orphaned and given to another worker. claimIdleMs must be larger than the longest handler (enforce it with TimeoutMiddleware), otherwise a slow message is processed twice in parallel. The whole batch counts as in flight from the moment it is received: the worker starts a message only if it was received less than claimIdleMs - handler timeout ago (claimIdleMs / 2 without TimeoutMiddleware) and releases the rest of the batch with a warning in the log. If the warning shows up, lower batchSize.
• RabbitMQ: orphans come back as soon as the connection closes; the limits are the heartbeat and consumer_timeout described above. x-delivery-limit equals maxDeliveries, so a message that keeps crashing the worker is dead-lettered by the broker as a last resort. On RabbitMQ 3.13 release() also counts as a delivery. Statistics are a snapshot: right after a release() the message may be in transit and missing from ready for a moment.
• Delay precision on RabbitMQ: delays use wait queues with fixed TTL tiers (default 1 s, 5 s, 30 s, 2 min, 10 min, 1 h). The largest tier not above the remaining delay is used and a message that comes back early waits again: precision is about one tier (1500 ms is delivered between 1.5 and 3.5 s). Beyond the largest tier the message cycles through it until due. Choose retry delays that match the tiers. On Redis delays are exact within the receive timeout; on InMemory they follow the clock.
• Wire format: messages are stored in a versioned format; a rolling deploy with old and new workers loses nothing. Unknown formats go to the DLQ and never crash the worker. See UPGRADE.md.

Redis version support

The level is detected once from INFO server; each capability is a strategy chosen at creation, never a version check in the hot path.

The main stream is never trimmed with MAXLEN (it would delete unread messages): acknowledged entries are deleted, and the trim is only a safety net that removes acknowledged entries. The CI runs the integration tests on Redis 6.0, 6.2, 7.0, 7.4, 8.0, 8.2 and 8.4 with ext-redis and Predis, and on RabbitMQ 3.13 and 4.x, with PHP 8.3, 8.4 and 8.5.

Benchmark

Indicative results: PHP 8.3 CLI with OPcache, Xdebug off, AMD Ryzen 9 9900X; servers in Docker on the same machine, reached through the published ports (Redis 8.4, RabbitMQ 4.x single node); 157-byte JSON payload, batches of 100, one process. "consume" includes receive, a no-op handler and the ack.

Redis levels forced on the same 8.4 server (ext-redis):

A single publish() is bound by the network round trip (one per message, plus the publisher confirm on RabbitMQ; here also by Docker's port forwarding): use publishBatch() for throughput. Consumption scales with the number of worker processes. The levels differ in the atomicity and round trips of maintenance operations (orphan recovery, trim), not in the hot path, so their numbers are within run-to-run noise.

Extending

A new backend (for example SQS) implements StrategyDtoInterface, TransportInterface (plus the optional capability interfaces it supports: setup, maintenance, stats, dead letters, batch ack, purge, browse) and TransportBuilderInterface, and is registered in the factory:

The core does not change. EnvelopeCodec converts messages to the versioned wire format, and the abstract TransportContractTestCase in the repository (tests/Contract) verifies the same semantics as the provided backends.

A custom client (company wrapper, framework facade) implements RedisClientInterface or AmqpConnectorInterface and is passed to the DTO constructor.

Development

See UPGRADE.md.

License

GPL-2.0-only.