Messaging

Multi-transport messaging with Redis, Kafka, RabbitMQ, Database and InMemory: one API for all.

Introduction

Messaging in PHP applications often means: committing to a single broker and wiring its client library deep into the application. Kafka code looks completely different from RabbitMQ code, and switching the transport means a rewrite. For tests you then need a running broker or mock constructs.

jardisadapter/messaging solves this with a unified publish/consume API that works identically across five transports:

Redis — Pub/Sub for fire-and-forget, Streams with Consumer Groups for persistent processing
Kafka — Producer/Consumer with Consumer Groups, SASL authentication
RabbitMQ — AMQP Exchange/Queue with ACK/NACK and routing
Database — Transactional Outbox Pattern via PDO, point-to-point and fan-out
InMemory — synchronous transport for tests without a broker

Additional features:

Priority failover — multiple transports with automatic fallback
Automatic serialization — string, object or array as payload, JSON encoding/decoding transparent
Lazy connection — connection only established on first publish() or consume()
External connections — integrate existing Redis, PDO, AMQP or Kafka clients
Graceful shutdown — SIGTERM/SIGINT handlers installed automatically

Installation

bash

composer require jardisadapter/messaging

GitHub: jardisAdapter/messaging

Optional PHP extensions:

Extension	For
`ext-redis`	Redis transport
`ext-amqp`	RabbitMQ transport
`ext-rdkafka`	Kafka transport
`ext-pdo`	Database transport

Basic Usage

Publish and Consume

php

use JardisAdapter\Messaging\Factory\ConnectionFactory;
use JardisAdapter\Messaging\Factory\PublisherFactory;
use JardisAdapter\Messaging\Factory\ConsumerFactory;
use JardisAdapter\Messaging\MessagePublisher;
use JardisAdapter\Messaging\MessageConsumer;
use JardisAdapter\Messaging\Handler\CallbackHandler;

$connFactory = new ConnectionFactory();
$pubFactory  = new PublisherFactory();
$conFactory  = new ConsumerFactory();

// Redis connection
$conn = $connFactory->redis('localhost', 6379);

// Create publisher and consumer
$publisher = new MessagePublisher($pubFactory->redis($conn));
$consumer  = new MessageConsumer($conFactory->redis($conn));

// Publish message (array → automatically JSON-encoded)
$publisher->publish('orders', [
    'order_id' => 42,
    'total' => 299.99,
    'customer_id' => 7,
]);

// Consume messages
$consumer->consume('orders', new CallbackHandler(
    function (string|array $message, array $metadata): bool {
        // $message is already decoded: ['order_id' => 42, ...]
        processOrder($message);
        return true;  // ACK
    }
));

Payload Types

php

// String — sent directly
$publisher->publish('logs', 'Simple log message');

// Array — JSON-encoded
$publisher->publish('events', ['type' => 'OrderCreated', 'data' => [...]]);

// Object — JSON-encoded (public properties or JsonSerializable)
$publisher->publish('events', $domainEvent);

Transports

Redis Pub/Sub

Fire-and-forget fanout. Subscribers must be connected at the time of publishing:

php

$conn = $connFactory->redis('localhost', 6379, password: 'secret');

$publisher = new MessagePublisher($pubFactory->redis($conn));
$consumer  = new MessageConsumer($conFactory->redis($conn));

Redis Streams

Persistent messages with optional Consumer Groups:

php

$publisher = new MessagePublisher($pubFactory->redis($conn, useStreams: true));
$consumer  = new MessageConsumer($conFactory->redis($conn, useStreams: true));

// Simple reading
$consumer->consume('orders', $handler);

// With Consumer Group — each message is processed by only one consumer in the group
$consumer->consume('orders', $handler, [
    'group'    => 'order-service',
    'consumer' => 'worker-1',
]);

Kafka

php

$producer = $connFactory->kafka('broker1:9092,broker2:9092',
    username: 'app', password: 'secret');

$consumer = $connFactory->kafkaConsumer('broker1:9092,broker2:9092',
    groupId: 'order-service',
    username: 'app', password: 'secret');

$publisher = new MessagePublisher($pubFactory->kafka($producer));
$msgConsumer = new MessageConsumer($conFactory->kafka($consumer));

RabbitMQ

php

$conn = $connFactory->rabbitMq('localhost', 5672, 'guest', 'guest', [
    'exchange_name' => 'app.events',
]);

$publisher = new MessagePublisher($pubFactory->rabbitMq($conn));
$consumer  = new MessageConsumer($conFactory->rabbitMq($conn, 'order-queue'));

Database (Transactional Outbox)

No broker needed, uses the application's own database as a message store:

php

use JardisAdapter\Messaging\Config\DatabaseTransportOptions;

$conn = $connFactory->database('mysql:host=localhost;dbname=app', 'user', 'pass');

$options = new DatabaseTransportOptions(
    table: 'domain_events',
    pollingIntervalMs: 1000,
    batchSize: 10,
    maxAttempts: 3,
    deleteAfterProcessing: false,
);

$publisher = new MessagePublisher($pubFactory->database($conn, $options));
$consumer  = new MessageConsumer($conFactory->database($conn, $options));

Fan-Out (Consumer Groups)

Multiple independent consumer groups process the same events:

php

// Email service
$consumer->consume('InvoiceCreated', $emailHandler, ['group' => 'email-service']);

// PDF service
$consumer->consume('InvoiceCreated', $pdfHandler, ['group' => 'pdf-service']);

Each group processes every event independently. Events without a group option are consumed in point-to-point mode.

InMemory (Tests)

Synchronous transport without a broker, ideal for unit tests:

php

use JardisAdapter\Messaging\Transport\InMemoryTransport;
use JardisAdapter\Messaging\Publisher\InMemoryPublisher;
use JardisAdapter\Messaging\Consumer\InMemoryConsumer;

$transport = new InMemoryTransport();

$publisher = new MessagePublisher(new InMemoryPublisher($transport));
$consumer  = new MessageConsumer(new InMemoryConsumer($transport));

$publisher->publish('orders', ['order_id' => 123]);
$consumer->consume('orders', $handler);

// Assertions
$transport->getMessageCount('orders');  // 0 (consumed)

Or via factories with a shared transport:

php

$transport = new InMemoryTransport();
$pubFactory->setSharedTransport($transport);
$conFactory->setSharedTransport($transport);

$publisher = new MessagePublisher($pubFactory->inMemory());
$consumer  = new MessageConsumer($conFactory->inMemory());

Priority Failover

Multiple transports with automatic fallback:

php

$publisher = new MessagePublisher(
    $pubFactory->redis($primaryRedis),      // First attempt
    $pubFactory->redis($secondaryRedis),    // Fallback
    $pubFactory->database($dbConn),         // Second fallback
);

// On failure of the first transport, the next is tried automatically
$publisher->publish('orders', $orderData);

The order in the constructor determines priority: the first transport has the highest priority. The fallback applies to MessagePublisher and MessageConsumer alike and triggers only on MessageException. Other exceptions (e.g. JsonException on an invalid payload) propagate immediately.

MessagingService (DI Container)

Lazy-loading orchestrator for DI containers, connection only established on first call:

php

use JardisAdapter\Messaging\MessagingService;

$messaging = new MessagingService(
    publisherFactory: fn() => new MessagePublisher(
        $pubFactory->redis($connFactory->redis('localhost'))
    ),
    consumerFactory: fn() => new MessageConsumer(
        $conFactory->redis($connFactory->redis('localhost'))
    ),
);

// Connection is established here
$messaging->publish('topic', $payload);
$messaging->consume('topic', $handler);

External Connections

Integrate existing client instances:

php

// Existing Redis instance
$conn = $connFactory->fromRedis($existingRedis, manageLifecycle: false);

// Existing PDO
$conn = $connFactory->fromPdo($existingPdo, manageLifecycle: false);

// Existing AMQP connection
$conn = $connFactory->fromAmqp($existingAmqp, exchangeName: 'app.events');

// Existing Kafka producer
$conn = $connFactory->fromKafkaProducer($existingProducer);

With manageLifecycle: false, lifecycle control remains with the external system: disconnect() is a no-op.

Message Validation

Payload validation is always active: the MessageValidator is already set in the constructor. withValidator() swaps it for a custom validator (it is not an on/off switch). Array payloads are validated; objects go straight through json_encode:

php

use JardisAdapter\Messaging\Validation\MessageValidator;

$publisher = (new MessagePublisher($pubFactory->redis($conn)))
    ->withValidator(new MessageValidator());

// Throws PublishException on arrays containing:
// - Resources
// - Closures

Handler Return Values

The handler return value controls ACK behavior:

Return	Behavior
`true`	ACK — message marked as processed
`false`	NACK — message requeued (RabbitMQ) or offset not committed (Kafka)
Exception	NACK + exception is rethrown

Error Handling

All exceptions extend MessageException:

php

use JardisSupport\Contract\Messaging\Exception\MessageException;

try {
    $publisher->publish('topic', $payload);
} catch (MessageException $e) {
    // Publish failed on all transports
}

Database Schema

For the database transport:

sql

CREATE TABLE domain_events (
    id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    topic VARCHAR(255) NOT NULL,
    payload TEXT NOT NULL,
    created_at DATETIME(6) NOT NULL,
    processed_at DATETIME(6) NULL DEFAULT NULL,
    attempts TINYINT UNSIGNED NOT NULL DEFAULT 0,
    last_error TEXT NULL DEFAULT NULL,
    INDEX idx_unprocessed (processed_at, created_at),
    INDEX idx_topic (topic, processed_at)
);

CREATE TABLE domain_event_subscriptions (
    id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    event_id BIGINT UNSIGNED NOT NULL,
    consumer_group VARCHAR(255) NOT NULL,
    processed_at DATETIME(6) NULL DEFAULT NULL,
    attempts TINYINT UNSIGNED NOT NULL DEFAULT 0,
    last_error TEXT NULL DEFAULT NULL,
    UNIQUE INDEX idx_event_group (event_id, consumer_group),
    INDEX idx_pending (consumer_group, processed_at)
);

Architecture

MessagingService                       ← Lazy-loading orchestrator
├── MessagePublisher                   ← Publisher facade (priority failover)
│   ├── RedisPublisher                 ← publish() / xAdd()
│   ├── KafkaPublisher                 ← produce() + flush()
│   ├── RabbitMqPublisher              ← AMQPExchange::publish()
│   ├── DatabasePublisher              ← INSERT INTO domain_events
│   └── InMemoryPublisher              ← InMemoryTransport
└── MessageConsumer                    ← Consumer facade (SIGTERM handler)
    ├── RedisConsumer                  ← subscribe() / xRead() / xReadGroup()
    ├── KafkaConsumer                  ← consume() poll loop
    ├── RabbitMqConsumer               ← AMQPQueue::get() + ACK/NACK
    ├── DatabaseConsumer               ← PDO polling (P2P + fan-out)
    └── InMemoryConsumer               ← Synchronous

Directory Structure

src/
├── MessagingService.php            ← Orchestrator
├── MessagePublisher.php            ← Publisher facade
├── MessageConsumer.php             ← Consumer facade
├── Config/
│   ├── ConnectionConfig.php
│   └── DatabaseTransportOptions.php
├── Connection/
│   ├── RedisConnection.php
│   ├── KafkaConnection.php
│   ├── KafkaConsumerConnection.php
│   ├── RabbitMqConnection.php
│   ├── DatabaseConnection.php
│   ├── External*.php               ← External wrappers
│   └── NullConnection.php
├── Factory/
│   ├── ConnectionFactory.php
│   ├── PublisherFactory.php
│   └── ConsumerFactory.php
├── Publisher/
│   ├── RedisPublisher.php
│   ├── KafkaPublisher.php
│   ├── RabbitMqPublisher.php
│   ├── DatabasePublisher.php
│   └── InMemoryPublisher.php
├── Consumer/
│   ├── RedisConsumer.php
│   ├── KafkaConsumer.php
│   ├── RabbitMqConsumer.php
│   ├── DatabaseConsumer.php
│   └── InMemoryConsumer.php
├── Transport/
│   └── InMemoryTransport.php
├── Handler/
│   └── CallbackHandler.php
├── Validation/
│   └── MessageValidator.php
└── Schema/
    └── domain_events.sql

API Reference

MessagingService

Method	Signature	Description
`publish`	`publish(string $topic, string\|object\|array $message, array $options = []): bool`	Publish message
`consume`	`consume(string $topic, MessageHandlerInterface $handler, array $options = []): void`	Consume messages

MessagePublisher

Method	Signature	Description
`publish`	`publish(string $topic, string\|object\|array $message, array $options = []): bool`	With failover
`withValidator`	`withValidator(MessageValidator $validator): static`	Add validator

MessageConsumer

Method	Signature	Description
`consume`	`consume(string $topic, MessageHandlerInterface $handler, array $options = []): void`	Blocking loop
`stop`	`stop(): void`	Graceful shutdown

Complete Example

Event-driven architecture with Redis Streams and database fallback:

php

use JardisAdapter\Messaging\Factory\ConnectionFactory;
use JardisAdapter\Messaging\Factory\PublisherFactory;
use JardisAdapter\Messaging\Factory\ConsumerFactory;
use JardisAdapter\Messaging\MessagePublisher;
use JardisAdapter\Messaging\MessageConsumer;
use JardisAdapter\Messaging\Handler\CallbackHandler;
use JardisAdapter\Messaging\Config\DatabaseTransportOptions;

$connFactory = new ConnectionFactory();
$pubFactory  = new PublisherFactory();
$conFactory  = new ConsumerFactory();

// Redis as primary transport, database as fallback
$redis = $connFactory->redis('redis.internal', 6379, password: $redisPassword);
$db    = $connFactory->database('mysql:host=localhost;dbname=app', 'user', 'pass');
$dbOpts = new DatabaseTransportOptions(table: 'domain_events');

$publisher = new MessagePublisher(
    $pubFactory->redis($redis, useStreams: true),
    $pubFactory->database($db, $dbOpts),
);

// Publish domain event
$publisher->publish('OrderCreated', [
    'order_id'    => $orderId,
    'customer_id' => $customerId,
    'total'       => 299.99,
    'items'       => $items,
]);

// Worker: process events with consumer group
$consumer = new MessageConsumer(
    $conFactory->redis($redis, useStreams: true),
);

$consumer->consume('OrderCreated', new CallbackHandler(
    function (string|array $message, array $metadata): bool {
        $orderId = $message['order_id'];

        // Send confirmation
        $mailer->send(buildConfirmationMail($orderId));

        // Reserve inventory
        $inventory->reserve($message['items']);

        return true;  // ACK
    }
), [
    'group'    => 'fulfillment-service',
    'consumer' => gethostname(),
]);

Core

Adapter

Support

Tools

DevOps

Messaging

Introduction

Installation

Basic Usage

Publish and Consume

Payload Types

Transports

Redis Pub/Sub

Redis Streams

Kafka

RabbitMQ

Database (Transactional Outbox)

Fan-Out (Consumer Groups)

InMemory (Tests)

Priority Failover

MessagingService (DI Container)

External Connections

Message Validation

Handler Return Values

Error Handling

Database Schema

Architecture

Directory Structure

API Reference

MessagingService

MessagePublisher

MessageConsumer

Complete Example

Messaging ​

Introduction ​

Installation ​

Basic Usage ​

Publish and Consume ​

Payload Types ​

Transports ​

Redis Pub/Sub ​

Redis Streams ​

Kafka ​

RabbitMQ ​

Database (Transactional Outbox) ​

Fan-Out (Consumer Groups) ​

InMemory (Tests) ​

Priority Failover ​

MessagingService (DI Container) ​

External Connections ​

Message Validation ​

Handler Return Values ​

Error Handling ​

Database Schema ​

Architecture ​

Directory Structure ​

API Reference ​

MessagingService ​

MessagePublisher ​

MessageConsumer ​

Complete Example ​

Messaging

Introduction

Installation

Basic Usage

Publish and Consume

Payload Types

Transports

Redis Pub/Sub

Redis Streams

Kafka

RabbitMQ

Database (Transactional Outbox)

Fan-Out (Consumer Groups)

InMemory (Tests)

Priority Failover

MessagingService (DI Container)

External Connections

Message Validation

Handler Return Values

Error Handling

Database Schema

Architecture

Directory Structure

API Reference

MessagingService

MessagePublisher

MessageConsumer

Complete Example