Crate ruststream_fred 

Redis and Valkey broker for the RustStream messaging framework, backed by the fred client.

Three Redis transports sit behind one broker. Redis Streams is the durable one, a log read through consumer groups with acknowledgement, redelivery and crash recovery. A Redis list is a work queue whose every entry goes to one consumer. Pub/Sub delivers to the subscribers connected when a message is published. Four topologies serve all three: a standalone server, a cluster, a sentinel set, and a fred Pool a service built for itself.

A service

A handler is an async fn over the decoded payload, mounted on a broker inside the application object, and the attribute writes the main:

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

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

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

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

cargo run -- run starts it. RedisBroker::standalone only records the address; the runtime connects the broker at startup and shuts it down when the service stops. A bare key in #[subscriber] reads that stream through the broker’s default consumer group.

Where things are

The guides take a Redis service from installation to production, in this order: getting started, choosing a transport, consumer groups and crash recovery, work queues on lists, retries, delays and dead letters, fan-out with Pub/Sub, side effects with the acknowledgement, transactions, replay and seeking, Redis Cluster, running in production, testing and interoperating with other clients. The architecture pages show how Redis delivers messages and where the crate sits between fred and the runtime, and the benchmarks what it costs.

Each topic below links to its component page on the site, which shows it in a service; the item’s own documentation keeps the detail.

Broker

• RedisBroker: RedisBroker, connected into ConnectedRedisBroker and closed into ClosedRedisBroker.
• Cluster and Sentinel: RedisBroker::cluster and RedisBroker::sentinel.
• from_pool: RedisBroker::from_pool wraps a fred Pool the service built.
• Pool: RedisBroker::pool, the connections publishes and settlements share.
• default_group: RedisBroker::default_group, the group a bare key reads through.
• Authentication: RedisBroker::credentials and RedisBroker::password.
• TLS: RedisBroker::tls with the tls-rustls, tls-rustls-ring or tls-native-tls feature.
• sentinel_credentials: RedisBroker::sentinel_credentials and RedisBroker::sentinel_password (feature sentinel-auth).
• credential_provider: RedisBroker::credential_provider, rotating credentials (feature credential-provider).
• Shutdown: ClosedRedisBroker, and RedisError::ShutDown for a handle used afterwards.
• Errors: RedisError.
• Prelude: stream::prelude, list::prelude, pubsub::prelude and the mixed-form prelude.

Subscriptions

• Subscriptions: the descriptors, what each settles and where its retry copy goes.
• block: RedisSubscribeExt::block, how long one read waits on the server.
• Batches: a slice handler, a native COUNT on a stream.
• Framing: the binary frame or the codec envelope of a list entry and a channel message.

Streams

• RedisStream: RedisStream, a stream read through a consumer group.
• reclaim: RedisStream::reclaim, the entries a dead consumer never acked.
• claiming: RedisStream::claiming, pending and fresh entries in one read (Redis 8.4).
• StreamStart: StreamStart, where a new group starts.

Positions

• RedisGroupPosition: RedisGroupPosition, where a consumer group resumes.
• start_at: a group moved to a position on every startup.
• Seek: RedisGroupSeeker, the group moved from a handler.
• EntryId: EntryId, a stream entry’s id.

Lists

• Lists: RedisList, a work queue.
• reliable: RedisList::reliable, at-least-once through a processing list.
• Recovery: RedisList::recovery_zset, the watchdog for a dead consumer’s entries.
• processing: RedisList::processing, and the hash tag a cluster needs.

Pub/Sub

• PubSub: RedisPubSub, one channel.
• RedisPubSubPattern: RedisPubSubPattern, every channel a glob matches.
• PubSubMode: PubSubMode, classic or sharded delivery.
• buffer: RedisPubSub::buffer, what a busy subscription holds.

Acknowledgement

• Acknowledgement: what ack and nack do on each form.
• retry_after: a delayed retry through the runtime’s copy.
• DelayedRetry: DelayedRetry, a delay queue that survives a restart.
• dead_letter: the retry cap and where a spent message goes.
• Counts: DELIVERY_COUNT_HEADER and IDLE_MS_HEADER from the claiming modes.
• out_retry: where a retry copy goes, named at the mount site.

Context

• StreamContext: context::StreamContext and the keys in context::keys.
• ConsumerGroup: context::keys::ConsumerGroup.
• FredPool: context::keys::FredPool, the connection pool in a handler.
• StreamBatchContext: context::StreamBatchContext, a batch handler’s context.
• PubSubContext: context::PubSubContext, the channel a message came on.

Pipelining

• pipeline: pipeline::RedisPipelineSteps, a window that settles a subscription’s reads together with its handlers’ commands.
• atomic: one MULTI / EXEC per delivery.
• Commands: pipeline::RedisPipeline, the commands a handler queues.
• bind: pipeline::Bindable, a slot’s publishes in the delivery’s round.
• InRound: pipeline::InRound, the reply in the delivery’s round.
• Flush: when a window leaves, and what each outcome does to it.

Publishing

• RedisPublish: RedisPublish, XADD to a stream, and its live RedisPublisher.
• RedisDefaultPublish: RedisDefaultPublish, each name written the way the service reads it.
• RedisPubSubPublish: RedisPubSubPublish, PUBLISH or SPUBLISH.
• RedisListPublish: RedisListPublish, LPUSH with an optional key expiry.
• partition_key: RedisPublishSteps::partition_key and RedisPublishOptions.

Transactions

• Transactions: RedisTransactionalPublish, one MULTI / EXEC block of publishes on a stream publisher; a cluster refuses it at startup.
• transaction: RedisTransaction, a transaction that owns its buffer.
• owned_transaction: the same over values.

Tooling

• AsyncAPI: the x-ruststream-redis extension of the generated document, with its server, channels and operations (feature asyncapi).
• Testing: the in-process Redis TestApp connects RedisBroker to (feature testing), and Live runs the same test against a server.

Cargo features

All off by default: the crate ships the three transports and the broker with no feature on. testing adds the in-process mode TestApp::start connects the broker through, and asyncapi the Redis extension of the generated document. tls-rustls, tls-rustls-ring and tls-native-tls map onto fred’s TLS backends. sentinel-auth adds credentials of their own for the sentinels, and credential-provider rotating credentials through a callback.

Re-exports

pub use list::RedisList;

pub use list::RedisListMessage;

pub use list::RedisListPublish;

pub use list::RedisListPublisher;

pub use list::RedisListSubscriber;

pub use pipeline::RedisListAtomic;

pub use pipeline::RedisListPipeline;

pub use pipeline::RedisPubSubAtomic;

pub use pipeline::RedisPubSubPatternAtomic;

pub use pipeline::RedisPubSubPatternPipeline;

pub use pipeline::RedisPubSubPipeline;

pub use pipeline::RedisStreamAtomic;

pub use pipeline::RedisStreamPipeline;

pub use pubsub::PubSubMode;

pub use pubsub::RedisPubSub;

pub use pubsub::RedisPubSubMessage;

pub use pubsub::RedisPubSubPattern;

pub use pubsub::RedisPubSubPublish;

pub use pubsub::RedisPubSubPublisher;

pub use pubsub::RedisPubSubSubscriber;

pub use stream::RedisStream;

pub use stream::StreamStart;

Modules

context

Per-delivery and per-batch context fields exposed to handlers.

list

Redis lists: a work queue whose every entry goes to one consumer.

pipeline

Consumer-side pipelining: a subscription settles in a window, and a handler queues its own Redis commands into the delivery’s segment of that window.

prelude

The core prelude plus everything a service mixing all three Redis forms writes.

pubsub

Redis Pub/Sub: delivery to the subscribers connected when a message is published.

stream

Redis Streams: a stream read through a consumer group, the durable form.

Structs

ClosedRedisBroker

The terminal witness returned by shutting down a ConnectedRedisBroker.

ConnectedRedisBroker

The typed witness that Broker::connect succeeded: it owns the live fred pool.

EntryId

A Redis Streams entry id: the <milliseconds>-<sequence> pair the server assigns to every entry.

RedisBroker

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

RedisDefaultPublish

The broker’s default policy, what a registration publishes through when its mount site names none: a reply, a retry copy, a dead-letter move.

RedisDefaultPublisher

The live form of RedisDefaultPublish: writes each name the way this connection’s subscriptions read it. Cheap to clone.

RedisGroupSeeker

Moves a consumer group’s cursor.

RedisMessage

A Redis Streams delivery, read from a consumer group via XREADGROUP or XAUTOCLAIM.

RedisPublish

The declaration half of the stream publisher: pure policy, constructible anywhere.

RedisPublishOptions

The per-message settings of every publisher and transaction in this crate.

RedisPublisher

The live stream publisher: RedisPublish paired with a connection. Cheap to clone.

RedisSubscriber

A Redis Streams subscription bound to a consumer group.

RedisTransaction

An owned Redis transaction, opened by transaction on a RedisPublisher.

RedisTransactionalPublish

The stream publisher’s policy for a mount that publishes in transactions: it pairs into the same RedisPublisher as RedisPublish, and only where the topology can run one.

TlsConfig

TLS configuration for a client.

Enums

DelayedRetry

How a subscription should handle retry_after / nack_after delays.

RedisError

Errors surfaced by the Redis broker implementation.

RedisGroupPosition

Where a consumer group’s cursor sits, for RedisGroupSeeker::seek and the start_at(..) clause of #[subscriber].

TlsConnector

An enum for interacting with various TLS libraries and interfaces.

Constants

DELIVERY_COUNT_HEADER

Header carrying the native Redis Streams delivery count, so a handler can branch on how many attempts a message has already had.

IDLE_MS_HEADER

Header carrying how long (milliseconds) the delivery had been pending before this read. Zero on an entry a claiming subscription read off the tail.

PARTITION_KEY_HEADER

The well-known header key for per-message routing / partitioning.

Traits

CredentialProvider

A trait that can be used to override the credentials used in each AUTH or HELLO command.

RedisPublishSteps

The steps this crate adds to the publish builder.

RedisSubscribeExt

The Redis subscription options a mount site names, chained after the core’s settings steps.