Skip to main content

SubscriptionSource

ruststream

Trait SubscriptionSource 

Source 
pub trait SubscriptionSource<C: ConnectedBroker> {
    type Subscriber: Subscriber;
    type Copies: CopyPath;

    // Required methods
    fn name(&self) -> &str;
    fn subscribe(
        self,
        connected: &C,
    ) -> impl Future<Output = Result<Self::Subscriber, C::Error>> + Send;

    // Provided methods
    fn declare_retry(self, declaration: &RetryDeclaration) -> Self
       where Self: Sized { ... }
    fn declare_retry_on(
        &self,
        connected: &C,
        declaration: &RetryDeclaration,
    ) -> Result<(), DeclareRetryError> { ... }
    fn channel_bindings(&self) -> Bindings { ... }
    fn operation_bindings(&self) -> Bindings { ... }
    fn message_bindings(&self) -> Bindings { ... }
}
Expand description

A description of one subscription, resolved against a connected broker at startup.

The runtime calls subscribe once, against the ConnectedBroker witness produced by Broker::connect, to obtain the live Subscriber. The associated Subscriber type lives on the source rather than the broker, so a single broker can offer several subscription kinds with different subscriber types (for example Redis pub/sub versus streams).

§Implementing it

Subscribe covers a subscription a name alone identifies. One that needs settings of the broker’s own (a consumer group, a durable name, a delivery policy) gets a descriptor type implementing this trait.

  • Give it an associated constructor, OrdersStream::new("orders", "workers"), not a free function: a service names it in the attribute, #[subscriber(OrdersStream::new(..))], and the macro reads the type out of the call. A builder chain on it works too (OrdersStream::new("orders").durable("workers")) as long as each method returns Self.
  • Derive Clone: a mount rebuilds the configuration per registration, so one definition can be mounted on two brokers at once.
  • A kind identified by one name and nothing else also implements FromName, which makes the bare kind legal in the attribute.
  • Answer Copies. AddressedCopies and NamedCopies pair a retry publisher for every registration from the broker’s DefaultPublish policy, so a descriptor declaring either over a broker without one does not compile. Where native redelivery depends on a field’s value rather than on the type (a queue without a delay setting has none of its own), the descriptor keeps a copy path open.
  • Read the registration’s cap and dead-letter destination in declare_retry where the broker has a mechanism for them.

conformance::harness::redelivery_address checks an AddressedCopies descriptor’s answer, and conformance::retry::broker_moves a BrokerMoves descriptor’s mapping of the cap. Settings a service adds at the mount site reach the descriptor through an extension trait over SubscriberBuilder, and protocol bindings for the AsyncAPI document come from its *_bindings methods.

§Examples

use ruststream::{ConnectedBroker, SubscriptionSource};

async fn open<C, S>(source: S, connected: &C) -> Result<S::Subscriber, C::Error>
where
    C: ConnectedBroker,
    S: SubscriptionSource<C>,
{
    source.subscribe(connected).await
}

Required Associated Types§

Source

type Subscriber: Subscriber

The subscriber type this source opens.

Source

type Copies: CopyPath

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

Answer AddressedCopies where one subscription reads one destination this process can publish to, and implement RedeliveryAddressed beside it - the copy path requires it. Answer NamedCopies where the subscription reads many (a wildcard, a filter, a pattern, a list) and the mount site has to name one. Answer BrokerMoves where the broker applies a delivery limit and a dead-letter destination on its own (a quorum queue with x-delivery-limit and an x-dead-letter-exchange, a Pub/Sub dead-letter policy, an SQS redrive policy, a Pulsar DeadLetterPolicy); that makes .out_retry(..) a compile error at every mount site of this descriptor, because there is no publisher of this process’s to customise.

Required Methods§

Source

fn name(&self) -> &str

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

Used for handler metadata and AsyncAPI generation; it need not be the only routing information the source carries.

Source

fn subscribe( self, connected: &C, ) -> impl Future<Output = Result<Self::Subscriber, C::Error>> + Send

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

§Errors

Returns ConnectedBroker::Error when the broker rejects the subscription or the transport fails.

Provided Methods§

Source

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

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

Called once per registration, with what the mount site declared with max_attempts(..) and dead_letter(..). A broker that applies a delivery limit and a dead-letter destination itself reads them here and configures the subscription with them - and only when both are declared, because a native dead-letter policy needs both. The default keeps the descriptor as it was, which leaves the declaration to the runtime.

§Examples
use std::borrow::Cow;

use ruststream::{AddressedCopies, ConnectedBroker, RedeliveryAddress};
use ruststream::{RedeliveryAddressed, RetryDeclaration, Subscribe, SubscriptionSource};

/// The broker crate's own channel call: declares a queue with its limit and dead-letter
/// destination.
trait DeclareQueue: ConnectedBroker {
    fn declare_queue(
        &self,
        name: &str,
        delivery_limit: Option<u32>,
        dead_letter: Option<&str>,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
}

/// A queue this broker declares itself, so it takes the declaration into its topology.
#[derive(Debug, Clone)]
struct Queue {
    name: Cow<'static, str>,
    delivery_limit: Option<u32>,
    dead_letter: Option<Cow<'static, str>>,
}

impl<C: Subscribe + DeclareQueue> SubscriptionSource<C> for Queue {
    type Subscriber = C::Subscriber;
    type Copies = AddressedCopies;

    fn name(&self) -> &str {
        &self.name
    }

    async fn subscribe(self, connected: &C) -> Result<Self::Subscriber, C::Error> {
        connected
            .declare_queue(&self.name, self.delivery_limit, self.dead_letter.as_deref())
            .await?;
        connected.subscribe(&self.name).await
    }

    fn declare_retry(mut self, declaration: &RetryDeclaration) -> Self {
        self.delivery_limit = declaration.max_attempts().map(|n| n.get());
        self.dead_letter = declaration.dead_letter().map(|d| Cow::Owned(d.to_owned()));
        self
    }
}

// The queue is one destination, so it answers where a copy reaches it again.
impl<C: Subscribe + DeclareQueue> RedeliveryAddressed<C> for Queue {
    async fn redelivery_address(&self, _connected: &C) -> Result<RedeliveryAddress, C::Error> {
        Ok(RedeliveryAddress::new(self.name.clone()))
    }
}
Source

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.

A descriptor of your own reads the declaration in declare_retry and turns it into topology there, so it leaves this one at its default, which carries nothing. The core’s Name source is the one descriptor that overrides it: a name alone says nothing about what a cap and a dead-letter destination mean for the subscription, so it hands both to Subscribe::declare_retry, where the broker answers for the name it is about to open.

§Errors

Returns DeclareRetryError when the broker refuses the declaration, which fails the registration at startup, naming the subscription.

§Examples
use ruststream::{ConnectedBroker, DeclareRetryError, RetryDeclaration, SubscriptionSource};

fn declare<C, S>(
    source: &S,
    connected: &C,
    declaration: &RetryDeclaration,
) -> Result<(), DeclareRetryError>
where
    C: ConnectedBroker,
    S: SubscriptionSource<C>,
{
    source.declare_retry_on(connected, declaration)?;
    Ok(())
}
Source

fn channel_bindings(&self) -> Bindings

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

This is where a broker’s own vocabulary reaches the document: a RabbitMQ queue’s durability and its exchange, a Kafka topic’s name where it differs from the channel id, a Pulsar namespace. The core never names a field of yours - it carries what you build with Binding and writes bindingVersion for you.

Three rules bound what belongs here. The value is computed from this descriptor alone, because the document is built before anything connects: a Kafka topic’s real partition count, the topic behind a Pub/Sub subscription and an SQS queue’s ARN cannot be reported from here at all. A credential never goes in, for the reason DescribeServer gives: the document is published and shared. And a protocol the specification has no binding for goes in Binding::extension, because the protocol keys are a closed list.

The default says nothing, and a broker that says nothing changes no document.

§Examples
use ruststream::asyncapi::{Binding, Bindings};
use ruststream::memory::MemoryBroker;
use ruststream::{Connected, SubscriptionSource};
use serde::Serialize;

#[derive(Serialize)]
struct AmqpChannel {
    queue: Queue,
}

#[derive(Serialize)]
struct Queue {
    name: String,
    durable: bool,
}

fn channel_bindings(&self) -> Bindings {
    let body = AmqpChannel {
        queue: Queue { name: self.name.clone(), durable: self.durable },
    };
    // A binding that fails to build is a binding the document goes without: a broker
    // never holds up a service over a description of itself.
    match Binding::new("amqp", "0.3.0", &body) {
        Ok(binding) => Bindings::new().with(binding),
        Err(_) => Bindings::new(),
    }
}
Source

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

The consumer’s own settings live here rather than on the channel: a NATS queue group, a Kafka consumer group, an MQTT QoS. The rules of channel_bindings apply unchanged.

§Examples
use ruststream::asyncapi::{Binding, Bindings};
use serde::Serialize;

#[derive(Serialize)]
struct NatsOperation {
    queue: String,
}

fn operation_bindings(&self) -> Bindings {
    let body = NatsOperation { queue: self.group.clone() };
    match Binding::new("nats", "0.1.0", &body) {
        Ok(binding) => Bindings::new().with(binding),
        Err(_) => Bindings::new(),
    }
}
Source

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.

A Kafka record’s key schema and where its schema id sits, a Pub/Sub ordering key. The rules of channel_bindings apply unchanged.

§Examples
use ruststream::asyncapi::{Binding, Bindings};
use serde::Serialize;

#[derive(Serialize)]
struct KafkaMessage {
    #[serde(rename = "schemaIdLocation")]
    schema_id_location: &'static str,
}

fn message_bindings(&self) -> Bindings {
    let body = KafkaMessage { schema_id_location: "payload" };
    match Binding::new("kafka", "0.5.0", &body) {
        Ok(binding) => Bindings::new().with(binding),
        Err(_) => Bindings::new(),
    }
}

Dyn Compatibility§

This trait is not dyn compatible.

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

Implementors§