Struct RedisBroker 

pub struct RedisBroker { /* private fields */ }

An unconnected Redis broker: the recorded topology and its options, no I/O performed yet.

Build it with standalone, cluster, sentinel, or from_pool. All four are synchronous and perform no I/O, so a Redis service composes with the synchronous #[ruststream::app] builder; the runtime dials once at startup through the consuming Broker::connect, which yields the ConnectedRedisBroker every subscription and publisher is reached from.

Examples

use ruststream_fred::stream::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[subscriber(RedisStream::new("orders").group("workers"))]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order");
    HandlerOutcome::ack()
}

// The broker is a plain value here; the runtime connects it at startup and shuts it down
// when the service stops.
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(handle);
        },
    )
}

Testing

A test hands TestApp the app main runs and addresses the broker by this type; the crate’s testing feature in [dev-dependencies] connects it to a Redis modelled in process.

use std::error::Error;

use ruststream::testing::TestApp;
use ruststream_fred::stream::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Outgoing, PartialEq, Serialize)]
struct Payment {
    id: u64,
    amount: u64,
}

#[subscriber(RedisStream::new("payments").group("workers"))]
async fn process(payment: &Payment) -> HandlerOutcome {
    if payment.amount == 0 {
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

/// The app `main` runs, and the one the tests hand the harness.
pub fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            b.include(process);
        })
}

pub async fn a_paid_payment_is_acknowledged() -> Result<(), Box<dyn Error>> {
    // The broker connects in process: nothing dials the address, and a publish is driven to
    // quiescence, so the assertion after it needs no waiting.
    let tb = TestApp::start(app()).await?;
    let payment = Payment { id: 1, amount: 100 };
    tb.broker::<RedisBroker>()
        .message(&payment)
        .to("payments")
        .publish()
        .await?;

    tb.broker::<RedisBroker>()
        .subscriber("payments")
        .assert_called_once()
        .with(&payment)
        .settled(HandlerOutcome::ack());
    Ok(())
}

Implementations

impl RedisBroker

pub fn standalone(url: impl Into<String>) -> Self

Creates a standalone-topology broker that connects to url when Broker::connect runs.

The broker’s Debug output shows the URL with its password, and any sentinelPassword parameter, as ***; the seed addresses of cluster and sentinel print the same way.

pub fn cluster(nodes: impl IntoIterator<Item = impl Into<String>>) -> Self

Creates a Redis Cluster broker from one or more host:port seed nodes.

Only one reachable node is needed; fred discovers the rest of the cluster on connect.

pub fn sentinel( service: impl Into<String>, sentinels: impl IntoIterator<Item = impl Into<String>>, ) -> Self

Creates a Sentinel-backed broker that tracks the primary named service, discovering it through the given sentinel host:port addresses.

pub fn from_pool(pool: Pool) -> Self

Wraps a fred pool the service built itself, for a setting the other constructors do not reach (a reconnection policy, performance tuning, a custom connection config).

The pool is built in the synchronous app builder and not connected there: Broker::connect dials it at startup, and an unreachable server fails startup as it does for an address. A pool the service has already connected is adopted as it is. The config and the reconnection policy the pool was built from are reused for the dedicated connections subscriptions read on.

Examples

use std::time::Duration;

use fred::clients::Pool;
use fred::types::config::{Config, ConnectionConfig, ReconnectPolicy, ServerConfig};
use ruststream_fred::stream::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[subscriber(RedisStream::new("orders").group("workers"))]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order = order.id, "order received");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let config = Config {
        server: ServerConfig::new_centralized("localhost", 6379),
        ..Config::default()
    };
    let connection = ConnectionConfig {
        connection_timeout: Duration::from_secs(2),
        ..ConnectionConfig::default()
    };
    // Reconnect forever, backing off from 100 ms up to 10 s between attempts.
    let policy = ReconnectPolicy::new_exponential(0, 100, 10_000, 2);
    let pool = Pool::new(config, None, Some(connection), Some(policy), 8)
        .expect("a pool of eight connections");

    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::from_pool(pool),
        |b| {
            b.include(handle);
        },
    )
}

pub const fn pool(self, size: usize) -> Self

Sets the connection-pool size: the connections publishes, settlements and handler commands share. Defaults to 4. Each subscription reads on a connection of its own besides.

pub fn default_group(self, group: impl Into<String>) -> Self

Sets a broker-wide default consumer group, enabling the bare-string #[subscriber("key")] form (Redis Streams always read through a group). Without it a bare-string subscription returns RedisError::InvalidOptions; name the group per subscription with RedisStream::group instead.

pub fn credentials( self, username: impl Into<String>, password: impl Into<String>, ) -> Self

Sets the ACL username and password used to authenticate on connect, applied on every topology (standalone, cluster, sentinel).

This maps onto fred’s Config.username / Config.password, so authentication works beyond the standalone redis://user:pass@host URL, which the bare cluster / sentinel seed lists cannot express. Credentials set here override any in a standalone URL.

For a password-only AUTH (the legacy requirepass, no ACL user) use password.

Examples

use ruststream_fred::stream::prelude::*;

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::cluster(["10.0.0.1:6379"]).credentials("worker", "s3cr3t"),
        |b| {
            b.include(handle);
        },
    )
}

pub fn password(self, password: impl Into<String>) -> Self

Sets a password-only AUTH (no ACL username; the legacy requirepass form), on every topology. Use credentials for an ACL user plus password.

Examples

use ruststream_fred::stream::prelude::*;

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::sentinel("mymaster", ["10.0.0.1:26379"]).password("s3cr3t"),
        |b| {
            b.include(handle);
        },
    )
}

pub fn tls(self, tls: impl Into<TlsConfig>) -> Self

Sets the TLS configuration used on connect, on every topology. Accepts a fred TlsConfig or anything convertible into one (for example a TlsConnector).

Available behind the tls-rustls, tls-rustls-ring, or tls-native-tls feature; a standalone broker can also enable TLS through a rediss:// / valkeys:// URL. The fred re-exports TlsConfig / TlsConnector provide default_rustls() / default_native_tls() shorthands for system-trust setups.

Examples

use std::error::Error;

use ruststream_fred::stream::prelude::*;

pub async fn serve() -> Result<(), Box<dyn Error>> {
    // System trust roots, no client certificate. The same connector works on every topology.
    let tls = TlsConnector::default_rustls()?;
    let broker = RedisBroker::cluster(["10.0.0.1:6379"])
        .credentials("worker", "s3cr3t")
        .tls(tls);
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(broker, |b| {
            b.include(handle);
        })
        .run()
        .await?;
    Ok(())
}

pub fn sentinel_credentials( self, username: impl Into<String>, password: impl Into<String>, ) -> Self

Sets distinct credentials for authenticating to the sentinel nodes, separate from the data-node credentials. Only meaningful on the sentinel topology.

Available behind the sentinel-auth feature.

Examples

use ruststream_fred::stream::prelude::*;

#[ruststream::app]
fn app() -> impl App {
    // The data nodes and the sentinels each check their own pair.
    let broker = RedisBroker::sentinel("mymaster", ["10.0.0.1:26379"])
        .credentials("worker", "data-pass")
        .sentinel_credentials("sentinel-user", "sentinel-pass");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        broker,
        |b| {
            b.include(handle);
        },
    )
}

pub fn sentinel_password(self, password: impl Into<String>) -> Self

Sets a password-only credential for authenticating to the sentinel nodes. Use sentinel_credentials for an ACL user plus password.

Available behind the sentinel-auth feature.

Examples

use ruststream_fred::stream::prelude::*;

#[ruststream::app]
fn app() -> impl App {
    let broker = RedisBroker::sentinel("mymaster", ["10.0.0.1:26379"])
        .password("data-pass")
        .sentinel_password("sentinel-pass");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        broker,
        |b| {
            b.include(handle);
        },
    )
}

pub fn credential_provider(self, provider: Arc<dyn CredentialProvider>) -> Self

Sets a dynamic credential provider that supplies (and can rotate) the username/password on each AUTH / HELLO, for IAM-style auth. Takes precedence over static credentials.

Available behind the credential-provider feature.

Examples

use std::sync::Arc;

use ruststream_fred::stream::prelude::*;

#[ruststream::app]
fn app() -> impl App {
    // `IamTokens` fetches a fresh token from the identity service on every `AUTH`.
    let broker = RedisBroker::standalone("redis://localhost:6379")
        .credential_provider(Arc::new(IamTokens));
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        broker,
        |b| {
            b.include(handle);
        },
    )
}

Trait Implementations

impl Broker for RedisBroker

async fn connect(self) -> Result<Self::Connected, Self::Error>

Opens (or adopts) the connection pool, consuming the unconnected form.

Errors

Returns RedisError::Connect when the recorded topology cannot be turned into a fred config or the pool cannot reach the server.

type Error = RedisError

The error type returned by broker-level operations.

type Connected = ConnectedRedisBroker

The connected form of this broker: the typed witness that connect succeeded.

fn bindable(self) -> Bindable<Self>

where Self: 'static,

Wraps the broker for publisher-token minting before registration: the returned Bindable hands out Bound tokens via its bind, and is itself what with_broker takes.

impl Clone for RedisBroker

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for RedisBroker

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl DescribeServer for RedisBroker

DescribeServer reports the host and port a client dials (the first seed for cluster and sentinel). Nothing else a configured address may carry reaches the description: a broker built from an already-connected pool describes no host at all.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl InProcess for RedisBroker

Available on crate feature testing only.

The in-process mode: the production connected broker over an in-process Redis, the transport a test runs the service’s own app against.

Every setting of this broker is read the way Broker::connect reads it: the topology’s address is parsed (a URL connect refuses is refused here), the pool keeps its size, the default group and the cluster flag carry over, and a cluster topology refuses what spans hash slots. What the model answers is described in the crate’s testing overview.

async fn connect_in_process(self) -> Result<Self::Connected, Self::Error>

Connects to the in-process transport, consuming the configuration.