Connection API Reference

Complete API reference for the CrazyGoat\RabbitStream\Client\Connection class.

Class Overview

namespace CrazyGoat\RabbitStream\Client;

class Connection
{
    // Static factory method
    public static function create(
        string $host = '127.0.0.1',
        int $port = 5552,
        string $user = 'guest',
        string $password = 'guest',
        string $vhost = '/',
        ?BinarySerializerInterface $serializer = null,
        ?LoggerInterface $logger = null,
        ?int $requestedFrameMax = null,
        ?int $requestedHeartbeat = null,
        ?int $maxDeliverFrameSize = null,
        ?StreamConnection $streamConnection = null,
        ?float $socketTimeout = null,
        ?TlsConfig $tls = null,
        ?int $initialFrameMax = null,
    ): self;
    
    // Stream management
    public function createStream(string $name, array $arguments = []): void;
    public function deleteStream(string $name): void;
    public function streamExists(string $name): bool;
    public function getStreamStats(string $name): array;
    public function getMetadata(array $streams): MetadataResponseV1;
    
    // Offset management
    public function queryOffset(string $reference, string $stream): ?int;
    public function storeOffset(string $reference, string $stream, int $offset): void;
    
    // Super streams
    public function createSuperStream(
        string $name,
        array $partitions = [],
        array $bindingKeys = [],
        array $arguments = []
    ): void;
    public function deleteSuperStream(string $name): void;
    public function route(string $routingKey, string $superStream): array;
    public function partitions(string $superStream): array;
    
    // Factory methods
    public function createProducer(
        string $stream,
        ?string $name = null,
        ?callable $onConfirm = null,
        int $maxPendingConfirms = Producer::DEFAULT_MAX_PENDING_CONFIRMS,
        float $redeclareTimeout = Producer::DEFAULT_REDECLARE_TIMEOUT,
        float $closeConfirmDrainTimeout = Producer::DEFAULT_CLOSE_CONFIRM_DRAIN_TIMEOUT,
    ): ProducerInterface;
    
    public function createConsumer(
        string $stream,
        OffsetSpec $offset,
        ?string $name = null,
        int $autoCommit = 0,
        int $initialCredit = 10,
        array $filterValues = [],
        bool $matchUnfiltered = false,
        bool $singleActiveConsumer = false,
        ?string $superStream = null,
        int $creditWindowBytes = Consumer::DEFAULT_CREDIT_WINDOW_BYTES,
        int $maxDecodeDepth = AmqpDecoder::MAX_RECURSION_DEPTH,
        bool $verifyCrc = true,
        int $maxBufferSize = 1000,
    ): ConsumerInterface;
    
    public function createSuperStreamProducer(
        string $superStream,
        ?RoutingStrategy $strategy = null,
        ?string $name = null,
        ?callable $onConfirm = null,
        int $maxPendingConfirms = Producer::DEFAULT_MAX_PENDING_CONFIRMS,
        float $redeclareTimeout = Producer::DEFAULT_REDECLARE_TIMEOUT,
    ): SuperStreamProducerInterface;
    
    public function createSuperStreamConsumer(
        string $superStream,
        OffsetSpec $offset,
        ?string $name = null,
        int $autoCommit = 0,
        int $initialCredit = 10,
        bool $singleActiveConsumer = false,
        int $creditWindowBytes = Consumer::DEFAULT_CREDIT_WINDOW_BYTES,
        int $maxDecodeDepth = AmqpDecoder::MAX_RECURSION_DEPTH,
        bool $verifyCrc = true,
        int $maxBufferSize = 1000,
    ): SuperStreamConsumerInterface;
    
    // Lifecycle
    public function readLoop(?int $maxFrames = null, ?float $timeout = null): int;
    public function isConnected(): bool;
    public function close(): void;
    
    // Protocol capabilities
    public function supportsCommandVersion(KeyEnum $key, int $version): bool;
    public function getSupportedCommandVersions(): array;
}

Connection::create()

Static factory method to create and establish a connection to RabbitMQ Streams.

public static function create(
    string $host = '127.0.0.1',
    int $port = 5552,
    string $user = 'guest',
    string $password = 'guest',
    string $vhost = '/',
    ?BinarySerializerInterface $serializer = null,
    ?LoggerInterface $logger = null,
    ?int $requestedFrameMax = null,
    ?int $requestedHeartbeat = null,
    ?int $maxDeliverFrameSize = null,
    ?StreamConnection $streamConnection = null,
    ?float $socketTimeout = null,
    ?TlsConfig $tls = null,
    ?int $initialFrameMax = null,
): self

Parameters

Parameter	Type	Required	Description
`$host`	`string`	No	RabbitMQ server hostname or IP. Default: `127.0.0.1`
`$port`	`int`	No	RabbitMQ Streams port. Default: `5552`
`$user`	`string`	No	Username for authentication. Default: `guest`
`$password`	`string`	No	Password for authentication. Default: `guest`
`$vhost`	`string`	No	Virtual host. Default: `/`
`$serializer`	`?BinarySerializerInterface`	No	Custom binary serializer. Default: `PhpBinarySerializer`
`$logger`	`?LoggerInterface`	No	PSR-3 logger for debugging. Default: `NullLogger`
`$requestedFrameMax`	`?int`	No	Requested control-frame max in bytes. `null` (default) accepts the server value but never raises the effective cap above `StreamConnection::DEFAULT_MAX_FRAME_SIZE`; `0` expresses no client preference, so the server value is used without that safety cap; a non-zero value is a deliberate raise or lower — see negotiation caveat below
`$requestedHeartbeat`	`?int`	No	Requested heartbeat interval in seconds. `null` (default) accepts the server value; `0` expresses no client preference, so the server value is used as well
`$maxDeliverFrameSize`	`?int`	No	Max size in bytes for incoming Deliver frames (key `0x0008`), which the broker does not bound by the negotiated `frame_max` — see Deliver frames need their own cap below. Default: `null` (`StreamConnection::DEFAULT_MAX_DELIVER_FRAME_SIZE`, 64MB)
`$streamConnection`	`?StreamConnection`	No	Pre-connected stream connection to reuse (advanced use, principally a test injection seam). Default: `null`. When supplied it is not connected here, but the returned `Connection` still closes it in `close()`.
`$socketTimeout`	`?float`	No	Per-I/O-call timeout in seconds, enforced by a non-blocking `stream_select()` deadline — bounds a single read or write so a stalled peer cannot hang the client. Must be > 0. Not used to build a connection when `$streamConnection` is supplied (configure the injected connection instead), but still validated. Default: `null` (`StreamConnection::DEFAULT_SOCKET_TIMEOUT`, 30 s)
`$tls`	`?TlsConfig`	No	TLS transport options (`CrazyGoat\RabbitStream\VO\TlsConfig`); passing one selects the encrypted `ssl://` transport (RabbitMQ stream TLS listener) instead of plaintext `tcp://`, with peer and hostname verification on by default. Ignored when `$streamConnection` is supplied (configure it there). Default: `null`.
`$initialFrameMax`	`?int`	No	Maximum frame size in bytes the client allows on frames it sends before `Open` completes. Mirrors the broker's `stream.initial_frame_max` (RabbitMQ 4.3 caps incoming frames at 8192 bytes until Open, regardless of the value later negotiated at Tune). Default: `null`, which applies `StreamConnection::DEFAULT_INITIAL_FRAME_SIZE` (8192). The value is applied to the connection — including a supplied `$streamConnection` — and overrides any pre-Open cap already set on it. Pass `initialFrameMax` to choose a different value, or `0` to disable the client-side pre-Open check.

Return Value

Connection - A fully established and authenticated connection

Exceptions

InvalidArgumentException - If requestedFrameMax, requestedHeartbeat, maxDeliverFrameSize, or initialFrameMax is negative, or if socketTimeout is not positive. This is CrazyGoat\RabbitStream\Exception\InvalidArgumentException, which implements RabbitStreamExceptionInterface while still extending the native \InvalidArgumentException
ProtocolException - If a request sent before Open completes serializes to a payload larger than initialFrameMax (the broker enforces stream.initial_frame_max until then); the exception names the offending command
AuthenticationException - If the server does not offer the PLAIN SASL mechanism. Invalid credentials are not reported as this type: the non-OK SaslAuthenticate response code is asserted during deserialization and raises a ProtocolException (of which AuthenticationException is a subclass)
UnexpectedResponseException - If the server returns an unexpected response during handshake
DeserializationException - If a handshake response frame cannot be deserialized
TimeoutException - If a handshake write does not complete within $socketTimeout, or a handshake response does not arrive within the read timeout (30 s)
ConnectionException - If the TCP connection cannot be established, or a read or write fails during the handshake

Deliver frames need their own cap

The broker does not enforce the negotiated frame_max on Deliver frames: a stream chunk is sent whole, however big it grew due to server-side coalescing (e.g. a fast producer publishing several batches without calling waitForConfirms() between them). StreamConnection therefore caps Deliver frames with a separate, larger limit (maxDeliverFrameSize, default 64MB) instead of the same limit used for every other incoming frame (StreamConnection::setMaxFrameSize() / DEFAULT_MAX_FRAME_SIZE, 8MB). See Performance Tuning → Frame Size Limits for the full write-up and a reproduction of the underlying broker behavior.

`frame_max` negotiation only lowers the default

If you don't pass requestedFrameMax, a broker advertising a huge frame_max during Tune (including 0xFFFFFFFF) cannot raise the effective control-frame cap above StreamConnection::DEFAULT_MAX_FRAME_SIZE — the negotiated value is only ever used to lower it. Pass requestedFrameMax explicitly to raise (or lower) the cap deliberately.

Outgoing frames over the limit fail fast

Two distinct windows apply, with two distinct exceptions:

Before Open completes, the broker enforces its own low stream.initial_frame_max (8192 bytes by default) on every incoming frame, regardless of the value negotiated at Tune. A request that serializes above the client-side initialFrameMax is rejected before anything is written, raising CrazyGoat\RabbitStream\Exception\ProtocolException naming the command (e.g. SaslAuthenticateRequestV1), instead of letting the broker drop the connection with an opaque "Frame too large".
After Open completes, frames are bound by the negotiated frame_max. An oversized frame is rejected before anything is written, raising CrazyGoat\RabbitStream\Exception\InvalidArgumentException with the frame size and the limit in the message.

In both cases the connection stays connected and usable. Previously the oversized frame was written, the broker closed the connection, and the failure surfaced later as an unrelated-looking "Cannot read: socket is not connected".

Connection Handshake Flow

When create() is called, it performs the following protocol handshake:

PeerProperties - Exchange client/server capabilities
SaslHandshake - Negotiate authentication mechanism (requires PLAIN)
SaslAuthenticate - Authenticate with username/password
Tune - Negotiate frame max and heartbeat settings
Open - Open the virtual host
ExchangeCommandVersions - Negotiate per-command versions (best effort; see below)

Version negotiation

After Open, create() sends ExchangeCommandVersions (0x001b) advertising the per-command versions this client implements, and stores the broker's reply for the lifetime of the connection. Only commands with more than one version are advertised — today only Publish (v1–v2) — because a single-version command is already covered by the v1 baseline, and RabbitMQ terminates the stream connection process when an advertised command key is one its release does not define.

The step is best effort: on a broker that does not implement the command (RabbitMQ < 3.11, where it is skipped entirely), does not answer within 5 seconds, or rejects it, the connection stays on the v1 baseline and create() succeeds. Check the result with:

use CrazyGoat\RabbitStream\Enum\KeyEnum;

// True only when the broker's reported range includes the version.
$connection->supportsCommandVersion(KeyEnum::PUBLISH, 2);

// The raw broker ranges, keyed by protocol command key (empty on fallback).
$ranges = $connection->getSupportedCommandVersions();

Producer::sendWithFilter() uses this automatically: a non-null filter value requires Publish v2 and throws a ProtocolException when the broker did not negotiate it, while a null filter value (and every broker without v2) publishes with the plain v1 frame.

Examples

Basic connection:

use CrazyGoat\RabbitStream\Client\Connection;

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

Connection with an explicit serializer:

use CrazyGoat\RabbitStream\Serializer\PhpBinarySerializer;

$connection = Connection::create(
    host: 'rabbitmq.example.com',
    serializer: new PhpBinarySerializer()
);

Connection with PSR-3 logger:

use Monolog\Logger;
use Monolog\Handler\StreamHandler;

$logger = new Logger('rabbitmq');
$logger->pushHandler(new StreamHandler('php://stderr'));

$connection = Connection::create(
    host: 'localhost',
    logger: $logger
);

Connection with custom frame size and heartbeat:

$connection = Connection::create(
    host: 'localhost',
    requestedFrameMax: 1048576,  // 1MB
    requestedHeartbeat: 30         // 30 seconds
);

Connection with a raised Deliver frame cap:

// Only needed if you expect broker-coalesced chunks bigger than the 64MB
// default (see "Deliver frames need their own cap" above).
$connection = Connection::create(
    host: 'localhost',
    maxDeliverFrameSize: 128 * 1024 * 1024,
);

Stream Management Methods

createStream()

Create a new stream with optional arguments.

/**
 * @param array<string, string> $arguments
 */
public function createStream(string $name, array $arguments = []): void

Parameters

Parameter	Type	Required	Description
`$name`	`string`	Yes	Name of the stream to create
`$arguments`	`array<string, string>`	No	Stream configuration arguments (e.g., `max-length-bytes`, `max-age`)

Return Value

void

Exceptions

UnexpectedResponseException - If the server returns an unexpected response
ProtocolException - If the stream already exists or arguments are invalid

Example

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

// Create a stream with retention policy
$connection->createStream('events', [
    'max-length-bytes' => '1073741824',  // 1GB max size
    'max-age' => '7D',                   // 7 days retention
]);

deleteStream()

Delete an existing stream.

public function deleteStream(string $name): void

Parameters

Parameter	Type	Required	Description
`$name`	`string`	Yes	Name of the stream to delete

Return Value

void

Exceptions

UnexpectedResponseException - If the server returns an unexpected response
ProtocolException - If the stream does not exist

Example

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

streamExists()

Check if a stream exists.

public function streamExists(string $name): bool

Parameters

Parameter	Type	Required	Description
`$name`	`string`	Yes	Name of the stream to check

Return Value

bool - true if the stream exists, false otherwise

Exceptions

UnexpectedResponseException - If the server returns an unexpected response

Example

if (!$connection->streamExists('my-stream')) {
    $connection->createStream('my-stream');
}

getStreamStats()

Get statistics for a stream.

/**
 * @return array<string, int>
 */
public function getStreamStats(string $name): array

Parameters

Parameter	Type	Required	Description
`$name`	`string`	Yes	Name of the stream

Return Value

array<string, int> - Associative array of statistics. Common keys include:

publishers - Number of active publishers
consumers - Number of active consumers
messages - Total number of messages in the stream
bytes - Total size of the stream in bytes

Exceptions

UnexpectedResponseException - If the server returns an unexpected response
ProtocolException - If the stream does not exist

Example

$stats = $connection->getStreamStats('my-stream');
echo "Messages: {$stats['messages']}\n";
echo "Size: {$stats['bytes']} bytes\n";
echo "Publishers: {$stats['publishers']}\n";
echo "Consumers: {$stats['consumers']}\n";

getMetadata()

Get metadata for one or more streams.

/**
 * @param array<int, string> $streams
 */
public function getMetadata(array $streams): MetadataResponseV1

Parameters

Parameter	Type	Required	Description
`$streams`	`array<int, string>`	Yes	Array of stream names to query

Return Value

MetadataResponseV1 - Response object containing metadata for each stream:

getStreamMetadata() - Returns array of StreamMetadata objects
Each StreamMetadata has:
- getStreamName() - Stream name
- getResponseCode() - Response code (OK or error)
- getLeaderReference() - Leader broker reference
- getReplicasReferences() - Replica broker references

Exceptions

UnexpectedResponseException - If the server returns an unexpected response

Example

use CrazyGoat\RabbitStream\Enum\ResponseCodeEnum;

$metadata = $connection->getMetadata(['stream1', 'stream2']);

foreach ($metadata->getStreamMetadata() as $streamMeta) {
    $name = $streamMeta->getStreamName();
    $code = $streamMeta->getResponseCode();
    
    if ($code === ResponseCodeEnum::OK->value) {
        echo "Stream {$name} exists\n";
        echo "Leader: broker reference {$streamMeta->getLeaderReference()}\n";
    } else {
        echo "Stream {$name} error: {$code}\n";
    }
}

Super Stream Methods

See the Super Streams Guide for a full walkthrough; this section is the reference for the four Connection methods it uses.

createSuperStream()

Create a super stream: a logical stream backed by several physical partition streams, with broker-side exchange bindings between them.

/**
 * @param string[] $partitions
 * @param string[] $bindingKeys
 * @param array<string, string> $arguments
 */
public function createSuperStream(
    string $name,
    array $partitions = [],
    array $bindingKeys = [],
    array $arguments = []
): void

Parameters

Parameter	Type	Required	Description
`$name`	`string`	Yes	Name of the super stream
`$partitions`	`string[]`	No	Names of the physical partition streams to create (e.g. `['orders-0', 'orders-1']`)
`$bindingKeys`	`string[]`	No	Exchange binding key per partition, same order as `$partitions` — matched by `route()`
`$arguments`	`array<string, string>`	No	Per-partition stream arguments, same keys as `createStream()` (e.g. `max-length-bytes`, `max-age`)

Return Value

void

Exceptions

UnexpectedResponseException - If the server returns an unexpected response
ProtocolException - If the super stream already exists or arguments are invalid

Example

$connection->createSuperStream(
    'orders',
    ['orders-0', 'orders-1', 'orders-2'],
    ['0', '1', '2'],
    ['max-age' => '24h']
);

deleteSuperStream()

Delete a super stream and all of its partitions.

public function deleteSuperStream(string $name): void

Parameters

Parameter	Type	Required	Description
`$name`	`string`	Yes	Name of the super stream to delete

Return Value

void

Exceptions

UnexpectedResponseException - If the server returns an unexpected response
ProtocolException - If the super stream does not exist

Example

$connection->deleteSuperStream('orders');

route()

Ask the broker which partition(s) a routing key maps to, via the exchange bindings created with createSuperStream()'s $bindingKeys.

/** @return string[] */
public function route(string $routingKey, string $superStream): array

Parameters

Parameter	Type	Required	Description
`$routingKey`	`string`	Yes	The key to route
`$superStream`	`string`	Yes	Name of the super stream

Return Value

string[] - The partition stream name(s) matching this routing key. Like the Java client, more than one partition can legitimately be returned for a single key (overlapping binding keys).

Exceptions

UnexpectedResponseException - If the server returns an unexpected response

Example

$streams = $connection->route('customer-123', 'orders');
echo implode(', ', $streams); // e.g. "orders-1"

Notes

One broker round trip per call. KeyRoutingStrategy (used by createSuperStreamProducer()) wraps this with an in-memory per-key cache.

partitions()

Resolve a super stream's partition (physical stream) names.

/** @return list<string> */
public function partitions(string $superStream): array

Parameters

Parameter	Type	Required	Description
`$superStream`	`string`	Yes	Name of the super stream

Return Value

list<string> - The partition stream names, e.g. ['orders-0', 'orders-1', 'orders-2']

Exceptions

UnexpectedResponseException - If the server returns an unexpected response
ProtocolException - If the super stream does not exist, or exists but currently has zero partitions

Example

$partitions = $connection->partitions('orders');
foreach ($partitions as $partition) {
    echo "Partition: {$partition}\n";
}

Notes

createSuperStreamProducer() and createSuperStreamConsumer() call this internally to resolve partitions

Offset Management Methods

queryOffset()

Query the last stored offset for a named consumer.

public function queryOffset(string $reference, string $stream): ?int

Parameters

Parameter	Type	Required	Description
`$reference`	`string`	Yes	Consumer name (reference)
`$stream`	`string`	Yes	Stream name

Return Value

int|null - The last stored offset for this consumer on this stream, or null when nothing has been stored yet for the pair (the broker's NO_OFFSET, 0x13). null is a normal first-run outcome, not an error.

Exceptions

UnexpectedResponseException - If the server returns an unexpected response
ProtocolException - If the broker returns any non-OK response code other than NO_OFFSET (for example STREAM_NOT_EXIST)

Example

$offset = $connection->queryOffset('my-consumer', 'my-stream');

if ($offset === null) {
    echo "No offset stored yet\n";
} else {
    echo "Last consumed offset: {$offset}\n";
}

storeOffset()

Store an offset for a named consumer.

public function storeOffset(string $reference, string $stream, int $offset): void

Parameters

Parameter	Type	Required	Description
`$reference`	`string`	Yes	Consumer name (reference)
`$stream`	`string`	Yes	Stream name
`$offset`	`int`	Yes	The next offset to consume, i.e. `lastProcessedOffset + 1` (see `Consumer::storeOffset()`)

Return Value

void

Exceptions

ConnectionException - If the socket is not connected or the write fails
InvalidArgumentException - If the serialized request exceeds the negotiated outgoing frame size
TimeoutException - If the frame cannot be written within the socket timeout

Notes

StoreOffset is a one-way command: the protocol defines no response for it, so no round trip is awaited and a broker error is not reported here.

Example

// Store offset 12345 for consumer 'my-consumer' on stream 'my-stream'
$connection->storeOffset('my-consumer', 'my-stream', 12345);

Factory Methods

Publisher and subscription ids are a per-connection resource

DeclarePublisher and Subscribe encode their id as a uint8, so one connection holds at most 256 producers and 256 consumers at a time (Connection::MAX_CONCURRENT_PUBLISHERS / MAX_CONCURRENT_SUBSCRIPTIONS).

close() on a producer or consumer hands its id back and drops the connection's reference to it, so a long-lived worker that creates and closes one producer per work unit runs indefinitely. Ids are handed out round-robin, so a freed id is only reused after the whole range has been used once — that keeps a late PublishConfirm/Deliver for a closed publisher/subscription from landing on a fresh one that shares its id.

Creating a 257th simultaneous producer (or consumer) throws ConnectionException naming the exhausted pool: close what you no longer need, or open another connection.

foreach ($workUnits as $unit) {
    $producer = $connection->createProducer($unit->stream);
    $producer->send($unit->payload);
    $producer->waitForConfirms();
    $producer->close();   // <- frees the publisher id
}

createProducer()

Create a new producer for publishing messages to a stream.

public function createProducer(
    string $stream,
    ?string $name = null,
    ?callable $onConfirm = null,
    int $maxPendingConfirms = Producer::DEFAULT_MAX_PENDING_CONFIRMS,
    float $redeclareTimeout = Producer::DEFAULT_REDECLARE_TIMEOUT,
    float $closeConfirmDrainTimeout = Producer::DEFAULT_CLOSE_CONFIRM_DRAIN_TIMEOUT,
): ProducerInterface

Parameters

Parameter	Type	Required	Description
`$stream`	`string`	Yes	Name of the stream to publish to
`$name`	`?string`	No	Unique producer name for deduplication. If provided, the producer reads back its publishing sequence on creation and resumes from it, so a reconnect/restart continues above the broker's stored sequence instead of replaying it.
`$onConfirm`	`?callable`	No	Callback invoked for each publish confirmation. Receives `ConfirmationStatus`.
`$maxPendingConfirms`	`int`	No	Back-pressure cap on outstanding publishes. Default: `Producer::DEFAULT_MAX_PENDING_CONFIRMS`.
`$redeclareTimeout`	`float`	No	How long a publish keeps retrying `DeclarePublisher` after a `MetadataUpdate` dropped the publisher. Default: `Producer::DEFAULT_REDECLARE_TIMEOUT` (5.0 s). See Producer::isStale().
`$closeConfirmDrainTimeout`	`float`	No	How long `close()` waits for in-flight publish confirms before abandoning them. Default: `Producer::DEFAULT_CLOSE_CONFIRM_DRAIN_TIMEOUT` (2.0 s). Must be `>= 0`.

Return Value

ProducerInterface - A producer instance ready to send messages

Exceptions

ProtocolException - If the stream does not exist
ConnectionException - If all 256 publisher ids of this connection are in use

Example

use CrazyGoat\RabbitStream\Client\ConfirmationStatus;

// Anonymous producer
$producer = $connection->createProducer('my-stream');

// Named producer with confirm callback
$producer = $connection->createProducer(
    'my-stream',
    name: 'my-producer',
    onConfirm: function (ConfirmationStatus $status) {
        if ($status->isConfirmed()) {
            echo "Confirmed: #{$status->getPublishingId()}\n";
        }
    }
);

See Also: Producer API Reference

createConsumer()

Create a new consumer for reading messages from a stream.

use CrazyGoat\RabbitStream\VO\OffsetSpec;

public function createConsumer(
    string $stream,
    OffsetSpec $offset,
    ?string $name = null,
    int $autoCommit = 0,
    int $initialCredit = 10,
    array $filterValues = [],
    bool $matchUnfiltered = false,
    bool $singleActiveConsumer = false,
    ?string $superStream = null,
    int $creditWindowBytes = Consumer::DEFAULT_CREDIT_WINDOW_BYTES,
    int $maxDecodeDepth = AmqpDecoder::MAX_RECURSION_DEPTH,
    bool $verifyCrc = true,
    int $maxBufferSize = 1000,
): ConsumerInterface

Parameters

Parameter	Type	Required	Description
`$stream`	`string`	Yes	Name of the stream to consume from
`$offset`	`OffsetSpec`	Yes	Starting offset specification (see `OffsetSpec` factory methods)
`$name`	`?string`	No	Consumer name for offset tracking. Required for `storeOffset()`, `queryOffset()`, and `singleActiveConsumer`.
`$autoCommit`	`int`	No	Auto-commit interval (number of messages). `0` disables auto-commit.
`$initialCredit`	`int`	No	Initial (and minimum) number of chunks in flight, 1–32767. Default: `10`
`$filterValues`	`array<int, string>`	No	Broker-side stream filtering values (`filter.0`, `filter.1`, ... properties). Chunk-granular — see the Consumer API reference.
`$matchUnfiltered`	`bool`	No	When `$filterValues` is non-empty, also deliver messages published with no filter value.
`$singleActiveConsumer`	`bool`	No	Enables single active consumer for this subscription. Requires `$name`.
`$superStream`	`?string`	No	Name of the super stream this partition belongs to.
`$creditWindowBytes`	`int`	No	Adaptive credit window in bytes (default 8 MiB). The consumer keeps `ceil(creditWindowBytes / observed average chunk size)` chunks in flight, never fewer than `$initialCredit`, never more than 32,767. `0` pins the window to `$initialCredit` chunks. See Flow Control.
`$maxDecodeDepth`	`int`	No	Maximum AMQP nesting depth accepted when a delivered message is decoded. Default `32`, which is ample for real messages; a deeper frame is rejected with `DeserializationException`. Raise it only for a producer that legitimately nests deeper — each level costs a PHP stack frame. Because decoding is lazy, the limit is enforced by the accessor (`Message::getBody()`), not by `read()`.
`$verifyCrc`	`bool`	No	Verify the CRC32 in every chunk header (default `true`). A mismatching chunk is rejected with `DeserializationException`. Disable only where the cost matters or corruption is handled upstream.
`$maxBufferSize`	`int`	No	Message-granular back-pressure ceiling: the target maximum number of unread messages held in memory (default `1000`). Must be positive. A delivered chunk is atomic and is never split or dropped, so the buffer can overshoot by the chunks already in flight; once the unread count reaches the limit no new credit is granted until `read()`/`readOne()` drains the buffer. See Consumer Flow Control.

OffsetSpec Factory Methods

OffsetSpec::first() - Start from the first message
OffsetSpec::last() - Start from the last chunk of messages, delivered in full
OffsetSpec::next() - Start at the end of the stream; receive messages published after subscription
OffsetSpec::offset(int $offset) - Start from a specific offset (inclusive); if retention has removed it, the broker attaches at the first offset still available
OffsetSpec::timestamp(int $timestamp) - Start at the first chunk with chunk timestamp >= the value, delivered in full (chunk-granular). The value is in milliseconds since the Unix epoch

Return Value

ConsumerInterface - A consumer instance ready to read messages

Exceptions

ProtocolException - If the stream does not exist or offset is invalid
ConnectionException - If all 256 subscription ids of this connection are in use
InvalidArgumentException - If $maxBufferSize is not positive, $initialCredit is outside 1–32767, $creditWindowBytes is negative, $maxDecodeDepth is below 1, or $singleActiveConsumer is set without $name

Example

use CrazyGoat\RabbitStream\VO\OffsetSpec;

// Consumer from the beginning
$consumer = $connection->createConsumer(
    'my-stream',
    OffsetSpec::first()
);

// Named consumer with auto-commit
$consumer = $connection->createConsumer(
    'my-stream',
    OffsetSpec::last(),
    name: 'my-consumer',
    autoCommit: 100,  // Auto-commit every 100 messages
    initialCredit: 50
);

// Consumer from specific offset
$consumer = $connection->createConsumer(
    'my-stream',
    OffsetSpec::offset(1000)
);

See Also: Consumer API Reference

createSuperStreamProducer()

Create a producer that publishes to a super stream's partitions, routing each message via a RoutingStrategy.

use CrazyGoat\RabbitStream\Client\Routing\RoutingStrategy;

public function createSuperStreamProducer(
    string $superStream,
    ?RoutingStrategy $strategy = null,
    ?string $name = null,
    ?callable $onConfirm = null,
    int $maxPendingConfirms = Producer::DEFAULT_MAX_PENDING_CONFIRMS,
    float $redeclareTimeout = Producer::DEFAULT_REDECLARE_TIMEOUT,
): SuperStreamProducerInterface

Parameters

Parameter	Type	Required	Description
`$superStream`	`string`	Yes	Name of the super stream to publish to
`$strategy`	`?RoutingStrategy`	No	Routing strategy. Default: `HashRoutingStrategy` — MurmurHash3 x86_32, seed `104729`, matching the Java/.NET clients exactly, so a routing key lands on the same partition regardless of which client language published it. See `KeyRoutingStrategy` for broker-resolved (exchange binding) routing instead.
`$name`	`?string`	No	Base producer name. Each partition's underlying `Producer` gets the derived name `"{$name}-{$partition}"`, so per-partition sequence dedup still works.
`$onConfirm`	`?callable`	No	Confirmation callback, passed through to every partition's `Producer`.
`$maxPendingConfirms`	`int`	No	Back-pressure cap, passed through to every partition's `Producer`. Default: `Producer::DEFAULT_MAX_PENDING_CONFIRMS`.
`$redeclareTimeout`	`float`	No	Re-declare timeout after a `MetadataUpdate`, passed through to every partition's `Producer`. Default: `Producer::DEFAULT_REDECLARE_TIMEOUT` (5.0 s).

Return Value

SuperStreamProducerInterface - resolves the super stream's partitions immediately (one partitions() round trip), but opens each partition's Producer lazily on first publish to that partition. A MetadataUpdate on any partition makes the next publish re-resolve the partition list — see SuperStreamProducer::refreshPartitions()

Exceptions

ProtocolException - If the super stream does not exist or has zero partitions

Example

use CrazyGoat\RabbitStream\Client\Routing\KeyRoutingStrategy;

// Default hash routing
$producer = $connection->createSuperStreamProducer('orders');
$producer->send('Order payload', routingKey: 'customer-123');

// Broker-resolved key routing instead
$producer = $connection->createSuperStreamProducer(
    'orders',
    new KeyRoutingStrategy($connection, 'orders'),
    name: 'order-service',
);

Notes

Partition membership is resolved once, at creation time — a topology change (partition leader change, or the super stream's partition set itself changing) is not automatically detected; recreate the SuperStreamProducer if you need to react to that.

createSuperStreamConsumer()

Create a consumer that subscribes to every partition of a super stream, all sharing the same consumer $name.

public function createSuperStreamConsumer(
    string $superStream,
    OffsetSpec $offset,
    ?string $name = null,
    int $autoCommit = 0,
    int $initialCredit = 10,
    bool $singleActiveConsumer = false,
    int $creditWindowBytes = Consumer::DEFAULT_CREDIT_WINDOW_BYTES,
    int $maxDecodeDepth = AmqpDecoder::MAX_RECURSION_DEPTH,
    bool $verifyCrc = true,
    int $maxBufferSize = 1000,
): SuperStreamConsumerInterface

Parameters

Parameter	Type	Required	Description
`$superStream`	`string`	Yes	Name of the super stream to consume from
`$offset`	`OffsetSpec`	Yes	Starting offset specification, applied to every partition
`$name`	`?string`	No	Consumer name, shared by every partition's subscription. Required for `singleActiveConsumer` — the broker groups the per-partition subscriptions into one single-active-consumer group per partition using this shared name.
`$autoCommit`	`int`	No	Auto-commit interval (messages), passed through to every partition's `Consumer`
`$initialCredit`	`int`	No	Initial flow control credits, passed through to every partition's `Consumer`
`$singleActiveConsumer`	`bool`	No	Enables single active consumer per partition. Requires `$name`.
`$creditWindowBytes`	`int`	No	Adaptive credit window in bytes (default 8 MiB). Passed through to every partition's `Consumer`, which keeps `ceil(creditWindowBytes / observed average chunk size)` chunks in flight, never fewer than `$initialCredit`, never more than 32,767. `0` pins the window to `$initialCredit` chunks. See Flow Control.
`$maxDecodeDepth`	`int`	No	Maximum AMQP nesting depth accepted when a delivered message is decoded. Default `32`; a deeper frame is rejected with `DeserializationException`. Passed through to every partition's `Consumer`.
`$verifyCrc`	`bool`	No	Verify the CRC32 in every chunk header (default `true`). A mismatching chunk is rejected with `DeserializationException`. Disable only where the cost matters or corruption is handled upstream.
`$maxBufferSize`	`int`	No	Message-granular back-pressure ceiling, passed through to every partition's `Consumer` (default `1000`). Must be positive. See `createConsumer()` for the chunk-vs-message contract.

Return Value

SuperStreamConsumerInterface - one plain Consumer per partition, aggregated

Exceptions

ProtocolException - If the super stream does not exist or has zero partitions
InvalidArgumentException - If $maxBufferSize is not positive, or another Consumer argument is out of range

Example

use CrazyGoat\RabbitStream\VO\OffsetSpec;

$consumer = $connection->createSuperStreamConsumer('orders', OffsetSpec::first(), name: 'order-processor');

foreach ($consumer->read(timeout: 5.0) as $message) {
    echo "[{$message->getStream()}] {$message->getBody()}\n";
    $consumer->storeOffset($message->getStream(), $message->getOffset() + 1);
}

Notes

Offset tracking is entirely per-partition — there is no aggregate, super-stream-wide offset. storeOffset()/queryOffset() on the returned consumer always take the partition name explicitly (Message::getStream() names it).

Lifecycle Methods

readLoop()

Process incoming server-push frames (deliveries, heartbeats, confirms).

public function readLoop(?int $maxFrames = null, ?float $timeout = null): int

Parameters

Parameter	Type	Required	Description
`$maxFrames`	`?int`	No	Maximum number of frames to process. `null` means process indefinitely.
`$timeout`	`?float`	No	Timeout in seconds. `null` means block indefinitely.

Return Value

int - Number of frames dispatched; 0 means the loop ended on timeout, stop() or disconnect without handling a frame

Exceptions

ConnectionException - If the connection is lost
DeserializationException - If a server-push frame cannot be deserialized
TimeoutException - If a reply this loop must send (a heartbeat echo, a server-close acknowledgement or a ConsumerUpdate reply) cannot be written within the socket timeout

Example

// Process one frame (useful in consumer loops)
$connection->readLoop(maxFrames: 1, timeout: 5.0);

// Process frames for up to 10 seconds
$connection->readLoop(timeout: 10.0);

// Run indefinitely (until connection closes)
$connection->readLoop();

Notes

Automatically handles heartbeats by echoing them back
Dispatches deliver frames to registered consumers
Dispatches confirm/error frames to registered producers
This method blocks until the specified condition is met

supportsCommandVersion()

Whether the broker reported support for a given version of a command during the ExchangeCommandVersions handshake.

public function supportsCommandVersion(KeyEnum $key, int $version): bool;

When the handshake was unavailable (or the broker did not list the command), the v1 baseline is assumed: true for version 1, false for anything higher.

Parameters

Parameter	Type	Description
`$key`	`KeyEnum`	Command to check, e.g. `KeyEnum::PUBLISH`
`$version`	`int`	Version to check (1-based)

Return Value

bool — true when the broker's reported range includes $version.

Example

use CrazyGoat\RabbitStream\Enum\KeyEnum;

if ($connection->supportsCommandVersion(KeyEnum::PUBLISH, 2)) {
    $producer->sendWithFilter('order created', filterValue: 'region-eu');
}

getSupportedCommandVersions()

The per-command version ranges the broker reported during the handshake.

/** @return array<int, CommandVersion> keyed by protocol command key */
public function getSupportedCommandVersions(): array;

Empty when ExchangeCommandVersions was rejected, unanswered or not implemented by the broker; callers should then assume v1 for every command.

Parameters

None

Return Value

array<int, CommandVersion> — supported ranges keyed by protocol command key.

Example

use CrazyGoat\RabbitStream\Enum\KeyEnum;

$ranges = $connection->getSupportedCommandVersions();
if (isset($ranges[KeyEnum::PUBLISH->value])) {
    echo "Publish up to v{$ranges[KeyEnum::PUBLISH->value]->getMaxVersion()}\n";
}

isConnected()

Whether the underlying socket is still valid.

public function isConnected(): bool

Parameters

None

Return Value

bool - true while the underlying stream resource is still valid

Notes

Local check of the stream resource only; it does not probe the broker
A peer that went away without the socket noticing still reports true until the next read or write fails

close()

Close the connection and all associated producers and consumers.

public function close(): void

Parameters

None

Return Value

void

Exceptions

UnexpectedResponseException - If the server returns an unexpected response during close

Example

try {
    // Use connection...
    $producer->send('message');
    $producer->waitForConfirms();
} finally {
    $connection->close();
}

Notes

Closes all producers and consumers first
Sends Close command to the server
Closes the underlying TCP connection
Safe to call multiple times (idempotent)
Automatically called in __destruct() if not already closed

Usage Patterns

Pattern 1: Basic Connection

Simple connection with automatic cleanup:

use CrazyGoat\RabbitStream\Client\Connection;

$connection = Connection::create('localhost', 5552, 'guest', 'guest');

// Use connection...
$producer = $connection->createProducer('my-stream');
$producer->send('Hello');

// Cleanup
$connection->close();

Pattern 2: Connection with an Explicit Serializer

Connection::create() defaults to PhpBinarySerializer; pass one explicitly to make the choice visible. For a custom wire format, implement BinarySerializerInterface (see Custom Serializer):

use CrazyGoat\RabbitStream\Client\Connection;
use CrazyGoat\RabbitStream\Serializer\PhpBinarySerializer;

$connection = Connection::create(
    host: 'rabbitmq.example.com',
    port: 5552,
    user: 'app',
    password: 'secret',
    vhost: '/app',
    serializer: new PhpBinarySerializer()
);

$producer = $connection->createProducer('events');
$producer->send('{"type":"user_login","user_id":123}');

Pattern 3: Connection with PSR-3 Logger

Enable debug logging for troubleshooting:

use CrazyGoat\RabbitStream\Client\Connection;
use Monolog\Logger;
use Monolog\Handler\StreamHandler;

$logger = new Logger('rabbitmq');
$logger->pushHandler(new StreamHandler('php://stderr', Logger::DEBUG));

$connection = Connection::create(
    host: 'localhost',
    logger: $logger
);

// All protocol operations will be logged

Pattern 4: Connection Pool Pattern

Managing multiple connections for high availability:

use CrazyGoat\RabbitStream\Client\Connection;

class ConnectionPool
{
    /** @var Connection[] */
    private array $connections = [];
    private int $currentIndex = 0;
    
    public function __construct(array $hosts)
    {
        foreach ($hosts as $host) {
            $this->connections[] = Connection::create($host);
        }
    }
    
    public function getConnection(): Connection
    {
        $connection = $this->connections[$this->currentIndex];
        $this->currentIndex = ($this->currentIndex + 1) % count($this->connections);
        return $connection;
    }
    
    public function closeAll(): void
    {
        foreach ($this->connections as $conn) {
            $conn->close();
        }
    }
}

// Usage
$pool = new ConnectionPool(['rabbit1.local', 'rabbit2.local', 'rabbit3.local']);
$producer = $pool->getConnection()->createProducer('events');

Pattern 5: Stream Management Workflow

Creating and managing streams programmatically:

use CrazyGoat\RabbitStream\Client\Connection;

$connection = Connection::create('localhost');

// Ensure stream exists with proper configuration
$streamName = 'events';
if (!$connection->streamExists($streamName)) {
    $connection->createStream($streamName, [
        'max-length-bytes' => '1073741824',  // 1GB
        'max-age' => '7D',                   // 7 days
    ]);
    echo "Created stream: {$streamName}\n";
}

// Get stream statistics
$stats = $connection->getStreamStats($streamName);
echo "Stream has {$stats['messages']} messages\n";

// Cleanup when needed
// $connection->deleteStream($streamName);

Error Handling

Common Errors

use CrazyGoat\RabbitStream\Client\Connection;
use CrazyGoat\RabbitStream\Exception\ConnectionException;
use CrazyGoat\RabbitStream\Exception\AuthenticationException;
use CrazyGoat\RabbitStream\Exception\UnexpectedResponseException;

try {
    $connection = Connection::create(
        host: 'rabbitmq.example.com',
        user: 'invalid',
        password: 'wrong'
    );
} catch (AuthenticationException $e) {
    echo "Authentication failed: {$e->getMessage()}\n";
    exit(1);
} catch (ConnectionException $e) {
    echo "Cannot connect: {$e->getMessage()}\n";
    exit(1);
}

try {
    // Try to create a stream that already exists
    $connection->createStream('existing-stream');
} catch (UnexpectedResponseException $e) {
    echo "Server error: {$e->getMessage()}\n";
}

Connection Recovery

use CrazyGoat\RabbitStream\Client\Connection;

function getConnectionWithRetry(
    string $host,
    int $maxRetries = 3,
    int $retryDelayMs = 1000
): Connection {
    $lastException = null;
    
    for ($i = 0; $i < $maxRetries; $i++) {
        try {
            return Connection::create(host: $host);
        } catch (\Exception $e) {
            $lastException = $e;
            if ($i < $maxRetries - 1) {
                usleep($retryDelayMs * 1000);
            }
        }
    }
    
    throw $lastException;
}

$connection = getConnectionWithRetry('rabbitmq.example.com');

FilesExpand file tree

connection.md

Latest commit

History

connection.md

File metadata and controls

Connection API Reference

Class Overview

Connection::create()

Parameters

Return Value

Exceptions

Deliver frames need their own cap

frame_max negotiation only lowers the default

Outgoing frames over the limit fail fast

Connection Handshake Flow

Version negotiation

Examples

Stream Management Methods

createStream()

Parameters

Return Value

Exceptions

Example

deleteStream()

Parameters

Return Value

Exceptions

Example

streamExists()

Parameters

Return Value

Exceptions

Example

getStreamStats()

Parameters

Return Value

Exceptions

Example

getMetadata()

Parameters

Return Value

Exceptions

Example

Super Stream Methods

createSuperStream()

Parameters

Return Value

Exceptions

Example

deleteSuperStream()

Parameters

Return Value

Exceptions

Example

route()

Parameters

Return Value

Exceptions

Example

Notes

partitions()

Parameters

Return Value

Exceptions

Example

Notes

Offset Management Methods

queryOffset()

Parameters

Return Value

Exceptions

Example

storeOffset()

Parameters

Return Value

Exceptions

Notes

Example

Factory Methods

`frame_max` negotiation only lowers the default