Struct Slot 

pub struct Slot<M, W, E, Pipe = PublishIdentity, Body = ()> { /* private fields */ }

One arena entry: the wired live value of the marker M, plus the publish path the include site gave it - its encode codec and the pipeline every message leaving the slot travels.

The wired value is the bound policy’s Live form, and the entry is a transparent window onto it: Deref reaches every method the live value offers - the broker capability vocabulary (TransactionalPublisher, OwnedTransactions, RequestReply) and any broker-defined capability trait alike, so a body pins the entry to the broker’s concrete live type (or bounds OutEntry::Wire with the broker’s trait) and calls it directly. Under each of the core capability bounds the entry also offers that capability’s typed form - message, begin, transaction - over the include site’s codec and the marker’s dictionary.

A body does not normally spell this type out: it names one parameter per marker and bounds it with OutEntry, which projects W, E and Pipe so a mount the body never sees cannot change its signature. Spelling the entry is for pinning it to a concrete live value.

Pipe is that publish path: the app’s own publish pipeline (the publish_layer chain) with the slot’s .transform(..) steps composed on top. It is PublishIdentity - nothing in the way, the bare leaf call - until a mount site names either.

Autoderef carries a method call, not a trait bound: a helper written as fn f<L: Lanes>(l: &L) rejects the entry a body holds. A broker crate that wants such helpers - or bodies generic over its capability - adds one blanket impl next to its trait, impl<M, W: Lanes, E, Pipe, Body> Lanes for Slot<M, W, E, Pipe, Body>, delegating through this Deref.

Body is the entry’s declared message set, () (any dictionary type) unless the #[subscriber] parameter’s third Out position narrows it; message checks it at compile time (see ContainsMessage).

Implementations

impl<M, W, E, Pipe, Body> Slot<M, W, E, Pipe, Body>

where M: OutSlot, W: Publisher, E: Codec + Send + Sync, Pipe: OutPipeline<W>,

pub fn message<'a, T, Index>( &'a self, value: &'a T, ) -> PublishBuilder<&'a Slot<M, W, E, Pipe, Body>, MessageBody<'a, T>, &'a E, HeadersUnset, <T as OutgoingDestination>::Form>

where Body: ContainsMessage<T, Index>, T: OutgoingDestination + PublishedThrough<M>,

Starts a typed publish through the slot, on the message type’s own wire (MessageWire): a serde::Serialize value encodes with the include site’s codec, a Serialized one carries its bytes and they leave as they are. The message type has to be in the marker’s #[publishes(..)] dictionary (see PublishedThrough) and, when the entry carries a declared message set, in that set (see ContainsMessage); everything else - the destination and the header contract - comes from the type’s #[derive(Outgoing)] declaration, so the builder demands exactly the positions that declaration leaves open.

Examples

use ruststream::runtime::{HandlerOutcome, Out};
use ruststream::{Outgoing, OutSlot, Publisher, subscriber};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct DoneMeta {
    task_id: u64,
}

#[derive(Outgoing, Serialize)]
#[outgoing(name = "chunks.progress")]
struct Progress {
    percent: u8,
}

#[derive(Outgoing, Serialize)]
#[outgoing(name = "chunks.done", headers = DoneMeta)]
struct ChunkDone {
    output_key: String,
}

#[derive(OutSlot)]
#[publishes(ChunkDone, Progress)]
struct Events;

#[subscriber("chunks.raw")]
async fn convert(
    event: &Event,
    Out(out): Out<impl Publisher, Events, (ChunkDone, Progress)>,
) -> HandlerOutcome {
    // No headers contract on Progress: publish straight away.
    if out.message(&Progress { percent: 100 }).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    // ChunkDone declares DoneMeta: with_headers is demanded by the contract.
    let done = ChunkDone { output_key: format!("out/{}", event.id) };
    let meta = DoneMeta { task_id: event.id };
    if out.message(&done).with_headers(&meta).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

impl<M, W, E, Pipe, Body> Slot<M, W, E, Pipe, Body>

where M: OutSlot, W: TransactionalPublisher, E: Send + Sync, Pipe: Send + Sync,

pub async fn begin( &self, ) -> Result<TransactionScope<'_, SlotPublisher<W, M>, &E, Slot<M, W, E, Pipe, Body>>, <W as Publisher>::Error>

Opens the entry’s broker transaction and returns the TransactionScope that owns it: publishes go through the scope, and commit or abort consume it, so a commit without a begin, a second commit, or a publish after settling do not compile.

This is the borrowed kind: the wired publisher carries at most one broker-side transaction, so one scope per entry is open at a time. The scope’s typed entry admits what the slot’s own does (the marker’s dictionary, narrowed by the parameter’s declared set), and publishes issued through it keep the slot’s test-capture attribution. They go to the broker as they are: the scope opens on the attributed leaf, so the slot’s transforms and the app-wide pipeline stay on the direct publish path (see TransactionScope).

Errors

Returns the publisher’s error when the broker refuses to start a transaction, or when one is already open on this handle.

Examples

use ruststream::runtime::{HandlerOutcome, Out};
use ruststream::{Outgoing, OutSlot, TransactionalPublisher, subscriber};
use serde::Serialize;

#[derive(Outgoing, Serialize)]
#[outgoing(name = "ledger.settled")]
struct Settled {
    id: u64,
}

#[derive(OutSlot)]
#[publishes(Settled)]
struct Journal;

#[subscriber("ledger.orders")]
async fn settle(
    order: &Order,
    Out(journal): Out<impl TransactionalPublisher, Journal, Settled>,
) -> HandlerOutcome {
    let Ok(scope) = journal.begin().await else {
        return HandlerOutcome::retry();
    };
    if scope.message(&Settled { id: order.id }).publish().await.is_err()
        || scope.commit().await.is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

impl<M, W, E, Pipe, Body> Slot<M, W, E, Pipe, Body>

where M: OutSlot, W: OwnedTransactions, E: Send + Sync, Pipe: Send + Sync,

pub async fn transaction( &self, ) -> Result<TypedTransaction<<W as OwnedTransactions>::Transaction, &E, Slot<M, W, E, Pipe, Body>>, <W as Publisher>::Error>

Opens an independent, caller-owned broker transaction and returns the TypedTransaction that owns it: publishes buffer into the value, and commit or abort consume it.

Every call opens its own transaction, so any number can be open on one entry at a time; the transaction’s typed entry admits what the slot’s own does. The buffer settles outside the slot, so its publishes land in the broker’s publish log and are not attributed to the slot (the documented capture boundary).

Errors

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

Examples

use ruststream::runtime::{HandlerOutcome, Out};
use ruststream::{Outgoing, OutSlot, OwnedTransactions, subscriber};
use serde::Serialize;

#[derive(Outgoing, Serialize)]
#[outgoing(name = "ledger.settled")]
struct Settled {
    id: u64,
}

#[derive(OutSlot)]
#[publishes(Settled)]
struct Ledger;

#[subscriber("ledger.orders")]
async fn settle(
    order: &Order,
    Out(ledger): Out<impl OwnedTransactions, Ledger, Settled>,
) -> HandlerOutcome {
    let Ok(mut txn) = ledger.transaction().await else {
        return HandlerOutcome::retry();
    };
    if txn.message(&Settled { id: order.id }).publish().await.is_err()
        || txn.commit().await.is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Trait Implementations

impl<M, W, E, Pipe, Body> Debug for Slot<M, W, E, Pipe, Body>

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

Formats the value using the given formatter.

impl<M, W, E, Pipe, Body> Deref for Slot<M, W, E, Pipe, Body>

type Target = W

The resulting type after dereferencing.

fn deref(&self) -> &W

Dereferences the value.

impl<M, W, E, Pipe, Body> OutEntry<M, Body> for Slot<M, W, E, Pipe, Body>

where W: Send + Sync, E: Codec, Pipe: OutPipeline<W>,

type Wire = W

The wired live value: the bound policy’s Live form, which a body bounds with the broker capability it needs.

type Enc = E

The codec the entry’s typed publishes encode with: the include site’s own, or the one a .codec(..) named for this slot.

type Pipe = Pipe

The publish path every message leaving the entry travels: the app’s publish pipeline with the slot’s .transform(..) steps composed on top. It is stated over the wired value, because a transform there writes that publisher’s own per-message options.

impl<M, W, E, Pipe, Body> OwnedTransactions for Slot<M, W, E, Pipe, Body>

where M: OutSlot, W: OwnedTransactions, E: Send + Sync, Pipe: OutPipeline<W>,

type Transaction = <W as OwnedTransactions>::Transaction

The buffer-owning transaction opened by transaction.

async fn transaction( &self, ) -> Result<<Slot<M, W, E, Pipe, Body> as OwnedTransactions>::Transaction, <Slot<M, W, E, Pipe, Body> as Publisher>::Error>

Opens a new transaction owned by the returned value.

impl<M, W, E, Pipe, Body> Publisher for Slot<M, W, E, Pipe, Body>

where M: OutSlot, W: Publisher, E: Send + Sync, Pipe: OutPipeline<W>,

type Payload = <W as Publisher>::Payload

How this transport consumes the payload: Lend where it reads the bytes and keeps nothing, Take where its client keeps them.

type Error = <Pipe as OutPipeline<W>>::Error<<W as Publisher>::Error>

The error type returned by publish.

type Options = <W as Publisher>::Options

The broker’s per-message settings: a QoS, a priority, an ordering key, whatever this transport lets one message differ from the next in.

async fn publish( &self, msg: OutgoingMessage<'_, <<Slot<M, W, E, Pipe, Body> as Publisher>::Payload as PayloadForm>::Form<'_>>, options: Option<&<Slot<M, W, E, Pipe, Body> as Publisher>::Options>, ) -> Result<(), <Slot<M, W, E, Pipe, Body> as Publisher>::Error>

Publishes a message to the broker, with the per-message settings the call site adjusted.

fn base_headers(&self) -> Option<&HeaderMap>

The headers this publisher contributes to every message it sends, underneath whatever the message itself names.

impl<M, W, E, Pipe, Body> RequestReply for Slot<M, W, E, Pipe, Body>

where M: OutSlot, W: RequestReply, E: Send + Sync, Pipe: OutPipeline<W>,

type Reply = <W as RequestReply>::Reply

The reply message type.

async fn request( &self, msg: OutgoingMessage<'_, <<Slot<M, W, E, Pipe, Body> as Publisher>::Payload as PayloadForm>::Form<'_>>, timeout: Duration, ) -> Result<<Slot<M, W, E, Pipe, Body> as RequestReply>::Reply, <Slot<M, W, E, Pipe, Body> as Publisher>::Error>

Publishes msg and awaits a single correlated reply, or fails after timeout.

impl<M, W, E, Pipe, Body> TransactionalPublisher for Slot<M, W, E, Pipe, Body>

where M: OutSlot, W: TransactionalPublisher, E: Send + Sync, Pipe: OutPipeline<W>,

async fn begin_transaction( &self, ) -> Result<(), <Slot<M, W, E, Pipe, Body> as Publisher>::Error>

Begins a new transaction on this publisher.

async fn commit( &self, ) -> Result<(), <Slot<M, W, E, Pipe, Body> as Publisher>::Error>

Commits the active transaction, making all buffered messages visible atomically.

async fn abort( &self, ) -> Result<(), <Slot<M, W, E, Pipe, Body> as Publisher>::Error>

Aborts the active transaction, discarding all buffered messages.