Publishing events

Publishing is one type and two methods: Publisher, with publish for one event and publish_batch for many events of one type. Everything else (topic, partition, headers, encoding) comes from the event type you defined in Defining events.

Create a publisher and publishCopy Anchor Link

A publisher wraps a connected IggyClient and the name of one stream. It can publish events of any type, keyed by any key type, into that stream:

use std::sync::Arc;
use wakuwaku::deps::iggy::prelude::{Client, IggyClient};
use wakuwaku::iggy::Publisher;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Arc::new(IggyClient::from_connection_string(
        "iggy://iggy:iggy@localhost:8090",
    )?);
    client.connect().await?;

    // Publish two events of order 42. Both go to topic `orders`, same partition.
    let publisher = Publisher::new(client.clone(), "shop")?;
    publisher
        .publish(OrderPlaced {
            order_id: 42,
            customer: "alice".into(),
            total_cents: 12_000,
        })
        .await?;
    publisher
        .publish(PaymentReceived {
            order_id: 42,
            amount_cents: 12_000,
        })
        .await?;

    Ok(())
}

Publisher::new does not talk to the server; it only checks that the stream name is a valid Iggy identifier. publish(event) then:

1. computes the key hash with key_hash(&event.key()), using the key type's PartitionMerge hasher;
2. converts the event into its body (IntoEventBody::into_event_body) and serializes it with MessageSer::to_bytes;
3. builds one Iggy message with the body as payload and the wakuwaku headers;
4. sends it to the stream, to the topic named by the key type's SUPER_PARTITION_NAME, partitioned by the key hash as Iggy message key.

Every event of one key therefore lands in the same partition, in publish order. Each publish call sends exactly one message; to send many events in fewer requests, use publish_batch.

Publishing many events at onceCopy Anchor Link

publish_batch(events) publishes any number of events of one event type in as few requests as their keys allow:

use wakuwaku::iggy::Publisher;
use wakuwaku::iggy::error::Error;

/// Record the payments of a settlement file.
pub async fn import_settlement(publisher: &Publisher) -> Result<(), Error> {
    publisher
        .publish_batch([
            PaymentReceived { order_id: 42, amount_cents: 5_000 },
            PaymentReceived { order_id: 43, amount_cents: 1_000 },
            PaymentReceived { order_id: 42, amount_cents: 7_000 },
        ])
        .await
}

The events are grouped by key hash, and each group is sent in one request. Here that is two requests: order 42's two payments (5 000, then 7 000), then order 43's payment. The rules:

• Order per key is kept. Within a group, events keep their order in events, so a key's events land in its partition in that order. Groups are sent one after another, in the order their key first appears. Events of different keys were never ordered against each other, so nothing else is promised.
• Serialization is all or nothing. Every event is serialized and built into a message before anything is sent. If one fails, the call returns the error and nothing is published.
• A send error stops the batch partway. The groups before the failing one are already published; the failing group and the ones after it are not. The error doesn't tell you where it stopped.
• Nothing is retried, as with publish.

Use it when one operation produces many events of a type: an import, a backfill, a command that touches many keys. It saves a round trip per event, and a key's events still arrive in order. Retrying the whole call after a send error publishes the earlier groups a second time, which is harmless when handlers are idempotent (as NoEventLose already requires, see Delivery modes). For events of different types, make one call per type, awaiting each before the next if their order within a key matters.

Sharing a publisherCopy Anchor Link

Publisher is Clone, and clones are cheap: they share the same IggyClient. Create it once at startup and register it in the ServiceBuilder so components can take it as their dependency:

use wakuwaku::ServiceCreation;
use wakuwaku::iggy::Publisher;
use wakuwaku::iggy::error::Error;

/// A component that publishes events. `Publisher` is its dependency.
pub struct Checkout {
    publisher: Publisher,
}

impl ServiceCreation for Checkout {
    type Dep = Publisher;
    fn create_service(publisher: Publisher) -> Self {
        Self { publisher }
    }
}

impl Checkout {
    pub async fn place_order(&self, order_id: u64, customer: String) -> Result<(), Error> {
        // ... store the order, then announce it.
        self.publisher
            .publish(OrderPlaced {
                order_id,
                customer,
                total_cents: 12_000,
            })
            .await
    }
}

let publisher = Publisher::new(client.clone(), "shop")?;
let services = ServiceBuilder::new(client).push(publisher);

Headers on the wireCopy Anchor Link

Every message carries three user headers, all raw bytes. The consumer reads them back to route, order and reduce the record without decoding its body. You only need this table when another program (a different language, a bridge, a replay tool) writes events that a wakuwaku consumer should read; the constants are in wakuwaku::iggy::events::headers.

The wire values inside wakuwaku-props are:

A foreign producer must also partition the same way: use the 8 little-endian bytes of the key hash as the Iggy message key, so the Iggy client picks the partition exactly as Publisher does.

What must exist on the serverCopy Anchor Link

wakuwaku creates as little as possible on the Iggy server:

publish to a missing topic fails with an Iggy error, and a consumer refuses to start without its main topic (Error::TopicNotFound). Create both as part of deployment, for example with a small idempotent step at startup using the Iggy client re-exported as wakuwaku::deps::iggy:

use wakuwaku::deps::iggy::prelude::{
    Identifier, IggyClient, IggyError, StreamClient, TopicClient, TopicCreateOptions,
};
use wakuwaku::iggy::PartitionKey;

/// Create the stream and the main topic if they don't exist yet.
pub async fn ensure_topology(client: &IggyClient) -> Result<(), IggyError> {
    let stream = Identifier::named("shop")?;
    if client.get_stream(&stream).await?.is_none() {
        client.create_stream("shop").await?;
    }

    let topic = OrderId::SUPER_PARTITION_NAME.0; // "orders"
    if client.get_topic(&stream, &Identifier::named(topic)?).await?.is_none() {
        // One partition, consumed as `vec![0]`. See "Choose the partition count up front".
        let options = TopicCreateOptions {
            partitions_count: Some(1),
            ..Default::default()
        };
        client.create_topic(&stream, topic, &options).await?;
    }
    Ok(())
}

Reading the topic name from SUPER_PARTITION_NAME keeps the setup and the event types from drifting apart. If partitions_count is left unset, the Iggy SDK's default is also a single partition; setting it keeps the count visible next to the consumers' partition lists.

ErrorsCopy Anchor Link

Publisher::new, publish and publish_batch return wakuwaku::iggy::error::Error. The variants you can get from publishing are:

Sends are not retried. When publish returns an error, the event was not written as far as wakuwaku knows, and it is up to you whether to retry, fail the request, or surface the error (for publish_batch, see what a send error leaves behind). Publishing writes to Iggy and nothing else; if the event must be published exactly when a database transaction commits, that coordination is your code's responsibility.

Next stepsCopy Anchor Link