Publishing and replies

There are two ways to publish: return a reply from a handler, or publish explicitly through a publisher injected into the handler with the Out parameter. Either way the handler never sees an unconnected publisher: registrations carry publish policies (pure declarations), and the runtime pairs them with the connected broker at startup.

Replying from a handler

Name a reply destination with publish(..) and return the reply value. The runtime encodes it and sends it:

examples/publishing.rs

use ruststream::subscriber;

// A `publish(..)` handler that does not read the app state omits the `Context` parameter entirely;
// it stays generic over the state and mounts on an app with any state type.
#[subscriber("requests", publish("responses"))]
async fn respond(req: &Request) -> Response {
    println!("responding to request {}", req.id);
    Response { ok: true }
}

Mount it with plain include. With nothing else said, the reply goes out through the broker's default publish policy under the default codec; to name the reply codec or add transforms, chain .publisher(..) with a [TypedPublisher] stack over the broker's publish policy (TypedPublisher::new uses the default codec; name one with TypedPublisher::with_codec). The stack is a declaration: the runtime pairs it with the connected broker at startup.

examples/publishing.rs

// static, per-publisher: a policy stack, composed at compile time and paired with
// the connected broker at startup
b.include(respond)
    .publisher(TypedPublisher::new(MemoryPublish).transform(EnvelopeTransform));
// the default reply wiring: the broker's default policy under the default codec
b.include(validate);

Decoding of the incoming request follows the scope (the scope codec set with with_broker_codec, else the default codec); the reply codec travels on the attached stack. See Codecs.

Controlling the acknowledgement

A plain reply form always publishes and acks. Return Result<Reply, HandlerResult> instead to take control: Ok(reply) publishes and acks, Err(result) publishes nothing and the dispatcher acts on the returned HandlerResult (HandlerResult::drop() to dead-letter, HandlerResult::retry() to ask for redelivery):

examples/publishing.rs

// `Ok` publishes the reply and acks; `Err` publishes nothing and the dispatcher acts on the
// returned HandlerResult (here: drop the malformed request instead of replying).
#[subscriber("validated-requests", publish("responses"))]
async fn validate(req: &Request) -> Result<Response, HandlerResult> {
    if req.id == 0 {
        return Err(HandlerResult::drop());
    }
    Ok(Response { ok: true })
}

The Result form is detected from the written signature, so spell it out (a type alias hiding the Result is treated as a plain reply type). Like any handler, a publishing handler may declare an optional second &mut Context parameter to read app state or publish manually.

If the reply publish itself fails (broker rejected it, connection lost), the incoming message is nacked with requeue = true: the broker redelivers it instead of the reply being silently lost. Make publishing handlers idempotent under redelivery.

Publishing from inside a handler

To publish to a destination other than a single reply (a computed destination, fan-out, side effects), take the publisher as a handler parameter with Out: the pattern Out(out): Out<MemoryPublisher> binds out to a live &MemoryPublisher inside the body. The source is attached where the handler is included, and the runtime pairs it after the broker connects, so the handler cannot observe a "not connected" publisher, and the state stays free of connection-bound values.

examples/publishing.rs

use ruststream::runtime::Out;

// The publisher arrives as a parameter (the Out marker): the source is attached at the include
// site, the runtime pairs it with the connected broker at startup, and the handler always holds
// a live publisher - no registry, no erased lookup, no state plumbing.
#[subscriber("ingress")]
async fn forward(event: &Event, Out(out): Out<MemoryPublisher>) -> HandlerResult {
    let payload = JsonCodec.encode(event).expect("serializable");
    let msg = OutgoingMessage::new("egress", payload.as_ref());
    if out.publish(msg).await.is_err() {
        return HandlerResult::retry();
    }
    HandlerResult::Ack
}

The include site names the source; for the scope's own broker it is just the publish policy:

examples/publishing.rs

b.include(forward).publisher(MemoryPublish);

An Out handler included without .publisher(..) panics when the application is built (not at runtime), naming the fix: an injected publisher has no broker-side default.

The parameter composes with every subscriber form: next to a Seek parameter, on a raw handler, and on batch(..) handlers (b.include_batch(f).publisher(..) - the whole page in, per-element destinations out). On the reply forms - publish(..) / publish_raw(..) and their batch counterpart - .publisher(..) stays the reply's own attachment and the injected publisher attaches with .out(..) instead, so a gateway can answer on a fixed destination while fanning side copies out through the injection:

examples/publishing.rs

// A reply form and an injected publisher in one handler: the reply answers on the fixed
// destination while an audit copy fans out through the Out parameter.
#[subscriber("gateway-requests", publish("gateway-responses"))]
async fn gateway(req: &Request, Out(out): Out<MemoryPublisher>) -> Result<Response, HandlerResult> {
    let audit = JsonCodec
        .encode(&Event { id: req.id })
        .expect("serializable");
    if out
        .publish(OutgoingMessage::new("gateway-audit", audit.as_ref()))
        .await
        .is_err()
    {
        return Err(HandlerResult::retry());
    }
    Ok(Response { ok: true })
}

examples/publishing.rs

// the reply keeps .publisher(..) (or its default); the Out parameter attaches
// with .out(..)
b.include(gateway).out(MemoryPublish);

Publishing to a different broker

When the handler consumes one broker and publishes to another (consume Kafka, forward to Redis), wrap the target broker with .bindable() and mint a bound token before registration: the token carries the instance identity a foreign scope cannot provide, and because tokens exist before any with_broker runs, registration order does not matter - a bidirectional bridge binds both directions up front. The token is then the source at the include site (shown here with two in-memory brokers, the shape is the same for any pair):

tests/out_injection.rs

let to_other = other.bind(MemoryPublish);
let app = RustStream::new(AppInfo::new("cross", "0.1.0"))
    .with_broker(other, |b| {
        let _ = b; // the target broker may mount its own handlers here
    })
    .with_broker(ingress_broker, |b| {
        b.include(crossing).publisher(to_other);
    });

A token shares a slot with the Bindable wrapper it was minted from, and registering that same wrapper (with_broker(bindable, ..)) is what lets startup fill the slot with the connected broker - so pairing needs no lookup and cannot pick a wrong instance, and a token whose broker never registers fails fast at pairing with a clear error. The same shape works for reply publishing (.publisher(token) on a publish("dest") handler) and for the batch forms. Outside a registration, a token pairs itself once startup connected its broker: running.publisher(token) hands a sibling task its live publisher - see Running beside another server. For the first publish at startup, no token is needed at all: the scope-level b.after_startup(policy, hook) runs the hook with an already-paired publisher once subscriptions are open (see Lifespan); the publishing example's seeding rides it.

The publish pipeline

Two kinds of transform run before a message leaves the process, and they compose:

• Static PublishTransform on a TypedPublisher, added with .transform(..). Zero-cost, per-destination transforms (an envelope, a fixed content type, or stamping the delivery's trace / correlation id onto the reply). They run first, closest to the value.
• Static PublishLayer on the application, added with .publish_layer(..). Cross-cutting concerns (publish metrics, a dead-letter wrapper) applied to every published message, around the send so they can observe its result. The chain composes into a concrete type (no dyn dispatch at all), so it becomes part of the app's type. A builder usually returns impl App and never spells it; name the concrete RustStream<L, St, PublishStack<MyMiddleware, PublishIdentity>> and the pipeline shows up there, while an app with no publish_layer keeps the default PublishIdentity. Each middleware must be Clone (the pipeline is cloned into each publishing handler), and the last one added runs outermost. The default (no middleware) is a direct send. For a middleware set decided at runtime, wrap it in a PublishDynStack (the publish counterpart of DynStack) and add that.

A static PublishTransform implements apply(&mut Outgoing<'_>, &PublishContext<'_, C>); the PublishContext is a read-only view of the delivery that produced the reply (its channel, the incoming headers, and the broker's typed per-delivery context by Field key), so a transform can carry a value from the incoming message onto the reply:

examples/publishing.rs

/// A static, per-publisher transform: stamps an envelope header on every outgoing message.
struct EnvelopeTransform;

impl<C> PublishTransform<C> for EnvelopeTransform {
    fn apply(&self, out: &mut Outgoing<'_>, _cx: &ruststream::runtime::PublishContext<'_, C>) {
        out.headers_mut().insert("x-envelope", b"1".to_vec());
    }
}

A batch handler's replies skip the per-message .transform(..) stack; add a transform there with .batch_transform(..), reusing a per-message PublishTransform via for_batch(transform).

A PublishLayer implements an around/next signature, so it can short-circuit, retry, or observe (reserve "dynamic" for PublishDynLayer inside a PublishDynStack):

examples/publishing.rs

/// A static, app-wide publish layer: observes every publish, then passes it on.
#[derive(Clone)]
struct AuditPublish;

impl PublishLayer for AuditPublish {
    async fn on_publish<'a, N: PublishPipeline, P: Publisher>(
        &'a self,
        out: &'a mut Outgoing<'a>,
        next: PublishNext<'a, N, P>,
    ) -> Result<(), Box<dyn Error + Send + Sync>> {
        println!("publishing to {}", out.name());
        next.run(out).await
    }
}

Both levels compose on the application:

examples/publishing.rs

RustStream::new(AppInfo::new("publishing", "0.1.0"))
    // app-wide layer: wraps every published reply
    .publish_layer(AuditPublish)
    .with_broker(broker, |b| {
        // the first publish: runs once connected and subscribed, with the transactional
        // wiring already paired
        b.after_startup(
            TypedPublisher::with_codec(MemoryPublish, JsonCodec).transactional(),
            async move |seeder| seed_events(seeder).await.map_err(std::io::Error::other),
        );
        // static, per-publisher: a policy stack, composed at compile time and paired with
        // the connected broker at startup
        b.include(respond)
            .publisher(TypedPublisher::new(MemoryPublish).transform(EnvelopeTransform));
        // the default reply wiring: the broker's default policy under the default codec
        b.include(validate);
        b.include(forward).publisher(MemoryPublish);
        // the reply keeps .publisher(..) (or its default); the Out parameter attaches
        // with .out(..)
        b.include(gateway).out(MemoryPublish);
        // .transactional() marks the wiring; the pairing checks that the policy's live
        // publisher implements TransactionalPublisher. Without it, each reply publishes
        // independently.
        b.include_batch(confirm)
            .publisher(TypedPublisher::new(MemoryPublish).transactional());
    })

The pipeline runs on the reply path (the publish(..) form). An injected Out publisher is the attached policy's live form, used directly, so compose any per-publisher transforms into the policy at the include site with TypedPublisher::transform. The full program is examples/publishing.rs.

Batch replies and transactions

A #[subscriber(batch(..), publish(..))] handler consumes a whole decoded batch and returns the replies for it - the consume-transform-produce pattern. Ok(replies) publishes every reply to the reply name and acks the batch; Err(result) publishes nothing and settles the whole batch with result (all-or-nothing: selective per-element outcomes do not compose with a transaction):

examples/publishing.rs

/// Confirms a whole page of orders; the replies become visible atomically on commit.
#[subscriber(batch("orders"), publish("confirmations"))]
async fn confirm(orders: &[Event]) -> Result<Vec<Event>, HandlerResult> {
    if orders.is_empty() {
        return Err(HandlerResult::drop()); // nothing published, whole batch settled
    }
    Ok(orders.iter().map(|o| Event { id: o.id }).collect())
}

Mount it with include_batch, chaining the reply wiring with .publisher(..):

examples/publishing.rs

// .transactional() marks the wiring; the pairing checks that the policy's live
// publisher implements TransactionalPublisher. Without it, each reply publishes
// independently.
b.include_batch(confirm)
    .publisher(TypedPublisher::new(MemoryPublish).transactional());

With a plain TypedPublisher, each reply publishes independently; a mid-batch failure retries the whole batch, so the earlier replies may be published again on redelivery (at-least-once). Calling .transactional() on the TypedPublisher switches the wiring to one broker transaction per batch: the runtime begins a transaction, publishes every reply, commits, and only then acks the incoming batch; any failure aborts, so replies are never half-visible. The transactional requirement is enforced where the wiring is consumed: mounting it needs the policy's live publisher to implement the TransactionalPublisher capability, so a broker without transactions still fails to compile. The single-message reply forms keep taking a plain TypedPublisher stack: a one-message transaction adds broker round-trips for no atomicity gain.

Manual transactions

Outside the batch-reply path, drive a transaction by hand: begin() on the transactional wiring opens a TransactionScope that owns the transaction. Publishes go through the scope, and commit() / abort() consume it - so a commit without a begin, a second commit, or a publish after settling are compile errors, not runtime surprises:

examples/publishing.rs

/// Seeds the reference events inside one broker transaction: both records become visible
/// together on commit, or not at all. The scope owns the transaction, so a commit without a
/// begin, a second commit, or a publish after settling do not compile. The wiring arrives
/// already paired (the scope's `after_startup` hands it over live), so seeding cannot race the
/// broker connect.
async fn seed_events<P>(
    seeder: Transactional<P, JsonCodec>,
) -> Result<(), Box<dyn Error + Send + Sync>>
where
    P: TransactionalPublisher,
{
    let mut scope = seeder.begin().await?;
    scope.publish("events", &Event { id: 1 }).await?;
    scope.publish("events", &Event { id: 2 }).await?;
    scope.commit().await?;
    Ok(())
}

The scope encodes values with the publisher's codec and sends them directly: per-publisher transforms and the app-wide publish_layer middleware belong to the dispatch path (they read the originating delivery) and do not run here. Dropping an unsettled scope logs a warning and leaves the broker transaction open on that handle - always settle explicitly.

The scope is the borrowed transaction kind: it borrows the handle's single broker-side transaction, so one scope per handle is open at a time. Brokers whose transactions are client buffers rather than producer state also implement the owned kind, OwnedTransactions: every transaction() call opens an independent transaction whose buffer lives in the returned Transaction value, so any number can be open concurrently on one handle and settling one never touches another. publish buffers into the value and commit() / abort() consume it - the same settle-by-consuming discipline as the scope - while dropping one merely discards its buffer (with a warning) instead of leaving a broker transaction open. Kafka-like brokers, whose client holds exactly one transaction per producer, implement only the borrowed kind.

The owned kind has typed sugar too: on a TypedPublisher whose publisher implements OwnedTransactions, transaction() opens a TypedTransaction that owns the broker transaction and encodes with the publisher's codec - let mut txn = typed.transaction().await?;, then txn.publish("orders", &value).await?; and txn.commit().await?;. Where .transactional() + begin() gives the borrowed scope (one per handle), any number of TypedTransactions can be open on one TypedPublisher at a time.

Batch publishing

There is no direct batch-publish API on Publisher. For most brokers (NATS, Kafka) the client already coalesces writes, so a per-message publish loop achieves the same throughput. Where a broker has a genuine pipeline primitive (Redis), the broker crate exposes it as a broker-specific capability.