PHPackages                             brainv/bananacue - PHPackages - PHPackages  [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register)

1. [Directory](/)
2. /
3. [Queues &amp; Workers](/categories/queues)
4. /
5. brainv/bananacue

ActiveLibrary[Queues &amp; Workers](/categories/queues)

brainv/bananacue
================

BananaCue — Universal PHP queue abstraction layer (SQS, RabbitMQ, Kafka, Redis, ActiveMQ, Pulsar, Database, In-Memory)

v1.0.1(2w ago)03↓100%MITPHPPHP &gt;=8.1

Since Mar 8Pushed 2w agoCompare

[ Source](https://github.com/brainv/bananacue)[ Packagist](https://packagist.org/packages/brainv/bananacue)[ RSS](/packages/brainv-bananacue/feed)WikiDiscussions main Synced today

READMEChangelogDependencies (1)Versions (3)Used By (0)

BananaCue — PHP Universal Queue Abstraction Layer
=================================================

[](#bananacue--php-universal-queue-abstraction-layer)

A single, consistent interface for the most popular message brokers and queuing backends — AWS SQS, RabbitMQ, Apache Kafka, Redis, ActiveMQ, Apache Pulsar, relational databases (via PDO), and an in-memory driver for tests.

---

File structure
--------------

[](#file-structure)

```
queue/
├── Contracts/
│   ├── QueueInterface.php   # The universal interface
│   └── Message.php          # Immutable message value-object
├── Drivers/
│   ├── AbstractDriver.php   # Shared serialization + default subscribe() loop
│   ├── SqsDriver.php        # AWS SQS
│   ├── RabbitMqDriver.php   # RabbitMQ (AMQP 0-9-1)
│   ├── KafkaDriver.php      # Apache Kafka (ext-rdkafka)
│   ├── RedisDriver.php      # Redis LIST + ZSET scheduler
│   ├── ActiveMqDriver.php   # ActiveMQ / any STOMP broker
│   ├── PulsarDriver.php     # Apache Pulsar (HTTP REST API)
│   ├── DatabaseDriver.php   # MySQL / PostgreSQL / SQLite (PDO)
│   └── InMemoryDriver.php   # In-process (testing)
├── Exceptions/
│   └── QueueException.php
├── QueueManager.php         # Factory + multi-connection manager
└── composer.json

```

---

Installation
------------

[](#installation)

```
composer require brainv/bananacue
```

Install the package(s) for your chosen driver:

DriverPackageSQS`aws/aws-sdk-php`RabbitMQ`php-amqplib/php-amqplib`Kafka`ext-rdkafka` (OS + PHP extension)Redis`ext-redis` OR `predis/predis`ActiveMQ`stomp-php/stomp-php`Pulsar`guzzlehttp/guzzle`Database`ext-pdo` (built into PHP)In-Memory*(none)*---

Quick Start
-----------

[](#quick-start)

```
use Queue\QueueManager;

$manager = new QueueManager([
    'default' => 'redis',
    'connections' => [
        'redis' => [
            'driver'   => 'redis',
            'host'     => 'localhost',
            'port'     => 6379,
        ],
        'sqs' => [
            'driver'    => 'sqs',
            'region'    => 'us-east-1',
            'queue_url' => 'https://sqs.us-east-1.amazonaws.com/123456789/my-queue',
            // credentials omitted → uses IAM role / env vars
        ],
    ],
]);

// Publish to the default (redis) connection
$manager->publish('orders', ['order_id' => 42, 'total' => 99.99]);

// Publish to a specific connection
$manager->connection('sqs')->publish('notifications', ['user_id' => 7]);

// Consume a single message
$msg = $manager->consume('orders');
if ($msg) {
    echo $msg->payload['order_id']; // 42
    $manager->ack($msg);
}

// Stream messages with a callback
$manager->subscribe('orders', function (\Queue\Contracts\Message $msg) use ($manager): bool {
    processOrder($msg->payload);
    $manager->ack($msg);
    return true; // return false to stop the loop
});

// Queue stats
echo $manager->size('orders'); // e.g. 5

// Purge a queue
$manager->purge('orders');

// Close all connections
$manager->disconnect();
```

---

Driver Configuration Reference
------------------------------

[](#driver-configuration-reference)

### AWS SQS

[](#aws-sqs)

```
'sqs' => [
    'driver'      => 'sqs',
    'region'      => 'us-east-1',
    'version'     => 'latest',
    'queue_url'   => 'https://sqs.us-east-1.amazonaws.com/ACCOUNT/QUEUE',
    'credentials' => ['key' => 'AKID…', 'secret' => '…'], // optional
    'endpoint'    => 'http://localhost:4566',               // optional: LocalStack
]
```

Publish options:

- `delay` (int) – Seconds before the message is visible (0–900)
- `group_id` (string) – Required for FIFO queues
- `attributes` (array) – SQS MessageAttributes

---

### RabbitMQ

[](#rabbitmq)

```
'rabbit' => [
    'driver'        => 'rabbitmq',
    'host'          => 'localhost',
    'port'          => 5672,
    'user'          => 'guest',
    'password'      => 'guest',
    'vhost'         => '/',
    'exchange'      => '',          // '' = default direct exchange
    'exchange_type' => 'direct',
    'durable'       => true,
]
```

Publish options: `priority`, `expiration` (seconds), `headers`, `routing_key`

Queue declare options: `max_priority`, `dlx` (dead-letter exchange), `ttl`

---

### Apache Kafka

[](#apache-kafka)

```
'kafka' => [
    'driver'             => 'kafka',
    'brokers'            => 'kafka1:9092,kafka2:9092',
    'group_id'           => 'my-consumer-group',
    'auto_offset_reset'  => 'earliest',  // or 'latest'
    'timeout_ms'         => 1000,
    'producer_config'    => [],          // extra librdkafka settings
    'consumer_config'    => [],
]
```

Publish options: `partition`, `key`, `headers`

---

### Redis

[](#redis)

```
'redis' => [
    'driver'        => 'redis',
    'host'          => 'localhost',
    'port'          => 6379,
    'password'      => null,
    'database'      => 0,
    'prefix'        => 'queue:',
    'block_timeout' => 1,     // BLPOP wait in seconds
]
```

Publish options: `delay` (seconds), `id`

---

### ActiveMQ (STOMP)

[](#activemq-stomp)

```
'activemq' => [
    'driver'           => 'activemq',
    'broker_uri'       => 'tcp://localhost:61613',
    'login'            => '',
    'passcode'         => '',
    'vhost'            => '/',
    'read_timeout'     => 0.5,
    'destination_type' => 'queue',  // 'queue' or 'topic'
]
```

Publish options: `persistent`, `priority`, `expires` (seconds), `headers`

---

### Apache Pulsar

[](#apache-pulsar)

```
'pulsar' => [
    'driver'       => 'pulsar',
    'base_url'     => 'http://localhost:8080',
    'tenant'       => 'public',
    'namespace'    => 'default',
    'subscription' => 'php-sub',
    'sub_type'     => 'Shared',   // Exclusive|Shared|Failover|Key_Shared
    'token'        => null,       // JWT bearer token
]
```

Publish options: `delay` (ms), `key`, `properties`

---

### Database (PDO)

[](#database-pdo)

```
'db' => [
    'driver'     => 'database',
    'dsn'        => 'mysql:host=localhost;dbname=myapp',
    'username'   => 'root',
    'password'   => 'secret',
    'table'      => 'queue_jobs',
    'reserve_for'=> 60,   // seconds before unreserved job is retried
]
```

Required DDL:

```
CREATE TABLE queue_jobs (
    id           BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    queue        VARCHAR(255) NOT NULL,
    payload      LONGTEXT     NOT NULL,
    attempts     TINYINT      NOT NULL DEFAULT 0,
    reserved_at  INT UNSIGNED NULL,
    available_at INT UNSIGNED NOT NULL,
    created_at   INT UNSIGNED NOT NULL,
    INDEX idx_queue_available (queue, available_at, reserved_at)
);
```

Publish options: `delay` (seconds)

---

### In-Memory (tests)

[](#in-memory-tests)

```
'test' => [
    'driver' => 'memory',
]
```

---

Registering a Custom Driver
---------------------------

[](#registering-a-custom-driver)

```
$manager->extend('ironmq', function (array $config): \Queue\Contracts\QueueInterface {
    return new IronMqDriver($config);
});
```

---

Message Object
--------------

[](#message-object)

```
$msg->payload   // mixed  – deserialized message body
$msg->id        // ?string – broker-assigned message ID
$msg->queue     // string  – queue / topic name
$msg->attempts  // int     – delivery attempts (0 = first)
$msg->meta      // array   – driver-specific metadata
$msg->handle    // mixed   – opaque ack handle (internal use)
```

---

Error Handling
--------------

[](#error-handling)

All drivers throw `Queue\Exceptions\QueueException` (extends `\RuntimeException`) on unrecoverable errors.

```
use Queue\Exceptions\QueueException;

try {
    $msg = $manager->consume('orders');
} catch (QueueException $e) {
    logger()->error('Queue error', ['message' => $e->getMessage()]);
}
```

###  Health Score

39

—

LowBetter than 87% of packages

Maintenance97

Actively maintained with recent releases

Popularity4

Limited adoption so far

Community6

Small or concentrated contributor base

Maturity43

Maturing project, gaining track record

 Bus Factor1

Top contributor holds 100% of commits — single point of failure

How is this calculated?**Maintenance (25%)** — Last commit recency, latest release date, and issue-to-star ratio. Uses a 2-year decay window.

**Popularity (30%)** — Total and monthly downloads, GitHub stars, and forks. Logarithmic scaling prevents top-heavy scores.

**Community (15%)** — Contributors, dependents, forks, watchers, and maintainers. Measures real ecosystem engagement.

**Maturity (30%)** — Project age, version count, PHP version support, and release stability.

###  Release Activity

Cadence

Every ~0 days

Total

2

Last Release

17d ago

### Community

Maintainers

![](https://www.gravatar.com/avatar/20c8423a9617943b3a79ca898ed8fec3b72ff97859427d78a4e3ece1b21fa0e1?d=identicon)[brainv](/maintainers/brainv)

---

Top Contributors

[![brainv](https://avatars.githubusercontent.com/u/3684?v=4)](https://github.com/brainv "brainv (2 commits)")

###  Code Quality

TestsPHPUnit

### Embed Badge

![Health badge](/badges/brainv-bananacue/health.svg)

```
[![Health](https://phpackages.com/badges/brainv-bananacue/health.svg)](https://phpackages.com/packages/brainv-bananacue)
```

###  Alternatives

[react/socket

Async, streaming plaintext TCP/IP and secure TLS socket server and client connections for ReactPHP

1.3k111.4M369](/packages/react-socket)[amphp/byte-stream

A stream abstraction to make working with non-blocking I/O simple.

391113.4M92](/packages/amphp-byte-stream)[league/geotools

Geo-related tools PHP 7.3+ library

1.4k5.2M23](/packages/league-geotools)[queue-interop/queue-interop

Promoting the interoperability of MQs objects. Based on Java JMS

47930.0M85](/packages/queue-interop-queue-interop)[amphp/sync

Non-blocking synchronization primitives for PHP based on Amp and Revolt.

18850.5M36](/packages/amphp-sync)[amphp/dns

Async DNS resolution for Amp.

19237.1M37](/packages/amphp-dns)

PHPackages © 2026

[Directory](/)[Categories](/categories)[Trending](/trending)[Changelog](/changelog)[Analyze](/analyze)
