Publish

Publish is each form's one policy and its default: replies, slots and lifecycle hooks publish through it with no per-message setting.

Each form's prelude names its own policy Publish: FilePublish on a stream file, StdioPublish on a pipe. A service that uses both brokers globs the crate's prelude and writes the two names in full. The policy holds no connection; the runtime pairs it with the connected broker at startup. A registration that names no policy for its reply publishes through Publish, so the bare reply clause needs nothing at the mount site. .out(Ledger, Publish) wires an Out slot, and after_startup(Publish, ..) hands a lifecycle hook a publisher. A reply goes to the stream key its type declares with #[outgoing(name = "receipts")]. A reply type that declares no key is published where the subscriber says, with reply("receipts"). A publish carries a stream key and a payload, and its headers ride inside the payload in an envelope. The policy has no per-message setting, so the publish builder gains no step from this crate and a handler body keeps the plain bound Out<impl Publisher, Marker>. Each publish is written on its own: two publishes of one handler are two separate writes.

Example

The service answers each order with a receipt and writes an audit line through the ledger slot.

Macros

examples/sea_publish.rs

/// The answer to an order, appended under the key the type declares.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order: u64,
}

/// The audit line of an order, on a key of its own.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "audit")]
struct Audited {
    order: u64,
}

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

/// Answers each order with a receipt and writes its audit line through the ledger slot.
#[subscriber(FileStream::new("orders"), reply)]
async fn confirm(
    order: &Order,
    Out(ledger): Out<impl Publisher, Ledger>,
) -> Result<Receipt, HandlerOutcome> {
    if let Err(error) = ledger.message(&Audited { order: order.id }).publish().await {
        tracing::warn!(order.id, %error, "the audit line was not written");
        return Err(HandlerOutcome::retry());
    }
    Ok(Receipt { order: order.id })
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/shop/orders.ss"),
        |b| {
            // The reply needs no policy: the runtime pairs the form's own `Publish`.
            b.include(confirm).out(Ledger, Publish).build();
        },
    )
}

Manual

examples/manual/sea_publish.rs

/// The answer to an order, appended under the key the type declares.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "receipts")]
struct Receipt {
    order: u64,
}

/// The audit line of an order, on a key of its own.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "audit")]
struct Audited {
    order: u64,
}

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

/// Answers each order with a receipt and writes its audit line through the ledger slot.
struct Confirm;

impl<Entry> Handle<Order, Receipt, Outs<(Entry,)>> for Confirm
where
    Entry: OutEntry<Ledger, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<Receipt, HandlerOutcome> {
        let ledger = outs.get(Ledger);
        if let Err(error) = ledger.message(&Audited { order: order.id }).publish().await {
            tracing::warn!(order.id, %error, "the audit line was not written");
            return Err(HandlerOutcome::retry());
        }
        Ok(Receipt { order: order.id })
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/shop/orders.ss"),
        |b| {
            // The reply needs no policy: the runtime pairs the form's own `Publish`.
            b.include(
                subscriber(FileStream::new("orders"), Confirm)
                    .reply()
                    .build(),
            )
            .out(Ledger, Publish)
            .build();
        },
    )
}