Struct RedisPubSubPattern 

pub struct RedisPubSubPattern { /* private fields */ }

Describes one Pub/Sub subscription on a glob pattern (PSUBSCRIBE), against a ConnectedRedisBroker.

Patterns are classic-only: sharded Pub/Sub routes by slot and has no PSUBSCRIBE, so this descriptor carries no mode to get wrong.

Examples

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

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

#[subscriber(RedisPubSubPattern::new("events.*"))]
async fn on_any_event(event: &Event, Ctx(channel): Ctx<keys::Channel>) -> HandlerOutcome {
    tracing::info!(event.kind, channel, "event");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            // A glob is no channel a `PUBLISH` can name, so the mount site says where a
            // retry copy goes.
            b.include(on_any_event)
                .out_retry(Publish::default())
                .to("events.retry");
        },
    )
}

Sharded Pub/Sub has no pattern subscription, so a pattern takes no mode:

use ruststream_fred::{PubSubMode, RedisPubSubPattern};

let events = RedisPubSubPattern::new("events.*").mode(PubSubMode::Sharded);

Implementations

impl RedisPubSubPattern

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

A subscription on every channel matching the glob pattern.

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(RedisPubSubPattern::new("events.*").buffer(nonzero!(10_000)))]
async fn on_event(event: &Event) -> HandlerOutcome {
    tracing::info!(event.kind, "event");
    HandlerOutcome::ack()
}

pub fn pattern(&self) -> &str

The glob this subscription matches channels against.

Trait Implementations

impl Clone for RedisPubSubPattern

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 RedisPubSubPattern

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

Formats the value using the given formatter.

impl IntoSource for RedisPubSubPattern

type Source = RedisPubSubPattern

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl SubscriptionSource<ConnectedRedisBroker> for RedisPubSubPattern

A pattern subscription reads every channel the glob matches, and a glob is not a channel a PUBLISH can name, so the mount site names where a retry copy goes: .out_retry(policy).to(..) for a fixed channel, or a transform that reads the delivery’s own channel.

type Subscriber = RedisPubSubSubscriber

The subscriber type this source opens.

type Copies = NamedCopies

Who publishes the copies this subscription’s retries are made of, and who names where they go: AddressedCopies, NamedCopies or BrokerMoves.

fn name(&self) -> &str

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

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(self, declaration: &RetryDeclaration) -> Self

where Self: Sized,

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

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.