Trait PublishExt 

pub trait PublishExt: Publisher {
    // Provided methods
    fn message<'a, T>(
        &'a self,
        value: &'a T,
    ) -> PublishBuilder<&'a Self, MessageBody<'a, T>, UnnamedCodec, HeadersUnset, <T as OutgoingDestination>::Form>
       where T: OutgoingDestination { ... }
    fn begin(
        &self,
    ) -> impl Future<Output = Result<TransactionScope<'_, Self, UnnamedCodec, AnyDeclared>, Self::Error>> + Send
       where Self: Sized + TransactionalPublisher { ... }
    fn owned_transaction(
        &self,
    ) -> impl Future<Output = Result<TypedTransaction<Self::Transaction, UnnamedCodec, AnyDeclared>, Self::Error>> + Send
       where Self: Sized + OwnedTransactions { ... }
}

The publish builder on any Publisher: message(..), the one entry point of a publish.

Blanket-implemented for every publisher, so a broker publisher held in the application state publishes through the same builder as an Out slot. The difference is the codec ladder: a bare publisher carries no codec of its own, so message(..) encodes with the crate’s DefaultCodec unless the call names one with with_codec(..). A surface that has a codec (an Out slot, a transaction scope opened on one) shadows this entry point with its own and uses that codec instead.

One entry point covers both wires. A serde::Serialize value takes the resolved codec; a Serialized value carries its own bytes and no codec runs on them, which is how a payload that is not a model of its own still travels as a declared type.

Examples

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

#[derive(Serialize, Outgoing)]
#[outgoing(name = "services.started")]
struct Started {
    service: &'static str,
}

/// Declares no destination, so the call site names one.
#[derive(Serialize, Outgoing)]
struct Seeded {
    rows: u64,
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("catalog", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The hook is handed the live publisher; `message(..)` is the same builder an `Out`
        // slot offers, encoding with the default codec.
        b.after_startup(Publish, async move |publisher| {
            publisher
                .message(&Seeded { rows: 1200 })
                .to("catalog.seeded")
                .publish()
                .await?;
            publisher
                .message(&Started { service: "catalog" })
                .publish()
                .await
        });
    })
}

Provided Methods

fn message<'a, T>( &'a self, value: &'a T, ) -> PublishBuilder<&'a Self, MessageBody<'a, T>, UnnamedCodec, HeadersUnset, <T as OutgoingDestination>::Form>

where T: OutgoingDestination,

Starts a typed publish of a #[derive(Outgoing)] value, encoded with the crate’s default codec (name another one with with_codec(..)). A Serialized value’s bytes leave as they are instead - the wire is the type’s own (MessageWire), and no codec runs on it.

The entry point exists whatever the build; what a build without a codec feature cannot do is encode, so a Serialize value published through it is a compile error naming the three ways out, and a Serialized value publishes as it always did.

fn begin( &self, ) -> impl Future<Output = Result<TransactionScope<'_, Self, UnnamedCodec, AnyDeclared>, Self::Error>> + Send

where Self: Sized + TransactionalPublisher,

Opens a broker transaction on this publisher and returns the TransactionScope that owns it.

The scope is the only handle on the transaction: publishes go through it, and it is consumed by commit or abort. A second begin on this handle, a commit without a begin, or a publish after the commit are not expressible - the methods do not exist on the types those states leave behind.

This is the typed sugar over the borrowed transaction kind (TransactionalPublisher): the scope claims the handle’s single broker-side transaction, so one scope per handle is open at a time. Brokers whose transactions are client buffers additionally offer the owned kind through owned_transaction, where every call opens an independent buffer-owning TypedTransaction.

The scope inherits this surface’s codec position, so it encodes with the crate default; name another one per publish with with_codec(..).

use ruststream::memory::MemoryBroker;
use ruststream::runtime::PublishExt;

let publisher = MemoryBroker::new().publisher();

let mut scope = publisher.begin().await?;
scope.publish("orders.settled", &42_u32).await?;
scope.commit().await?;

Errors

Returns the publisher’s error when the broker refuses to start a transaction.

fn owned_transaction( &self, ) -> impl Future<Output = Result<TypedTransaction<Self::Transaction, UnnamedCodec, AnyDeclared>, Self::Error>> + Send

where Self: Sized + OwnedTransactions,

Opens an owned broker transaction and returns the TypedTransaction that owns it.

The typed sugar over the owned transaction kind (OwnedTransactions), the counterpart of the borrowed begin: every call opens its own independent transaction and the returned value owns its buffer, so any number can be open on one publisher at a time - settling one never touches another. Kafka-like brokers, whose client holds exactly one transaction per producer, offer only the borrowed kind.

The name spells out owned because OwnedTransactions::transaction - the raw capability this wraps - is in scope wherever this is, and two same-named trait methods on one publisher would be ambiguous at the call.

use ruststream::memory::MemoryBroker;
use ruststream::runtime::PublishExt;

let publisher = MemoryBroker::new().publisher();

let mut orders = publisher.owned_transaction().await?;
let mut audit = publisher.owned_transaction().await?; // concurrent with `orders`
orders.publish("orders.settled", &42_u32).await?;
audit.publish("audit.trail", &7_u32).await?;
orders.commit().await?;
audit.commit().await?;

Errors

Returns the publisher’s error when the broker refuses to open a transaction; pure client-buffer implementations are infallible in practice.

Dyn Compatibility

This trait is not dyn compatible.

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

Implementors

impl<P> PublishExt for P

where P: Publisher + ?Sized,