Working with Stream Queues

1. Declaring Stream Queue

Property	Value	Description
type	`stream`	Declares the queue as a stream queue
x-max-length-bytes	`1073741824`	Maximum size of the stream in bytes (default: 1GB)
x-message-ttl	`3600000`	Time-to-live for messages in milliseconds (default: no TTL)
x-overflow	`reject-publish`	Behavior when stream reaches max size (`reject.publish` or `drop.head`)
x-stream-max-segment-size-bytes	`52428800`	Maximum size of a stream segment in bytes (default: 50MB)

Example: Declaring a Stream Queue with RabbitMQ Management UI

{
  "name": "my-stream-queue",
  "type": "stream",
  "arguments": {
    "x-max-length-bytes": 1073741824,
    "x-message-ttl": 3600000,
    "x-overflow": "reject.publish"
  }
}

Example: Declaring a Stream Queue with RabbitMQ CLI

rabbitmqadmin declare queue name=my-stream-queue type=stream arguments='{"x-max-length-bytes":1073741824,"x-message-ttl":3600000,"x-overflow":"reject.publish"}'

2. Understanding Log-Based Storage

Aspect	Description	Details
Storage Model	How messages are stored	Messages are stored in a log-like structure, where each message is appended to the end of the log. The log is segmented into files for efficient handling and management.
Message Persistence	How messages are persisted to disk	Messages are written sequentially to log segments, ensuring durability. Each segment can be configured with a size limit and retention policy.
Log Segments	Structure of log files	Logs are divided into segments, each with a fixed size (e.g., 50MB). Once a segment is full, a new one is created. Old segments can be deleted based on retention policies.
Indexing	How messages are indexed for fast retrieval	Each segment has an index file that maps message offsets to their physical locations in the log, enabling fast random access.
Replication	How logs are replicated across nodes	Log segments can be replicated to multiple nodes for high availability and fault tolerance. Replication is handled at the segment level.

Example: Log Segment Configuration

{"log": {"segment_size_mb": 50, "retention_days": 7}}

Note: Log-based storage in RabbitMQ Stream Queues is designed for high throughput and low latency, making it ideal for scenarios where message order and durability are critical.

3. Setting Retention Policy

Policy	Description	Configuration
max-age	Retention period based on message age	`"max-age": "7d"`
max-size	Retention based on total message size	`"max-size": "10GB"`
max-messages	Retention based on number of messages	`"max-messages": "1000000"`

Example: Configure Stream Queue Retention

{
  "type": "stream",
  "arguments": {
    "max-age": "7d",
    "max-size": "10GB"
  }
}

4. Using Offset Tracking

Offset tracking in RabbitMQ Stream Queues allows consumers to track their position within the stream, enabling them to resume consumption from where they left off in case of interruptions.

Property	Description	Values
offset	Represents the position in the stream	`0`, `1`, ..., `stream_length`
offset_type	Type of offset tracking to use	`first`, `last`, `next`, `specific`

Example: Consuming from a specific offset

const amqp = require('amqplib');

async function consumeFromOffset() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queueName = 'stream-queue';
  const offset = 100;

  // Set the consumer to start from a specific offset
  await channel.consume(queueName, (msg) => {
    console.log('Message:', msg.content.toString());
    channel.ack(msg);
  }, { consumerTag: 'consumer-1', noAck: false });

  // Manually set the offset to resume from
  await channel.setConsumerOffset('consumer-1', offset);

  console.log(`Consuming from offset ${offset}`);
}

consumeFromOffset().catch(console.error);

Note: Offset tracking is particularly useful in scenarios where consumers need to handle failures gracefully and resume from the last processed message.

5. Implementing Stream Consumers

Stream consumers in RabbitMQ are used to read messages from stream queues. They can be configured with various parameters to control how messages are consumed and processed.

Parameter	Description	Default Value	Notes
consumer_tag	Unique identifier for the consumer	Generated by RabbitMQ	Must be unique within the connection
ack_mode	Controls how messages are acknowledged	`manual`	Can be `manual` or `auto`
stream_offset	Starting offset for message consumption	`first`	Can be `first`, `last`, `next`, or a specific offset

Example: Basic Stream Consumer Implementation

const amqp = require('amqplib');

async function consumeStreamMessages() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queueName = 'my-stream-queue';
  const consumerTag = 'my-consumer';

  await channel.assertQueue(queueName, { durable: true, x-queue-type: 'stream' });

  await channel.consume(queueName, (msg) => {
    console.log('Received message:', msg.content.toString());
    channel.ack(msg);
  }, { consumerTag });

  console.log('Stream consumer started');
}

consumeStreamMessages().catch(console.error);

Example: Stream Consumer with Manual Offset

const amqp = require('amqplib');

async function consumeStreamWithOffset() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queueName = 'my-stream-queue';
  const consumerTag = 'offset-consumer';

  await channel.assertQueue(queueName, { durable: true, x-queue-type: 'stream' });

  await channel.consume(queueName, (msg) => {
    console.log('Received message:', msg.content.toString());
    channel.ack(msg);
  }, { 
    consumerTag,
    stream_offset: 100
  });

  console.log('Stream consumer started from offset 100');
}

consumeStreamWithOffset().catch(console.error);

Note: Stream consumers can be configured to use either automatic or manual acknowledgment modes. Manual acknowledgment is recommended for better control over message processing.

Warning: When using manual acknowledgment, it's crucial to ensure that messages are acknowledged or rejected to prevent message loss.

6. Publishing to Streams

Property	Description	Values
routing_key	Used to route messages to the correct stream	`string`
message_id	Unique identifier for the message	`string`
timestamp	Timestamp when the message was published	`integer`
headers	Additional metadata for the message	`map`

Example: Publishing to a Stream with AMQP

const amqp = require('amqplib');

async function publishToStream() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const exchange = 'stream-exchange';
  const routingKey = 'stream-queue';

  await channel.assertExchange(exchange, 'direct', { durable: true });
  await channel.publish(exchange, routingKey, Buffer.from('Hello, Stream!'), {
    headers: { source: 'node-app' },
    messageId: 'msg-123',
    timestamp: Date.now()
  });

  await channel.close();
  await connection.close();
}

publishToStream().catch(console.error);

Note: Ensure that the exchange type is set to direct for routing messages to stream queues.

7. Using Stream Filtering

Stream filtering in RabbitMQ allows you to filter messages as they are consumed from a stream queue. This feature is particularly useful for reducing the amount of data that needs to be processed by consumers.

Filter Type	Description	Usage
message-id	Filter by message ID	`stream.filter.message-id = "msg123"`
content-type	Filter by content type	`stream.filter.content-type = "application/json"`
custom-header	Filter by custom header values	`stream.filter.custom-header.my-header = "value"`

Example: Filtering Messages by Content Type

const amqp = require('amqplib');

async function filterByContentType() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queueName = 'stream-queue';
  const filter = { 'content-type': 'application/json' };

  // Consume messages with a filter
  channel.consume(queueName, (msg) => {
    if (msg && msg.content) {
      console.log('Received message:', msg.content.toString());
    }
  }, { noAck: true });

  // Apply filter to stream queue
  await channel.addStreamFilter(queueName, filter);
}

filterByContentType().catch(console.error);

Note: Stream filtering is supported only for stream queues and requires RabbitMQ 3.12 or later.

8. Understanding Replication

Note: Stream queue replication is handled automatically by RabbitMQ, ensuring high availability and fault tolerance.

Replication Factor	Description	Notes
replication.factor	Number of nodes that must store a copy of the stream	Default is `1`, meaning only the leader node stores the stream. Set to `2` or higher for redundancy.
min.healthy.nodes	Minimum number of nodes required to keep the stream operational	If less than this number are available, the stream becomes unavailable until recovery.

Example: Creating a Stream with Replication

{"type": "stream", "name": "my-stream", "arguments": {"replication.factor": 2, "min.healthy.nodes": 1}}

Replication Behavior

Streams are replicated across multiple nodes based on replication.factor.
Each replica is stored on a separate node to ensure fault tolerance.
If a node fails, the stream continues operating from other replicas.

9. Using Streams for Event Sourcing

Example: Basic Stream Configuration for Event Sourcing

const streamConfig = {
  "max-age": "7d",
  "max-size": "100GB",
  "stream-max-segment-size": "50MB"
};

// Create stream with retention policy
await rabbitmq.createStream("user-events", streamConfig);

Property	Value	Description
max-age	`7d`	Maximum age of messages before deletion
max-size	`100GB`	Total size limit for stream storage
stream-max-segment-size	`50MB`	Maximum size of individual stream segments

Note: Event streams are ideal for capturing business events with strong durability guarantees.

Event Sourcing vs Traditional Messaging

Event Sourcing: Stores all changes as a sequence of events
Traditional Messaging: Sends messages to queues for processing

Event Sourcing Workflow

Application emits domain events
Events are written to stream in order
Consumers read from stream for processing
Stream retention policies manage data lifecycle

10. Comparing Streams vs Queues

Feature	Stream	Queue
Message Order	Guaranteed order within a partition	Guaranteed order per consumer
Storage	Stored on disk with persistence	Stored in memory and disk
Scalability	Horizontal scaling via partitions	Vertical scaling only
Durability	High durability with replication	Configurable durability
Consumption Model	Consumer groups with offset management	Single consumer or fan-out
Use Case	High throughput, low latency, large datasets	Reliable message delivery, simple workflows

When to Use Streams vs Queues

Streams: Ideal for real-time analytics, log aggregation, and scenarios requiring high throughput with guaranteed order within partitions.
Queues: Suitable for traditional message queuing, where reliability and simple consumer models are prioritized.

11. Configuring Stream Performance

Configuration Parameter	Description	Default Value	Notes
stream.max.segment.size	Maximum size of a stream segment file	`50MB`	Adjust based on available disk space and performance needs
stream.max.segments	Maximum number of segments per stream	`100`	Controls how many segment files are kept before rotation
stream.flush.interval	Interval between flushes to disk	`1000ms`	Lower values increase durability but may impact performance
stream.flush.bytes	Number of bytes to accumulate before flushing	`10MB`	Higher values can improve throughput at the cost of latency
stream.max.inflight.messages	Maximum number of unacknowledged messages per consumer	`1000`	Adjust to control memory usage and consumer flow control

Example: Stream Performance Configuration

{stream: {max.segment.size: "50MB",max.segments: 100,flush.interval: "1000ms",flush.bytes: "10MB",max.inflight.messages: 1000}}

Note: These settings are typically configured in the RabbitMQ configuration file or via management UI.

12. Monitoring Stream Metrics

Example: Collecting Stream Metrics via Management Plugin

curl -u guest:guest http://localhost:15672/api/queues/%2f/stream-queue-name

Metric	Description	Unit	Collection Method
stream.messages	Total number of messages in the stream	count	Management API /queues/
stream.bytes	Total size of messages in the stream	bytes	Management API /queues/
stream.consumer.count	Number of active consumers	count	Management API /queues/
stream.backlog	Number of messages not yet consumed	count	Management API /queues/
stream.lag	Time lag between producer and consumer	seconds	Management API /queues/

Note: Stream metrics are available via the RabbitMQ Management Plugin and can be queried using the HTTP API.

Warning: High stream lag may indicate consumer performance issues or backpressure.

Key Metrics for Stream Monitoring

stream.messages: Indicates stream size and potential storage pressure.
stream.consumer.count: Helps identify consumer scaling needs.
stream.lag: Critical for latency-sensitive applications.