PHP code example of crazy-goat / rabbit-stream

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

crazy-goat / rabbit-stream example snippets

use CrazyGoat\RabbitStream\Client\Connection;

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

$producer = $connection->createProducer('my-stream', name: 'my-producer');
$producer->send('hello world');
$producer->waitForConfirms(timeout: 5);
$producer->close();

$connection->close();

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

$connection = Connection::create(
    host: 'localhost',
    port: 5551, // TLS stream port
    tls: new TlsConfig(
        cafile: '/etc/ssl/ca.pem', // optional: custom CA bundle
        // localCert: '/etc/ssl/client.crt',   // optional: client certificate
        // localPk: '/etc/ssl/client.key',     // (e.g. for EXTERNAL SASL)
    ),
);

$connection = Connection::create(
    host: 'localhost',
    port: 5551,
    tls: new TlsConfig(verifyPeer: false, verifyPeerName: false),
);

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

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

$consumer = $connection->createConsumer('my-stream', offset: OffsetSpec::first());
while ($messages = $consumer->read(timeout: 5)) {
    foreach ($messages as $msg) {
        echo $msg->getBody() . "\n";
    }
}
$consumer->close();

$connection->close();

use CrazyGoat\RabbitStream\Client\Connection;
use CrazyGoat\RabbitStream\Client\ConfirmationStatus;

// Connect (handshake and authentication handled automatically)
$connection = Connection::create(
    host: '127.0.0.1',
    user: 'guest',
    password: 'guest'
);

// Create a producer for 'my-stream'
$producer = $connection->createProducer(
    stream: 'my-stream',
    onConfirm: function (ConfirmationStatus $status): void {
        if ($status->isConfirmed()) {
            echo "Message {$status->getPublishingId()} confirmed\n";
        }
    }
);

// Send a message
$producer->send("Hello, RabbitMQ Stream!");

// Drive the loop to receive confirmations (optional, blocking)
$connection->readLoop(maxFrames: 1);

// Close producer and connection
$producer->close();
$connection->close();

use CrazyGoat\RabbitStream\Client\AmqpMessageDecoder;
use CrazyGoat\RabbitStream\Client\OsirisChunkParser;

// ... subscribe to stream and receive Deliver response

$chunk = $deliverResponse->getChunk();
$entries = OsirisChunkParser::parse($chunk);

// Decode AMQP 1.0 messages into Message objects
$messages = AmqpMessageDecoder::decodeAll($entries);

foreach ($messages as $message) {
    echo "Offset: {$message->getOffset()}\n";
    echo "Body: {$message->getBody()}\n";
    echo "Content-Type: {$message->getContentType()}\n";
    echo "Message-ID: {$message->getMessageId()}\n";
}

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

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

// Named consumer with auto-commit every 100 messages
// The name is used to persist the offset on the server
$consumer = $connection->createConsumer(
    stream: 'my-stream',
    offset: OffsetSpec::first(),
    name: 'my-consumer-group',
    autoCommit: 100,
);

while ($messages = $consumer->read(timeout: 5)) {
    foreach ($messages as $msg) {
        echo $msg->getBody() . "\n";
    }
}
$consumer->close(); // stores final offset automatically

// On next startup, resume from the stored offset
$storedOffset = $connection->queryOffset('my-consumer-group', 'my-stream');
$consumer = $connection->createConsumer(
    stream: 'my-stream',
    offset: OffsetSpec::offset($storedOffset + 1),
    name: 'my-consumer-group',
    autoCommit: 100,
);

$connection->close();

use CrazyGoat\RabbitStream\StreamConnection;
use CrazyGoat\RabbitStream\Request\PeerPropertiesRequestV1;
...