Slots

An Out slot hands a handler a live publisher as a parameter: one marker per slot, bound to a publish policy at the mount site.

A slot is for a destination chosen at run time, or for several destinations at once, another broker's included. Out(out): Out<impl Publisher> names a capability, never a broker's publisher type, so the same handler mounts on a production broker and on its in-process test transport. The capability is Publisher, or a narrower one: TransactionalPublisher, OwnedTransactions or RequestReply. A single unnamed slot is DefaultSlot. A handler with several slots names a marker per parameter: a unit struct that derives OutSlot. Its #[publishes(..)] list is what may leave through the slot, and what the generated AsyncAPI document reports for it. An optional third position, Out<impl Publisher, Marker, (A, B)>, narrows what this handler publishes through the slot.

The mount site binds each marker with .out(marker, policy), in any order, and commits the chain with .build(). A transform written after an .out(..) belongs to that slot. A slot that is never bound, a marker the handler does not declare, and a publish of a type outside the list do not compile. A body publishes through the slot with the publish builder. A slot may also take a reply: one handler returns its answer and publishes through slots beside it. The runtime's own retry copies leave through one more slot, Retry.

Example

One unnamed slot.

Macros

examples/out_slots.rs

/// One unnamed slot: the publisher arrives live, and `Event` leaves the destination to the call.
#[subscriber("ingress")]
async fn forward(event: &Event, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    if out.message(event).to("egress").publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/out_slots.rs

/// One unnamed slot: the entry arrives live, and `Event` leaves the destination to the call.
struct Forward;

impl<Egress> Handle<Event, (), Outs<(Egress,)>> for Forward
where
    Egress: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        event: &Event,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if outs
            .get(DefaultSlot)
            .message(event)
            .to("egress")
            .publish()
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Two named slots, each with what it may publish.

Macros

examples/out_slots.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders.confirmed")]
struct Confirmed {
    id: u64,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct Audit {
    id: u64,
}

/// A slot marker lists what may leave through the slot.
#[derive(OutSlot)]
#[publishes(Confirmed)]
struct Orders;

#[derive(OutSlot)]
#[publishes(Audit)]
struct Trail;

#[subscriber("orders.incoming")]
async fn confirm(
    event: &Event,
    Out(orders): Out<impl Publisher, Orders>,
    Out(trail): Out<impl Publisher, Trail>,
) -> HandlerOutcome {
    if orders
        .message(&Confirmed { id: event.id })
        .publish()
        .await
        .is_err()
        || trail
            .message(&Audit { id: event.id })
            .to("audit")
            .publish()
            .await
            .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/out_slots.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Confirmed {
    id: u64,
}

impl OutgoingDestination for Confirmed {
    type Form = FixedName;

    const DESTINATION: &'static str = "orders.confirmed";
}

impl MessageHeaders for Confirmed {
    type Contract = NoHeaders;
}

#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Audit {
    id: u64,
}

impl OutgoingDestination for Audit {
    type Form = CallerName;
}

impl MessageHeaders for Audit {
    type Contract = NoHeaders;
}

/// A slot marker lists what may leave through the slot.
struct Orders;

impl OutSlot for Orders {
    const NAME: &'static str = "Orders";
    // `Confirmed` fixes its destination, so no transform on this slot may name one.
    type Destination = Reads;
}

impl PublishedThrough<Orders> for Confirmed {}

struct Trail;

impl OutSlot for Trail {
    const NAME: &'static str = "Trail";
    // `Audit` leaves its destination open, so a transform on this slot may name it.
    type Destination = Names;
}

impl PublishedThrough<Trail> for Audit {}

struct Confirm;

impl<O, T> Handle<Event, (), Outs<(O, T)>> for Confirm
where
    O: OutEntry<Orders, Wire: Publisher>,
    T: OutEntry<Trail, Wire: Publisher>,
{
    async fn handle(
        &self,
        event: &Event,
        outs: &Outs<(O, T)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if outs
            .get(Orders)
            .message(&Confirmed { id: event.id })
            .publish()
            .await
            .is_err()
            || outs
                .get(Trail)
                .message(&Audit { id: event.id })
                .to("audit")
                .publish()
                .await
                .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Macros

examples/out_slots.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(forward).out(DefaultSlot, Publish).build();
        // Each marker binds its own policy, in any order.
        b.include(confirm)
            .out(Orders, Publish)
            .out(Trail, Publish)
            .build();
    })
}

Manual

examples/manual/out_slots.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("ingress", Forward).build())
            .out(DefaultSlot, Publish)
            .build();
        // Each marker binds its own policy, in any order.
        b.include(subscriber("orders.incoming", Confirm).build())
            .out(Orders, Publish)
            .out(Trail, Publish)
            .build();
    })
}