pub trait PublishPolicy<C: ConnectedBroker> {
type Live;
// Required method
fn pair(
self,
connected: &C,
) -> impl Future<Output = Result<Self::Live, PairError>> + Send;
// Provided methods
fn channel_bindings(&self, channel: &str) -> Bindings { ... }
fn operation_bindings(&self, channel: &str) -> Bindings { ... }
fn message_bindings(&self, channel: &str) -> Bindings { ... }
fn reply_address_location(&self) -> Option<&'static str> { ... }
}Expand description
The declaration half of a publisher: pure policy, no connection, no publish surface.
A broker publisher is a bundle of policy (an exchange, a queue timeout, a transactional id)
paired with the live connection. PublishPolicy is that bundle alone, freely constructible
anywhere - before startup, in router definitions, in configuration - because it holds no
connection and no broker instance identity. pair joins it with a
ConnectedBroker witness to produce the live Publisher, so “not connected” is not
representable on this path: a publisher exists only after the connection does.
The policy is also where a broker’s per-message settings (Publisher::Options) get their
defaults: the mount site configures the policy, pairing hands the live publisher whatever it
fixed, and a call site adjusts single fields over that.
This is the publish-side mirror of SubscriptionSource. The
reply wiring a mount site’s chain builds over a policy is itself a PublishPolicy, resolved
functorially: pairing swaps the leaf for its live publisher and keeps the codec and transform
stacks the chain named, fully monomorphized.
pair is async and fallible because some brokers do real work when a publisher comes alive
(a transactional producer initializing its transactions); for most it is a cheap constructor
call. It reports the type-erased PairError: a broker wraps its own error with
PairError::new. A policy pairs once, at startup, so pair never reaches the hot path.
§Implementing it
A broker ships one policy and one live form per genuine publishing mode, and the choice of mode
is a transition of the policy type, not a runtime flag: a transactional_id(..) step moves
the plain policy to a distinct transactional policy whose live form implements
TransactionalPublisher, and the plain publisher has no
transactional surface at all. Where the plain policy is usable with its defaults, the
connected form also implements DefaultPublish. The in-memory broker’s MemoryPublish and
MemoryRequest, empty structs because they have no options, are the minimal reference.
§Examples
use ruststream::memory::{MemoryBroker, MemoryPublish};
use ruststream::runtime::PublishExt;
use ruststream::{Broker, Outgoing, PublishPolicy, Serialized};
// Bytes that are already the payload, so this example needs no codec feature.
#[derive(Outgoing, Serialized)]
struct Order(Vec<u8>);
let policy = MemoryPublish; // no broker in sight
let connected = MemoryBroker::new().connect().await?;
let publisher = policy.pair(&connected).await?; // live only past this point
publisher.message(&Order(b"{}".to_vec())).to("orders").publish().await?;Required Associated Types§
Required Methods§
Provided Methods§
Sourcefn channel_bindings(&self, channel: &str) -> Bindings
fn channel_bindings(&self, channel: &str) -> Bindings
What a publish through this policy adds to its channel in the generated AsyncAPI
document.
The publish-side mirror of
SubscriptionSource::channel_bindings, and
the same three rules bound it. The value is computed from the policy and the name it is
handed, because the document is built before anything connects, so a Kafka topic’s real
partition count cannot come from here. A credential never goes in, for the reason
DescribeServer gives. And a protocol the specification has no
binding for goes in Binding::extension.
The policy answers for every position it is bound on: the reply of a reply(..)
registration, an Out slot, and the publisher a dead-lettered
delivery leaves through. A Bound token answers with the policy
it carries, so a cross-broker publish is described by the broker it reaches. The channel’s
servers list does not follow it: that comes from the label the registration’s own broker
was registered under, which is the only broker the runtime can name for a channel.
channel is the destination the mount site resolved for that position - the reply type’s
own name or the reply("dest") clause, the slot entry’s name, the dead_letter("dlq")
declaration - and it is what the document reports as the channel’s address. An SNS topic
and an SQS queue are named by their binding’s required name field, and this is where
that name comes from: a policy declares a broker’s settings and never a destination. Where
a naming transform decides the destination per delivery, the document reports no address
and the mount site’s fallback name arrives here instead.
The default says nothing, and a policy that says nothing changes no document.
§Examples
use ruststream::asyncapi::{Binding, Bindings};
use serde::Serialize;
#[derive(Serialize)]
struct SnsChannel {
name: String,
}
fn channel_bindings(&self, channel: &str) -> Bindings {
let body = SnsChannel { name: channel.to_owned() };
// 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("sns", "0.1.0", &body) {
Ok(binding) => Bindings::new().with(binding),
Err(_) => Bindings::new(),
}
}Sourcefn operation_bindings(&self, channel: &str) -> Bindings
fn operation_bindings(&self, channel: &str) -> Bindings
What a publish through this policy adds to its send operation in the document.
The producer’s own settings live here rather than on the channel: a Kafka client id, an
MQTT QoS, an SQS delay. The rules of channel_bindings apply
unchanged, channel included: it is the same resolved destination.
A reply has no send operation - it is the reply of the operation it answers, and the
specification gives that object no bindings - so what this returns reaches the document
from a slot and from a dead-letter destination only.
§Examples
use ruststream::asyncapi::{Binding, Bindings};
use serde::Serialize;
#[derive(Serialize)]
struct SqsOperation {
queues: Vec<SqsQueue>,
}
#[derive(Serialize)]
struct SqsQueue {
name: String,
}
fn operation_bindings(&self, channel: &str) -> Bindings {
let body = SqsOperation { queues: vec![SqsQueue { name: channel.to_owned() }] };
match Binding::new("sqs", "0.3.0", &body) {
Ok(binding) => Bindings::new().with(binding),
Err(_) => Bindings::new(),
}
}Sourcefn message_bindings(&self, channel: &str) -> Bindings
fn message_bindings(&self, channel: &str) -> Bindings
What a publish through this policy adds to the messages that leave through it.
A Kafka record’s key schema, a Pub/Sub ordering key. The rules of
channel_bindings apply unchanged, channel included: it is
the same resolved destination.
§Examples
use ruststream::asyncapi::{Binding, Bindings};
use serde::Serialize;
#[derive(Serialize)]
struct GooglePubSubMessage {
#[serde(rename = "orderingKey")]
ordering_key: &'static str,
}
fn message_bindings(&self, _channel: &str) -> Bindings {
let body = GooglePubSubMessage { ordering_key: "tenant" };
match Binding::new("googlepubsub", "0.2.0", &body) {
Ok(binding) => Bindings::new().with(binding),
Err(_) => Bindings::new(),
}
}Sourcefn reply_address_location(&self) -> Option<&'static str>
fn reply_address_location(&self) -> Option<&'static str>
Where a client finds the address of a reply this policy publishes, as a runtime expression.
A broker that answers a request through a reply-to header names that header here:
"$message.header#/reply-to" is the specification’s form, and the part after the # is
a JSON Pointer into the request’s headers. The document then reports the reply channel
with address: null and puts the expression in the operation’s reply.address.location,
so a reader knows the destination is decided per delivery and where to read it.
It reaches the document only where a PublishTransform
on the reply position declares
Destination = Names: that transform is
what actually redirects the reply, and without it the reply goes to the name the mount
site declared, which is what the document reports.
The default is None: a broker whose replies carry no such address says nothing, and the
document keeps the declared name.
§Examples
/// NATS replies go to the inbox a request names in its `reply-to` header.
fn reply_address_location(&self) -> Option<&'static str> {
Some("$message.header#/reply-to")
}Dyn Compatibility§
This trait is not dyn compatible.
In older versions of Rust, dyn compatibility was called "object safety".