RedisBroker

RedisBroker connects a service to Redis or Valkey; standalone only records the URL, and the runtime connects it at startup.

standalone takes the URL of one server, such as redis://localhost:6379. The URL may carry a user and password (redis://user:pass@host:6379) and a database number (/2). A rediss:// or valkeys:// URL switches TLS on. The constructor does no I/O, so the whole service fits the synchronous #[ruststream::app] builder. A cluster, a Sentinel deployment and a pool the service built have constructors of their own. The settings that follow are builder steps too: the pool size, the default consumer group and the credentials.

At startup the runtime connects the broker and receives a ConnectedRedisBroker. The connect opens the pool and waits for it, so an unreachable server fails startup instead of the first publish. Every subscription and every publisher of the service opens against that connected form. At shutdown the runtime turns it into a ClosedRedisBroker.

Every connection the broker opens reconnects after the server drops it, with no limit on the attempts. The first attempt follows after 100 ms, and each next one twice as late, at most 10 s apart. A command issued while the connection is down waits for it. A Pub/Sub subscription subscribes again. A consumer group the server lost is created again and reads the stream from its start.

Example

The service reads the stream orders in the group workers. cargo run --example redis_broker --features testing -- run starts it against a server on localhost:6379.

Macros
Manual

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

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(take);
        },
    )
}

examples/manual/redis_broker.rs
struct Take;

impl Handle<Order> for Take {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order taken");
        ready(Ok(()))
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(subscriber(RedisStream::new("orders").group("workers"), Take).build());
        },
    )
}

Example​

Example