KinesisPublish

KinesisPublish, Publish in the prelude, writes each record with one PutRecord; it is the broker's default, so a reply needs no policy.

A mount site names it wherever a registration publishes: .out_reply(Publish::default()) for a reply, .out(Slot, Publish::default()) for a slot, .out_retry(Publish::default()) for retry copies. after_startup(Publish::default(), ..) hands the same publisher to a startup hook. The policy holds no connection; the runtime pairs it with the connected broker at startup into a KinesisPublisher. A registration that names no policy for its reply publishes through this one.

A record goes to the stream its destination names: the reply type's #[outgoing(name)], a reply("stream") clause, or to("stream") on the publish. A stream is named by its name. A publish returns once the service has stored the record, or with the KinesisError::Publish it reported. The record hands the service the bytes the codec wrote, with the message's headers in an envelope in front of them when it has any. A stream takes a record of up to 1 MiB, data and partition key together, unless its owner raised the limit, which goes up to 10 MiB.

Each record carries a partition key, which picks its shard and its order. A call names it with the partition_key step, and a mount site with KinesisPublishSettings.

Example

The handler replies with a confirmation through the default policy and hands a shipment to a slot that names the policy.

Macros
Manual

examples/kinesis_publishing_publish.rs
/// Confirms each order with its return value and hands it to the warehouse through a slot.
#[subscriber(KinesisStream::new("orders"), reply)]
async fn confirm(order: &Order, Out(shipments): Out<impl Publisher, Shipments>) -> Confirmation {
    let shipment = Shipment { order: order.id };
    if let Err(err) = shipments.message(&shipment).publish().await {
        tracing::warn!(subscription = "orders", order.id, error = %err, "shipment not published");
    }
    Confirmation { order: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        // The reply names no policy and takes the broker's default; the slot names one.
        b.include(confirm)
            .out(Shipments, Publish::default())
            .build();
    })
}

examples/manual/kinesis_publishing_publish.rs
/// Confirms each order with its return value and hands it to the warehouse through a slot.
struct Confirm;

impl<Entry> Handle<Order, Confirmation, Outs<(Entry,)>> for Confirm
where
    Entry: OutEntry<Shipments, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<Confirmation, HandlerOutcome> {
        let shipment = Shipment { order: order.id };
        if let Err(err) = outs.get(Shipments).message(&shipment).publish().await {
            tracing::warn!(subscription = "orders", order.id, error = %err, "shipment not published");
        }
        Ok(Confirmation { order: order.id })
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        // The reply names no policy and takes the broker's default; the slot names one.
        b.include(
            subscriber(KinesisStream::new("orders"), Confirm)
                .reply()
                .build(),
        )
        .out(Shipments, Publish::default())
        .build();
    })
}

Example​

Example