Struct Workers 

pub struct Workers { /* private fields */ }

Concurrency policy for one subscriber’s dispatch loop, declared with the workers(..) or threads(..) macro argument (or Workers::sequential by default).

• workers(n): n long-lived workers, tasks of the runtime the app runs on, process up to n deliveries of the subscriber concurrently; a free worker takes the next one. Back-pressure holds: the stream is not polled while n deliveries are in flight. Global processing order is lost by design.
• workers(n, by_key): the same n workers as sequential lanes; a delivery goes to the lane picked by hashing its partition_key, so per-key ordering is preserved. Messages without a key rotate over the lanes.
• threads(n): n dedicated threads of the subscription’s own, each with a current-thread runtime, for a handler that computes. The subscription’s loop stays on the app’s runtime and spreads deliveries over the threads round-robin; decoding, the handler and settling run on the thread a delivery was handed to, and only there. The app’s runtime stays free for I/O, timers and the other subscriptions.
• threads(n, by_key): the same threads as sequential lanes by partition key.

The workers or threads and the queues that feed them are made when the subscription starts: handing a delivery to one allocates nothing.

The default is sequential dispatch (workers(1)).

Implementations

impl Workers

pub const fn sequential() -> Workers

Sequential dispatch: one delivery at a time, in stream order. The default.

pub const fn pool(count: NonZero<usize>) -> Workers

A pool of up to count concurrent deliveries. One behaves like sequential dispatch.

pub const fn keyed(count: NonZero<usize>) -> Workers

count sequential lanes keyed by the message partition_key: per-key ordering is preserved. One lane behaves like sequential dispatch.

pub const fn threads(count: NonZero<usize>) -> Workers

count dedicated threads for a CPU-bound handler: the subscription’s deliveries are handled on threads of its own, each running a current-thread runtime, so the handler’s computation holds up none of the app runtime’s threads. The router-chain spelling of threads(n).

Examples

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

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

#[subscriber("images.resize")]
async fn resize(image: &Image) -> HandlerOutcome {
    tracing::info!(image.id, "resized");
    HandlerOutcome::ack()
}

fn images() -> Router<MemoryBroker, impl RouterDef<MemoryBroker>> {
    // Resizing computes, so it runs on eight threads of its own.
    Router::new()
        .include(resize)
        .workers(Workers::threads(nonzero!(8)))
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("images", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include_router(images());
    })
}

pub const fn threads_keyed(count: NonZero<usize>) -> Workers

count dedicated threads as sequential lanes keyed by the message partition_key: a key always lands on the same thread, so per-key ordering is preserved. The router-chain spelling of threads(n, by_key).

Examples

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

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

#[subscriber("images.resize")]
async fn resize(image: &Image) -> HandlerOutcome {
    tracing::info!(image.id, "resized");
    HandlerOutcome::ack()
}

fn images() -> Router<MemoryBroker, impl RouterDef<MemoryBroker>> {
    // One tenant's images stay in order: a key always lands on the same thread.
    Router::new()
        .include(resize)
        .workers(Workers::threads_keyed(nonzero!(4)))
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("images", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include_router(images());
    })
}

Trait Implementations

impl Clone for Workers

fn clone(&self) -> Workers

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Copy for Workers

impl Debug for Workers

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

Formats the value using the given formatter.

impl Default for Workers

fn default() -> Workers

Returns the “default value” for a type.

impl Eq for Workers

impl PartialEq for Workers

fn eq(&self, other: &Workers) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl StructuralPartialEq for Workers