Struct ConnectedRedisBroker 

pub struct ConnectedRedisBroker { /* private fields */ }

The typed witness that Broker::connect succeeded: it owns the live fred pool.

Every subscription and every publisher is reached from here, so “not connected” is not representable. ConnectedBroker::shutdown consumes it, which makes a publish or subscribe after shutdown a compile error for the owner of the handle.

Implementations

impl ConnectedRedisBroker

pub async fn subscribe( &self, def: RedisStream, ) -> Result<RedisSubscriber, RedisError>

Opens a stream subscription described by def.

Ensures the consumer group exists (XGROUP CREATE ... MKSTREAM, ignoring an already-existing group) before returning the subscriber.

Errors

Returns RedisError::ShutDown when the connection was already torn down, RedisError::InvalidOptions when def names no consumer group, or RedisError::Subscribe when the group cannot be created.

pub async fn subscribe_pubsub( &self, def: RedisPubSub, ) -> Result<RedisPubSubSubscriber, RedisError>

Opens a Pub/Sub subscription on one channel, described by def, on a dedicated client.

Errors

Returns RedisError::ShutDown when the connection was already torn down, RedisError::Connect when the dedicated client cannot connect, or RedisError::Subscribe when the subscribe command fails.

pub async fn subscribe_pubsub_pattern( &self, def: RedisPubSubPattern, ) -> Result<RedisPubSubSubscriber, RedisError>

Opens a Pub/Sub subscription on a glob (PSUBSCRIBE), described by def, on a dedicated client.

Errors

Returns RedisError::ShutDown when the connection was already torn down, RedisError::Connect when the dedicated client cannot connect, or RedisError::Subscribe when the subscribe command fails.

pub async fn subscribe_list( &self, def: RedisList, ) -> Result<RedisListSubscriber, RedisError>

Opens a list (work-queue) subscription described by def.

Errors

Returns RedisError::ShutDown when the connection was already torn down, RedisError::InvalidOptions when def names a recovery ZSET without a min_idle, or RedisError::Connect when the connection the subscription reads on cannot connect.

pub fn publisher(&self) -> RedisPublisher

Returns a stream publisher (XADD) bound to this connection.

pub fn default_publisher(&self) -> RedisDefaultPublisher

Returns the publisher of the broker’s default policy, RedisDefaultPublish, which writes each name the way this connection’s subscriptions read it.

pub fn pubsub_publisher( &self, publish: RedisPubSubPublish, ) -> RedisPubSubPublisher

Returns a Pub/Sub publisher configured by publish (mode and envelope codec).

pub fn list_publisher(&self, publish: RedisListPublish) -> RedisListPublisher

Returns a list publisher (LPUSH) configured by publish (envelope codec and key TTL).

pub fn pool_handle(&self) -> Result<Pool, RedisError>

Returns a clone of the underlying pool, for advanced operations not covered by the wrapper.

Errors

Returns RedisError::ShutDown once the connection was torn down.

Trait Implementations

impl ConnectedBroker for ConnectedRedisBroker

async fn shutdown(self) -> Result<Self::Closed, Self::Error>

Closes every pooled connection and marks the shared connection dead, so publishers handed out earlier report RedisError::ShutDown instead of running against a closed pool.

Errors

Returns RedisError::Connect when the QUIT roundtrip fails.

type Error = RedisError

The error type returned by connected-broker operations.

type Closed = ClosedRedisBroker

The terminal witness returned by shutdown.

impl Debug for ConnectedRedisBroker

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

Formats the value using the given formatter.

impl DefaultPublish for ConnectedRedisBroker

type Policy = RedisDefaultPublish

The broker’s plain publish policy, constructible with its defaults.

impl PublishPolicy<ConnectedRedisBroker> for RedisPublish

fn channel_bindings(&self, channel: &str) -> Bindings

An XADD carries its headers as entry fields and the policy has no settings of its own, so what the document learns from here is which Redis structure the channel is and which stream key the entries are appended to.

type Live = RedisPublisher

The live form this policy pairs into: a Publisher for a leaf policy, or the live wiring form for a combinator stack (a typed publisher over a policy pairs into the same typed publisher over the live leaf).

fn pair( self, connected: &ConnectedRedisBroker, ) -> impl Future<Output = Result<Self::Live, PairError>>

Pairs the policy with a connected broker, producing the live publisher.

fn operation_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to its send operation in the document.

fn message_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to the messages that leave through it.

fn reply_address_location(&self) -> Option<&'static str>

Where a client finds the address of a reply this policy publishes, as a runtime expression.

impl PublishPolicy<ConnectedRedisBroker> for RedisDefaultPublish

fn channel_bindings(&self, channel: &str) -> Bindings

Described as the stream publish, the family of a name no subscription of the service reads; the document is built before any subscription opens, so a name that turns out to be read as a list or a channel is described as a stream.

type Live = RedisDefaultPublisher

The live form this policy pairs into: a Publisher for a leaf policy, or the live wiring form for a combinator stack (a typed publisher over a policy pairs into the same typed publisher over the live leaf).

fn pair( self, connected: &ConnectedRedisBroker, ) -> impl Future<Output = Result<Self::Live, PairError>>

Pairs the policy with a connected broker, producing the live publisher.

fn operation_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to its send operation in the document.

fn message_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to the messages that leave through it.

fn reply_address_location(&self) -> Option<&'static str>

Where a client finds the address of a reply this policy publishes, as a runtime expression.

impl PublishPolicy<ConnectedRedisBroker> for RedisListPublish

fn channel_bindings(&self, channel: &str) -> Bindings

The list key the policy pushes onto, the expiry it re-arms there on every push, and how headers are framed beside the payload.

type Live = RedisListPublisher

The live form this policy pairs into: a Publisher for a leaf policy, or the live wiring form for a combinator stack (a typed publisher over a policy pairs into the same typed publisher over the live leaf).

fn pair( self, connected: &ConnectedRedisBroker, ) -> impl Future<Output = Result<Self::Live, PairError>>

Pairs the policy with a connected broker, producing the live publisher.

fn operation_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to its send operation in the document.

fn message_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to the messages that leave through it.

fn reply_address_location(&self) -> Option<&'static str>

Where a client finds the address of a reply this policy publishes, as a runtime expression.

impl PublishPolicy<ConnectedRedisBroker> for RedisPubSubPublish

fn channel_bindings(&self, channel: &str) -> Bindings

The channel a PUBLISH goes out on, the delivery mode it goes out in, and how headers are framed beside the payload.

type Live = RedisPubSubPublisher

The live form this policy pairs into: a Publisher for a leaf policy, or the live wiring form for a combinator stack (a typed publisher over a policy pairs into the same typed publisher over the live leaf).

fn pair( self, connected: &ConnectedRedisBroker, ) -> impl Future<Output = Result<Self::Live, PairError>>

Pairs the policy with a connected broker, producing the live publisher.

fn operation_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to its send operation in the document.

fn message_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to the messages that leave through it.

fn reply_address_location(&self) -> Option<&'static str>

Where a client finds the address of a reply this policy publishes, as a runtime expression.

impl RedeliveryAddressed<ConnectedRedisBroker> for RedisList

The list key: the subscription pops from it and RedisListPublish pushes onto it, so a deferred copy lands in the same queue the delivery came from.

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<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 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 RedeliveryAddressed<ConnectedRedisBroker> for RedisStream

The stream key, in every read mode: an XADD there is read by the consumer group this subscription belongs to.

Under reclaim the copy is read by the group’s tail reader rather than by this subscription, which only ever takes entries already pending on another consumer. That is the topology the mode is written for - a reclaim subscriber runs beside a new one on the same group - so the copy reaches the handler either way.

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

Where a publish reaches this subscription again.

impl Subscribe for ConnectedRedisBroker

type Copies = AddressedCopies

A bare name opens a consumer group over that stream key, and an XADD there is read by the group again, so the name is both ends and a retry copy needs nothing from the mount site.

type Subscriber = RedisSubscriber

The subscriber type opened by a by-name subscription.

async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error>

Opens a subscription to name, producing this broker’s Subscriber.

fn declare_retry( &self, name: &str, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Takes what a registration mounted by a bare name declared about its retries, before the subscription opens.

impl SubscriptionSource<ConnectedRedisBroker> for RedisList

type Copies = AddressedCopies

A retry copy is an LPUSH this process makes, and the list key is where it goes.

type Subscriber = RedisListSubscriber

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.

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 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.

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.

impl SubscriptionSource<ConnectedRedisBroker> for RedisStream

type Copies = AddressedCopies

A retry copy is an XADD this process makes, and the stream key is where it goes.

type Subscriber = RedisSubscriber

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.

impl TestableBroker for ConnectedRedisBroker

Available on crate feature testing only.

The harness’s view of the connection: the in-process server it injects into and reads the writes of, and, in both modes, where a publish is delivered.

Panics

inject and published panic on a broker connected with connect: the harness drives only the connection connect_in_process produced, and a live connection has no log to read and no synchronous way to take a message.

fn inject(&self, message: OutgoingMessage<'_>)

Writes the message the way the default publisher writes its name: an XADD, or the LPUSH or PUBLISH of the subscription that reads it, in that subscription’s framing. A name only pattern subscriptions read goes out with a raw PUBLISH, as the producers they listen to publish it: an XADD would reach none of them.

fn published(&self, name: &str) -> Vec<RawMessage>

Every write to name, decoded the way the subscription reading it decodes: an entry’s fields, or a list element or a channel message in its subscription’s framing. A requeue’s copy and a delayed entry the sweep adds back are writes too.

fn routes(&self, destination: &str, subscriptions: &[&str]) -> Vec<usize>

Redis’s routing: a stream entry is delivered once to every consumer group reading the stream (to one consumer of each), a list element to one consumer, and a channel message to every subscription of the channel and every pattern subscription whose glob matches it. A name no subscription reads exactly is reached by the pattern subscriptions alone.

Subscriptions sharing a name are told apart by nothing but the name, so a publish is owed by the first position of each name as many times as that name receives it. In process that is where it goes: a group’s new entries go to its earliest open subscription.

fn install_coordinator(&self, coordinator: Coordinator)

Installs the harness coordinator into this broker’s bus for a test run. Idempotent: a second install on the same broker is ignored.

fn backlog(&self) -> Backlog

What a subscription opened by name receives of the messages published to that name before it opened, as the real broker answers it: Backlog::Missed for publish/subscribe, Backlog::Delivered for a queue or a log that keeps them.