Struct Pipelined 

pub struct Pipelined<Descriptor, Mode = Plain> { /* private fields */ }

A subscription source with a window, as the .pipeline() mount step makes it.

The descriptor is a RedisStream, a RedisList, a RedisPubSub or a RedisPubSubPattern. Mode is Plain until .atomic() makes it Atomic. A service does not build one: it names the descriptor in #[subscriber(..)] and chains pipeline where it mounts the handler. The type shows up in the builder’s type, under the names RedisStreamPipeline and its siblings.

Examples

use ruststream_fred::stream::prelude::*;

#[subscriber(RedisStream::new("orders").group("workers"))]
async fn record(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
    if pipeline.hset("orders:by-id", (order.id.to_string(), "seen")).await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(record.pipeline());
        },
    )
}

Trait Implementations

impl<Descriptor: Clone, Mode> Clone for Pipelined<Descriptor, Mode>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<Descriptor: Debug, Mode: WindowMode> Debug for Pipelined<Descriptor, Mode>

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

Formats the value using the given formatter.

impl<Mode: WindowMode> RedeliveryAddressed<ConnectedRedisBroker> for Pipelined<RedisStream, Mode>

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

Where a publish reaches this subscription again.

impl<Mode: WindowMode> RedeliveryAddressed<ConnectedRedisBroker> for Pipelined<RedisList, Mode>

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

Where a publish reaches this subscription again.

impl<Mode: WindowMode> RedeliveryAddressed<ConnectedRedisBroker> for Pipelined<RedisPubSub, Mode>

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

Where a publish reaches this subscription again.

impl<Mode: WindowMode> SubscriptionSource<ConnectedRedisBroker> for Pipelined<RedisStream, Mode>

type Subscriber = PipelinedSubscriber<RedisSubscriber, StreamForm>

The subscriber type this source opens.

type Copies = AddressedCopies

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.

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

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

fn channel_bindings(&self) -> Bindings

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

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

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

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.

impl<Mode: WindowMode> SubscriptionSource<ConnectedRedisBroker> for Pipelined<RedisPubSub, Mode>

type Subscriber = PipelinedSubscriber<ChannelReader, PubSubForm>

The subscriber type this source opens.

type Copies = AddressedCopies

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.

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

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

fn channel_bindings(&self) -> Bindings

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

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

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

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.

impl<Mode: WindowMode> SubscriptionSource<ConnectedRedisBroker> for Pipelined<RedisPubSubPattern, Mode>

A pattern reads every channel its glob matches and names none a publish could reach, so the mount site names where a retry copy goes, as on the pattern subscription without a window.

type Subscriber = PipelinedSubscriber<ChannelReader, PubSubForm>

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.

impl<Mode: WindowMode> SubscriptionSource<ConnectedRedisBroker> for Pipelined<RedisList, Mode>

type Subscriber = PipelinedSubscriber<ListReader, ListForm>

The subscriber type this source opens.

type Copies = AddressedCopies

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.

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

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

fn channel_bindings(&self) -> Bindings

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

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

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

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.

impl<Descriptor> Windowed for Pipelined<Descriptor, Plain>

type Atomic = Pipelined<Descriptor, Atomic>

The same source with an atomic window.

fn into_atomic(self) -> Self::Atomic

Makes the window atomic.