Struct RedisPubSub 

pub struct RedisPubSub { /* private fields */ }

Describes one Pub/Sub subscription on a single channel, against a ConnectedRedisBroker.

A glob is RedisPubSubPattern instead.

Examples

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

#[derive(Deserialize)]
struct Event {
    kind: String,
}

// Classic: `SUBSCRIBE`, broadcast to every node of a cluster.
#[subscriber(RedisPubSub::new("events"))]
async fn on_event(event: &Event) -> HandlerOutcome {
    tracing::info!(event.kind, "event");
    HandlerOutcome::ack()
}

// Sharded: `SSUBSCRIBE`, served by the node that owns the channel's slot.
#[subscriber(RedisPubSub::new("metrics").mode(PubSubMode::Sharded))]
async fn on_metric(event: &Event) -> HandlerOutcome {
    tracing::info!(event.kind, "metric");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
        RedisBroker::cluster(["10.0.0.1:6379"]),
        |b| {
            b.include(on_event);
            b.include(on_metric);
        },
    )
}

Implementations

impl RedisPubSub

pub fn new(channel: impl Into<String>) -> Self

A subscription on the exact channel.

pub const fn mode(self, mode: PubSubMode) -> Self

Sets the delivery mode. Defaults to PubSubMode::Classic.

pub fn codec(self, codec: impl Codec + 'static) -> Self

Decodes the header/payload envelope with codec (must match the publisher). Without it the default binary framing is used. Either way the payload arrives as it was published.

pub const fn buffer(self, messages: NonZeroUsize) -> Self

Sets how many messages the subscription holds that its handler has not taken yet.

Redis pushes a Pub/Sub message the moment it is published, and the subscription keeps it until the handler asks for the next one. The buffer bounds how many may wait. Once it is full, each new message drops the oldest, and the subscription logs a warning with the channel and the count it lost. The default is fred’s own (PerformanceConfig::broadcast_channel_capacity, 32 messages in fred 10.1).

Raise it for a handler that falls behind bursts: set it to the largest burst the handler must absorb. The room is reserved when the subscription opens, about 600 bytes per message of buffer whatever the payloads weigh, because fred sizes all eight of a client’s notification channels by it.

Examples

use ruststream::nonzero;
use ruststream_fred::pubsub::prelude::*;

// A handler that falls behind a burst of up to ten thousand events loses none of them.
#[subscriber(RedisPubSub::new("events").buffer(nonzero!(10_000)))]
async fn on_event(event: &Event) -> HandlerOutcome {
    tracing::info!(event.kind, "event");
    HandlerOutcome::ack()
}

pub fn channel(&self) -> &str

The channel this subscription listens on.

Trait Implementations

impl Clone for RedisPubSub

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 RedisPubSub

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

Formats the value using the given formatter.

impl IntoSource for RedisPubSub

type Source = RedisPubSub

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl RedeliveryAddressed<ConnectedRedisBroker> for RedisPubSub

The channel itself.

The answer is about the address, not about the command: reaching a PubSubMode::Sharded subscription still takes a publisher in the same mode, which is the policy the mount site names with out_retry.

fn redelivery_address( &self, _connected: &ConnectedRedisBroker, ) -> impl Future<Output = Result<RedeliveryAddress, RedisError>>

Where a publish reaches this subscription again.

impl SubscriptionSource<ConnectedRedisBroker> for RedisPubSub

type Copies = AddressedCopies

Pub/Sub cannot acknowledge, so every retry is a copy this process publishes, and the channel is where it goes.

type Subscriber = RedisPubSubSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

async fn subscribe( self, connected: &ConnectedRedisBroker, ) -> Result<Self::Subscriber, RedisError>

Opens the subscription against the connected broker. Called once at startup.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.