Stream Management

This guide covers creating, managing, and monitoring RabbitMQ Streams using the high-level Connection API and low-level protocol commands.

Overview

RabbitMQ Streams are persistent, replicated message logs. Unlike traditional queues, streams:

Store messages durably until retention policies delete them
Support multiple consumers reading from different positions
Scale horizontally with partitioning (super streams)
Provide at-least-once delivery guarantees

Creating Streams

Basic Stream Creation

Create a simple stream with just a name:

use CrazyGoat\RabbitStream\Client\Connection;

$connection = Connection::create();

// Create a basic stream
$connection->createStream('my-stream');

Stream with Arguments

Configure streams with retention policies and other options:

$connection->createStream('my-stream', [
    'max-length-bytes' => '1073741824',     // 1 GB max size
    'max-age' => '24h',                      // 24 hour retention
    'stream-max-segment-size-bytes' => '500000000',  // 500 MB segments
    'initial-cluster-size' => '3',           // 3 replicas
    'queue-leader-locator' => 'balanced',    // Balanced leader placement
]);

Available Stream Arguments

Argument	Type	Description	Example
`max-length-bytes`	string	Maximum total size of stream in bytes	`'1000000000'` (1 GB)
`max-age`	string	Maximum age of messages	`'24h'`, `'7d'`
`stream-max-segment-size-bytes`	string	Maximum size of segment files	`'500000000'` (500 MB)
`initial-cluster-size`	string	Initial number of replicas	`'3'`
`queue-leader-locator`	string	Policy for leader placement	`'client-local'`, `'balanced'`

Error Handling: STREAM_ALREADY_EXISTS

Handle the case when a stream already exists — the high-level API throws ProtocolException when the server answers with an error code:

use CrazyGoat\RabbitStream\Exception\ProtocolException;
use CrazyGoat\RabbitStream\Enum\ResponseCodeEnum;

try {
    $connection->createStream('my-stream', ['max-age' => '1h']);
    echo "Stream created successfully\n";
} catch (ProtocolException $e) {
    if ($e->getResponseCode() === ResponseCodeEnum::STREAM_ALREADY_EXISTS) {
        echo "Stream already exists - continuing\n";
    } else {
        throw $e;
    }
}

Deleting Streams

Basic Deletion

Delete a stream and all its data:

$connection->deleteStream('my-stream');

Error Handling: STREAM_NOT_EXIST

Handle the case when the stream doesn't exist:

use CrazyGoat\RabbitStream\Exception\ProtocolException;
use CrazyGoat\RabbitStream\Enum\ResponseCodeEnum;

try {
    $connection->deleteStream('my-stream');
    echo "Stream deleted successfully\n";
} catch (ProtocolException $e) {
    if ($e->getResponseCode() === ResponseCodeEnum::STREAM_NOT_EXIST) {
        echo "Stream does not exist - nothing to delete\n";
    } else {
        throw $e;
    }
}

Checking Stream Existence

Using streamExists()

The high-level API provides a simple boolean check:

if ($connection->streamExists('my-stream')) {
    echo "Stream exists\n";
} else {
    echo "Stream does not exist\n";
}

How It Works Internally

The streamExists() method queries stream metadata and checks the response code:

// Equivalent low-level implementation — StreamConnection is the raw
// protocol connection, used here directly (low-level API).
use CrazyGoat\RabbitStream\StreamConnection;
use CrazyGoat\RabbitStream\Client\Connection;
use CrazyGoat\RabbitStream\Enum\ResponseCodeEnum;
use CrazyGoat\RabbitStream\Request\MetadataRequestV1;

// Create a raw StreamConnection and complete the handshake on it:
$stream = new StreamConnection($host, $port);
$stream->connect();
$connection = Connection::create(host: $host, port: $port, streamConnection: $stream);

$stream->sendMessage(new MetadataRequestV1(['my-stream']));
$response = $stream->readMessage();

$exists = false;
foreach ($response->getStreamMetadata() as $meta) {
    if ($meta->getStreamName() === 'my-stream') {
        $exists = $meta->getResponseCode() === ResponseCodeEnum::OK->value;
        break;
    }
}

Stream Statistics

Getting Statistics

Retrieve metrics about a stream:

$stats = $connection->getStreamStats('my-stream');

foreach ($stats as $key => $value) {
    echo "$key: $value\n";
}

Available Statistics

Statistic	Description
`first_offset`	Offset of the first message in the stream
`last_offset`	Offset of the last message in the stream
`committed_chunk_id`	ID of the last committed chunk
`chunk_count`	Total number of chunks in the stream

Calculating Message Count

$stats = $connection->getStreamStats('my-stream');

$firstOffset = $stats['first_offset'] ?? 0;
$lastOffset = $stats['last_offset'] ?? 0;
$messageCount = $lastOffset - $firstOffset + 1;

echo "Stream contains $messageCount messages\n";
echo "Number of chunks: " . ($stats['chunk_count'] ?? 0) . "\n";

Stream Metadata

Querying Metadata

Get detailed topology information about streams:

use CrazyGoat\RabbitStream\Response\MetadataResponseV1;

// Query single stream
$metadata = $connection->getMetadata(['my-stream']);

// Query multiple streams at once
$metadata = $connection->getMetadata(['stream1', 'stream2', 'stream3']);

Response Structure

The MetadataResponseV1 contains two main sections:

// Broker information
foreach ($metadata->getBrokers() as $broker) {
    echo "Broker {$broker->getReference()}: ";
    echo "{$broker->getHost()}:{$broker->getPort()}\n";
}

// Stream metadata
foreach ($metadata->getStreamMetadata() as $streamMeta) {
    echo "Stream: {$streamMeta->getStreamName()}\n";
    echo "  Response Code: {$streamMeta->getResponseCode()}\n";
    echo "  Leader: Broker {$streamMeta->getLeaderReference()}\n";
    echo "  Replicas: " . implode(', ', $streamMeta->getReplicasReferences()) . "\n";
}

Multi-Stream Queries

Querying multiple streams in a single request is more efficient:

$streamNames = ['orders', 'events', 'logs'];
$metadata = $connection->getMetadata($streamNames);

foreach ($metadata->getStreamMetadata() as $streamMeta) {
    $name = $streamMeta->getStreamName();
    $code = $streamMeta->getResponseCode();
    
    if ($code === ResponseCodeEnum::OK->value) {
        $leader = $streamMeta->getLeaderReference();
        $replicas = $streamMeta->getReplicasReferences();
        echo "$name: leader=$leader, replicas=" . count($replicas) . "\n";
    } else {
        echo "$name: error - " . ResponseCodeEnum::from($code)->getMessage() . "\n";
    }
}

Low-Level Commands

When to Use Low-Level API

Use the low-level protocol classes when:

Debugging protocol issues
Learning the RabbitMQ Stream protocol
Implementing custom connection logic
Need fine-grained error handling

Available Classes

Request Classes:

CreateRequestV1 - Create a stream
DeleteStreamRequestV1 - Delete a stream
MetadataRequestV1 - Query stream metadata
StreamStatsRequestV1 - Get stream statistics

Response Classes:

CreateResponseV1 - Create operation result
DeleteStreamResponseV1 - Delete operation result
MetadataResponseV1 - Metadata query result
StreamStatsResponseV1 - Statistics query result

Low-Level Example

use CrazyGoat\RabbitStream\StreamConnection;
use CrazyGoat\RabbitStream\Client\Connection;
use CrazyGoat\RabbitStream\Request\CreateRequestV1;
use CrazyGoat\RabbitStream\Request\MetadataRequestV1;
use CrazyGoat\RabbitStream\Request\StreamStatsRequestV1;
use CrazyGoat\RabbitStream\Request\DeleteStreamRequestV1;

// Create a raw StreamConnection and complete the handshake on it:
$stream = new StreamConnection($host, $port);
$stream->connect();
$connection = Connection::create(host: $host, port: $port, streamConnection: $stream);

// Create
$stream->sendMessage(new CreateRequestV1('my-stream'));
$createResponse = $stream->readMessage();

// Get metadata
$stream->sendMessage(new MetadataRequestV1(['my-stream']));
$metadata = $stream->readMessage();

// Get stats
$stream->sendMessage(new StreamStatsRequestV1('my-stream'));
$stats = $stream->readMessage();

// Delete
$stream->sendMessage(new DeleteStreamRequestV1('my-stream'));
$deleteResponse = $stream->readMessage();

Complete Working Example

use CrazyGoat\RabbitStream\Client\Connection;
use CrazyGoat\RabbitStream\Enum\ResponseCodeEnum;

$connection = Connection::create(
    host: '127.0.0.1',
    port: 5552,
    user: 'guest',
    password: 'guest',
    vhost: '/'
);

try {
    $streamName = 'events-stream';
    
    // 1. Create stream with retention policy
    echo "Creating stream...\n";
    $connection->createStream($streamName, [
        'max-length-bytes' => '1073741824',  // 1 GB
        'max-age' => '24h',                   // 24 hours
    ]);
    echo "Stream created\n";
    
    // 2. Check existence
    if ($connection->streamExists($streamName)) {
        echo "Stream exists: yes\n";
    }
    
    // 3. Get metadata
    $metadata = $connection->getMetadata([$streamName]);
    foreach ($metadata->getStreamMetadata() as $meta) {
        echo "Leader: Broker {$meta->getLeaderReference()}\n";
        echo "Replicas: " . implode(', ', $meta->getReplicasReferences()) . "\n";
    }
    
    // 4. Get statistics
    $stats = $connection->getStreamStats($streamName);
    echo "First offset: {$stats['first_offset']}\n";
    echo "Last offset: {$stats['last_offset']}\n";
    echo "Chunk count: {$stats['chunk_count']}\n";
    
    // 5. Delete stream
    echo "Deleting stream...\n";
    $connection->deleteStream($streamName);
    echo "Stream deleted\n";
    
} finally {
    $connection->close();
}

Best Practices

Idempotent Creation Pattern

Make your code resilient to stream already existing:

function ensureStreamExists(Connection $connection, string $name, array $args = []): void
{
    try {
        $connection->createStream($name, $args);
    } catch (\Exception $e) {
        // Check if it's already exists error
        if (strpos($e->getMessage(), 'already exists') !== false) {
            return; // Stream exists, that's fine
        }
        throw $e;
    }
}

// Usage
ensureStreamExists($connection, 'my-stream', ['max-age' => '7d']);

Proper Error Handling

Always handle specific error codes:

use CrazyGoat\RabbitStream\Enum\ResponseCodeEnum;

function handleStreamError(ResponseCodeEnum $code, string $operation): void
{
    match ($code) {
        ResponseCodeEnum::OK => null, // Success, no action needed
        ResponseCodeEnum::STREAM_ALREADY_EXISTS => 
            throw new \RuntimeException("Stream already exists during $operation"),
        ResponseCodeEnum::STREAM_NOT_EXIST => 
            throw new \RuntimeException("Stream not found during $operation"),
        ResponseCodeEnum::ACCESS_REFUSED => 
            throw new \RuntimeException("Access denied during $operation"),
        ResponseCodeEnum::PRECONDITION_FAILED => 
            throw new \RuntimeException("Invalid arguments during $operation"),
        default => throw new \RuntimeException("Stream $operation failed: " . $code->getMessage()),
    };
}

Metadata Caching

Cache metadata to reduce server load:

use CrazyGoat\RabbitStream\Client\Connection;
use CrazyGoat\RabbitStream\VO\StreamMetadata;

class StreamMetadataCache
{
    private array $cache = [];
    private int $ttl;
    
    public function __construct(private Connection $connection, int $ttlSeconds = 60)
    {
        $this->ttl = $ttlSeconds;
    }
    
    public function getMetadata(string $stream): ?StreamMetadata
    {
        $key = $stream;
        $now = time();
        
        if (isset($this->cache[$key]) && $this->cache[$key]['expires'] > $now) {
            return $this->cache[$key]['data'];
        }
        
        $response = $this->connection->getMetadata([$stream]);
        foreach ($response->getStreamMetadata() as $meta) {
            if ($meta->getStreamName() === $stream) {
                $this->cache[$key] = [
                    'data' => $meta,
                    'expires' => $now + $this->ttl,
                ];
                return $meta;
            }
        }
        
        return null;
    }
}

Cleanup in Finally Blocks

Always ensure streams are cleaned up:

$streamName = 'temp-stream-' . uniqid();

try {
    $connection->createStream($streamName);
    // ... use the stream ...
} finally {
    // Always clean up, even on error
    try {
        $connection->deleteStream($streamName);
    } catch (\Exception $e) {
        // Log but don't throw from cleanup
        error_log("Failed to delete stream: " . $e->getMessage());
    }
}

Navigation Menu

Repository navigation

stream-management.md

stream-management.md

Stream Management

Overview

Creating Streams

Basic Stream Creation

Stream with Arguments

Available Stream Arguments

Error Handling: STREAM_ALREADY_EXISTS

Deleting Streams

Basic Deletion

Error Handling: STREAM_NOT_EXIST

Checking Stream Existence

Using streamExists()

How It Works Internally

Stream Statistics

Getting Statistics

Available Statistics

Calculating Message Count

Stream Metadata

Querying Metadata

Response Structure

Multi-Stream Queries

Low-Level Commands

When to Use Low-Level API

Available Classes

Low-Level Example

Complete Working Example

Best Practices

Idempotent Creation Pattern

Proper Error Handling

Metadata Caching

Cleanup in Finally Blocks

See Also

FilesExpand file tree

stream-management.md

Latest commit

History

stream-management.md

File metadata and controls

Stream Management

Overview

Creating Streams

Basic Stream Creation

Stream with Arguments

Available Stream Arguments

Error Handling: STREAM_ALREADY_EXISTS

Deleting Streams

Basic Deletion

Error Handling: STREAM_NOT_EXIST

Checking Stream Existence

Using streamExists()

How It Works Internally

Stream Statistics

Getting Statistics

Available Statistics

Calculating Message Count

Stream Metadata

Querying Metadata

Response Structure

Multi-Stream Queries

Low-Level Commands

When to Use Low-Level API

Available Classes

Low-Level Example

Complete Working Example

Best Practices

Idempotent Creation Pattern

Proper Error Handling

Metadata Caching

Cleanup in Finally Blocks

See Also