Trait SubscriberSettings 

pub trait SubscriberSettings: Declared {
    // Provided methods
    fn name(
        self,
        name: impl Into<Cow<'static, str>>,
    ) -> <Self::Settings as NameStep>::Out
       where Self::Settings: NameStep { ... }
    fn workers(
        self,
        count: NonZero<usize>,
    ) -> <Self::Settings as WorkersStep>::Out
       where Self::Settings: WorkersStep { ... }
    fn workers_by_key(
        self,
        count: NonZero<usize>,
    ) -> <Self::Settings as WorkersStep>::Out
       where Self::Settings: WorkersStep { ... }
    fn threads(
        self,
        count: NonZero<usize>,
    ) -> <Self::Settings as WorkersStep>::Out
       where Self::Settings: WorkersStep { ... }
    fn threads_by_key(
        self,
        count: NonZero<usize>,
    ) -> <Self::Settings as WorkersStep>::Out
       where Self::Settings: WorkersStep { ... }
    fn on_failure(
        self,
        policies: FailurePolicies,
    ) -> <Self::Settings as FailureStep>::Out
       where Self::Settings: FailureStep { ... }
    fn start_at<P>(self, position: P) -> <Self::Settings as StartAtStep<P>>::Out
       where Self::Settings: StartAtStep<P> { ... }
    fn batch(self, size: NonZero<usize>) -> <Self::Settings as BatchStep>::Out
       where Self::Settings: BatchStep { ... }
    fn map_source<F>(self, f: F) -> <Self::Settings as MapSourceStep<F>>::Out
       where Self::Settings: MapSourceStep<F> { ... }
}

The declarative settings, chainable on a #[subscriber] definition and on the builder alike.

Every method is available exactly while its setting is open: the attribute fixes what it names, and the corresponding step trait is implemented only for the open state, so a second naming fails to compile with a message of its own.

The order in a chain follows from what each step does to the source: name comes first because it constructs the source, and a broker’s own settings then transform it through map_source - which is also where a broker’s batching options live, after the core’s own batch has named the batch size.

Examples

use ruststream::memory::prelude::*;
use serde::Deserialize;

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

/// Everything is left to the mount site: the name, the pool, the failure policy.
#[subscriber]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "audited");
    HandlerOutcome::ack()
}

/// One instance per shard, each reading its own subscription.
fn app(shard: u8) -> RustStream {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(
            audit
                .name(format!("orders-{shard}"))
                .workers(nonzero!(4))
                .on_failure(FailurePolicies::default().with_decode(FailurePolicy::Skip)),
        );
    })
}

Provided Methods

fn name( self, name: impl Into<Cow<'static, str>>, ) -> <Self::Settings as NameStep>::Out

where Self::Settings: NameStep,

Names the subscription, building the kind the attribute fixed.

Available while the attribute left the source open (#[subscriber] for the by-name source, #[subscriber(Kind)] for a named kind); the kind is built through FromName.

fn workers(self, count: NonZero<usize>) -> <Self::Settings as WorkersStep>::Out

where Self::Settings: WorkersStep,

Processes up to count deliveries (or batches) of this subscriber concurrently. The mount-site spelling of workers(n).

fn workers_by_key( self, count: NonZero<usize>, ) -> <Self::Settings as WorkersStep>::Out

where Self::Settings: WorkersStep,

Runs count sequential lanes keyed by the message’s partition key, preserving per-key ordering. The mount-site spelling of workers(n, by_key).

fn threads(self, count: NonZero<usize>) -> <Self::Settings as WorkersStep>::Out

where Self::Settings: WorkersStep,

Handles the deliveries (or batches) of this subscriber on count dedicated threads of its own, for a handler that computes. The mount-site spelling of threads(n); it fills the same position as workers, so a subscription names one of the two.

fn threads_by_key( self, count: NonZero<usize>, ) -> <Self::Settings as WorkersStep>::Out

where Self::Settings: WorkersStep,

Runs count dedicated threads as sequential lanes keyed by the message’s partition key, preserving per-key ordering. The mount-site spelling of threads(n, by_key).

fn on_failure( self, policies: FailurePolicies, ) -> <Self::Settings as FailureStep>::Out

where Self::Settings: FailureStep,

Sets the policies applied to a handler panic and to a delivery that fails to materialize. The mount-site spelling of on_failure(panic = .., decode = ..).

fn start_at<P>(self, position: P) -> <Self::Settings as StartAtStep<P>>::Out

where Self::Settings: StartAtStep<P>,

Opens the subscription at position instead of the broker’s default. The mount-site spelling of start_at(<position>).

fn batch(self, size: NonZero<usize>) -> <Self::Settings as BatchStep>::Out

where Self::Settings: BatchStep,

Opens the subscription in batches of at most size messages: the one parameter every batch handler names, and the only one the framework carries down to the broker.

The broker builds the batch - XREADGROUP COUNT, a pull batch, a poll limit, or the framework’s own client-side buffer where the transport has no batching of its own - and what the body sees is exactly what the broker delivered, never a slice of it. Everything else about how a batch forms (a block timeout, a consumer group, a prefetch window) is the broker’s own vocabulary, chained after this on its subscription source.

Mandatory on a batch body (&[T] and friends), whatever else its signature carries: mounting one without it does not compile. A single-message body takes no batch, so the step is not offered there at all - how many deliveries it handles at once is workers.

fn map_source<F>(self, f: F) -> <Self::Settings as MapSourceStep<F>>::Out

where Self::Settings: MapSourceStep<F>,

Transforms the source under construction.

The hook a broker crate layers its own settings trait on: core cannot know that a subscription has a JetStream stream or a consumer group, so the broker declares a trait over SubscriberBuilder bound to its own source type and implements each method as one map_source call. The bound means those methods simply do not exist on a builder for another broker.

One core setting changes the source type instead of a state slot: start_at(..) wraps the descriptor in StartAt, so the broker covers that case with a second impl over the wrapped source, each method one StartAt::map_inner call. A step that depends on a core setting (a resource per dedicated thread, a window sized to the batch) reads it off the builder with dispatch, batch_size and subscription_name instead of asking the user to name it twice. A step sees each value as it stands when it is called: the attribute’s settings are fixed before any mount-site step, and a step chained before .threads(n) sees the default sequential dispatch. The publish-side twin over a policy is MapPublisher.

Examples

use std::time::Duration;

use ruststream::Buffered;
use ruststream::memory::prelude::*;
use ruststream::runtime::{Declared, SubscriberBuilder};
use serde::Deserialize;

/// A broker crate's mount-site step for a transport that delivers one message at a time:
/// the batches are assembled on the client, and a partial one closes `wait` after it
/// opened.
pub trait Linger {
    type Out;
    fn linger(self, wait: Duration) -> Self::Out;
}

impl<Def: Declared, State, DefCodec> Linger
    for SubscriberBuilder<Def, MemorySource, State, DefCodec>
{
    type Out = SubscriberBuilder<Def, Buffered<MemorySource>, State, DefCodec>;

    fn linger(self, wait: Duration) -> Self::Out {
        self.map_source(|source| Buffered::new(source).max_wait(wait))
    }
}

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

#[subscriber(MemorySource)]
async fn settle(orders: &[Order]) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "settled a batch");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(
            settle
                .name("orders")
                .batch(nonzero!(64))
                .linger(Duration::from_millis(20)),
        );
    })
}

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<D> SubscriberSettings for D

where D: Declared,