Brokers

Handlers, routers, codecs, and middleware are broker-agnostic. You can move a service to another broker by changing one line at with_broker.

The framework ships a full in-memory broker for queues that stay inside a single application. Brokers backed by an external service are separate crates you add as a dependency.

Each broker crate has its own documentation site, linked in the Docs column and from the Brokers menu.

Broker	Crate	Transport	Docs
Memory	ruststream (feature memory)	in-process queue, no external service	this site
NATS	ruststream-nats	Core NATS and JetStream	powersemmi.github.io/ruststream-nats
Redis	ruststream-fred	Redis Streams (standalone, cluster, sentinel)	powersemmi.github.io/ruststream-fred
RabbitMQ	ruststream-lapin	AMQP 0.9.1 (queues, exchanges, publisher confirms, direct reply-to)	powersemmi.github.io/ruststream-lapin
Kafka	ruststream-rdkafka	Apache Kafka (consumer groups, tracked commits, transactions, exactly-once pipelines)	powersemmi.github.io/ruststream-rdkafka
AMQP 1.0	ruststream-amqp	ActiveMQ Artemis, RabbitMQ 4.x, Azure Service Bus, and the wider AMQP 1.0 family (request/reply, transactions)	powersemmi.github.io/ruststream-amqp
Google Cloud Pub/Sub	ruststream-gcp-pubsub	Pub/Sub over the official client (ordering keys, exactly-once acknowledgement, dead-letter policies)	powersemmi.github.io/ruststream-gcp-pubsub
AWS SQS / SNS	ruststream-sqs-sns	SQS queues with SNS fan-out (FIFO groups, visibility management, native deferred retry)	powersemmi.github.io/ruststream-sqs-sns
Apache Pulsar	ruststream-pulsar	Pulsar topics and patterns (subscription modes, dead-letter policies, repositioning)	powersemmi.github.io/ruststream-pulsar
MQTT 5	ruststream-rumqttc	MQTT v5 (QoS levels, shared groups, retained messages)	powersemmi.github.io/ruststream-rumqttc
ZeroMQ	ruststream-zeromq	Brokerless PUSH/PULL, PUB/SUB, and DEALER/ROUTER request/reply over TCP and IPC	powersemmi.github.io/ruststream-zeromq
Stream files / stdio	ruststream-sea-file	Persistent replayable stream files and shell pipelines; zero infrastructure, full repositioning	powersemmi.github.io/ruststream-sea-file
AWS Kinesis	ruststream-kinesis	Kinesis data streams (shard leasing, checkpointing, repositioning)	powersemmi.github.io/ruststream-kinesis

To implement a broker for another transport, see Broker authors.

Switching brokers

Every broker is constructed synchronously. The runtime connects it when the application starts. The examples below differ only in the line that constructs the broker.

MemoryNATSRedisRabbitMQKafka

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, RustStream};

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include_router(routes::orders()))
}

use ruststream::runtime::{AppInfo, RustStream};
use ruststream_nats::NatsBroker;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(NatsBroker::new("nats://localhost:4222"), |b| {
            b.include_router(routes::orders())
        })
}

use ruststream::runtime::{AppInfo, RustStream};
use ruststream_fred::RedisBroker;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            b.include_router(routes::orders())
        })
}

use ruststream::runtime::{AppInfo, RustStream};
use ruststream_lapin::LapinBroker;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(LapinBroker::new("amqp://localhost:5672"), |b| {
            b.include_router(routes::orders())
        })
}

use ruststream::runtime::{AppInfo, RustStream};
use ruststream_rdkafka::KafkaBroker;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(
            KafkaBroker::new(["localhost:9092"]).default_group("orders"),
            |b| {
                b.include_router(routes::orders())
            },
        )
}

Each broker crate documents its own connection options. When a subscription needs broker-specific options (consumer groups, durable names), you can write that broker's descriptor in the #[subscriber(..)] attribute; see broker-specific descriptors.