EosPipeline

EosPipeline::publish sends a record into the open window by hand, with the source coordinates the Source key gives.

EosPipeline is the live form of EosPublish, so a slot bound to that policy holds one. A handler reaches it through the slot's Borrow<EosPipeline> bound, Out<impl Borrow<EosPipeline>>, and the manual form names it as the slot's wired value. The delivery's coordinates are a SourceOffset, read with Ctx<Source> (context::keys::Source). publish(&source, record, options) adds the record to the window and pairs it with that offset; None keeps the producer's own partition choice. A handler publishes as many records as it needs, then acks: the settled delivery is what lets the window commit. A failed publish poisons the window, which aborts and is delivered again, so the handler answers it with retry(). The record is an OutgoingMessage the handler encodes itself. id returns the pipeline id.

Example

A cart is split into one record per item, and every record commits with the cart's offset.

Macros
Manual

examples/kafka_eos_pipeline.rs
/// Splits a cart into one record per item, all committed with the cart's offset.
#[subscriber(KafkaTopic::new("carts").commit(Commit::Transactional("checkout-1".into())))]
async fn split(
    cart: &Cart,
    Ctx(source): Ctx<Source>,
    Out(window): Out<impl Borrow<EosPipeline>>,
) -> HandlerOutcome {
    // The slot holds the paired pipeline, and `Borrow` is the bound that reaches it.
    let pipeline: &EosPipeline = (**window).borrow();
    for item in &cart.items {
        let line = CartLine {
            cart_id: cart.id,
            item: item.clone(),
        };
        let Ok(payload) = serde_json::to_vec(&line) else {
            return HandlerOutcome::drop();
        };
        let record = OutgoingMessage::new("cart-lines", payload.as_slice());
        // A failed publish aborts the window, which redelivers the cart.
        if pipeline.publish(&source, record, None).await.is_err() {
            return HandlerOutcome::retry();
        }
    }
    HandlerOutcome::ack()
}

examples/manual/kafka_eos_pipeline.rs
/// Splits a cart into one record per item, all committed with the cart's offset.
struct Split;

impl<W> Handle<Cart, (), Outs<(W,)>, KafkaContext> for Split
where
    W: OutEntry<DefaultSlot, Wire = EosPipeline>,
{
    async fn handle(
        &self,
        cart: &Cart,
        outs: &Outs<(W,)>,
        ctx: &mut Context<'_, KafkaContext>,
    ) -> Result<(), HandlerOutcome> {
        let pipeline: &EosPipeline = outs.get(DefaultSlot);
        let source = ctx.context(Source);
        for item in &cart.items {
            let line = CartLine {
                cart_id: cart.id,
                item: item.clone(),
            };
            let Ok(payload) = serde_json::to_vec(&line) else {
                return Err(HandlerOutcome::drop());
            };
            let record = OutgoingMessage::new("cart-lines", payload.as_slice());
            // A failed publish aborts the window, which redelivers the cart.
            if pipeline.publish(&source, record, None).await.is_err() {
                return Err(HandlerOutcome::retry());
            }
        }
        Ok(())
    }
}

Macros
Manual

examples/kafka_eos_pipeline.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("checkout-svc"),
        |b| {
            b.include(split)
                .out(DefaultSlot, EosPublish::new("checkout-1"))
                .build();
        },
    )
}

examples/manual/kafka_eos_pipeline.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("checkout-svc"),
        |b| {
            let carts = subscriber("carts", Split).map_source(|_| {
                KafkaTopic::new("carts").commit(Commit::Transactional("checkout-1".into()))
            });
            b.include(carts.build())
                .out(DefaultSlot, EosPublish::new("checkout-1"))
                .build();
        },
    )
}

Example​

Example