The stream builder, for event-driven applications that would otherwise repeat producer and consumer setup.

In the previous section, the high level SDK introduced the connection string to simplify the client configuration and indeed, it does provide a simple way to connect to the server. However, when you write an event based application, you typically encounter one or more of the following scenarios:

The consumer and producer operate on the same stream and topic.
The consumer operates on different streams and topics.
The producers operate on different streams and topics.
Add consumers dynamically at runtime.
Add producers dynamically at runtime.

The stream builder provides a convenient way to create the iggy client, producer and consumer for these use cases. All source code examples are located in the examples folder of the iggy repository. Also, if you encounter a problem with any of the examples below, please ask in the community discord.

The examples below use iggy/iggy. For a new local development instance, with no IGGY_ROOT_USERNAME or IGGY_ROOT_PASSWORD overrides, use:

cargo run --bin iggy-server -- --fresh --with-default-root-credentials

Environment credentials take precedence over the flag, and bootstrap settings do not replace recovered credentials. --fresh deletes this replica's local data; use disposable development data. A new standalone server without explicit credentials generates a root password; a new cluster requires explicit credentials.

Use the SDK and server from the same checkout for unreleased changes. The examples import PrintEventConsumer from the local iggy_examples crate and can run as the targets in examples/rust. In your own application, use the implementation shown below instead of that import.

IggyStream Builder

In the first case, the IggyStream Builder offers a fast and efficient way to get you started:

use iggy::prelude::*;
use iggy_examples::shared::stream::PrintEventConsumer;
use std::str::FromStr;
use tokio::sync::oneshot;

const IGGY_URL: &str = "iggy://iggy:iggy@127.0.0.1:8090";

#[tokio::main]
async fn main() -> Result<(), IggyError> {
    let stream_config = IggyStreamConfig::default();
    let (client, producer, mut consumer) =
        IggyStream::with_client_from_connection_string(IGGY_URL, &stream_config).await?;

    let (sender, receiver) = oneshot::channel();
    tokio::spawn(async move {
        match consumer.consume_messages(&PrintEventConsumer {}, receiver).await {
            Ok(_) => {}
            Err(err) => eprintln!("Failed to consume messages: {err}"),
        }
    });

    producer
        .send_one(IggyMessage::from_str("Hello World")?)
        .await?;

    // Wait a bit for all messages to arrive.
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;

    sender.send(()).expect("Failed to send shutdown signal");
    client.delete_stream(stream_config.stream_id()).await?;
    client.shutdown().await?;

    Ok(())
}

As you can see, the IggyStream builder is used to create the iggy client, producer and consumer. Here, we use the default configuration to get started quickly, but you can always customize the configuration (IggyStreamConfig::from_stream_topic or the two per-side configs) to fit your requirements. If you already have a connected client, use IggyStream::build(&client, &stream_config) instead, which returns just the producer and consumer.

The consumer is bound as mut: consume_messages (and the underlying message stream) requires mutable access.

The consume_messages method is not part of the IggyConsumer. It comes from the IggyConsumerMessageExt trait, which is re-exported by iggy::prelude. The PrintEventConsumer implements the MessageConsumer trait and is invoked to process each incoming message. A simple implementation is shown below:

use iggy::clients::consumer::ReceivedMessage;
use iggy::consumer_ext::MessageConsumer;
use iggy::prelude::IggyError;

#[derive(Debug)]
pub struct PrintEventConsumer {}

impl MessageConsumer for PrintEventConsumer {
    async fn consume(&self, message: ReceivedMessage) -> Result<(), IggyError> {
        // Extract message payload as raw bytes
        let raw_message = message.message.payload.as_ref();
        // Convert raw bytes into string
        let payload = String::from_utf8_lossy(raw_message);
        // Print message
        println!(
            "Message received: {} at offset: {} in partition ID: {}",
            payload, message.message.header.offset, message.partition_id
        );

        Ok(())
    }
}

IggyStreamProducer Builder

When you implement the producer side, you can use the IggyStreamProducer to get started:

use iggy::prelude::*;
use std::str::FromStr;

const IGGY_URL: &str = "iggy://iggy:iggy@127.0.0.1:8090";

#[tokio::main]
async fn main() -> Result<(), IggyError> {
    println!("Build iggy client and producer");
    // For customization, use the `new` or `from_stream_topic` constructor
    let config = IggyProducerConfig::default();
    let (client, producer) = IggyStreamProducer::with_client_from_url(IGGY_URL, &config).await?;

    println!("Send 3 test messages...");
    producer
        .send_one(IggyMessage::from_str("Hello World")?)
        .await?;
    producer
        .send_one(IggyMessage::from_str("Hola Iggy")?)
        .await?;
    producer
        .send_one(IggyMessage::from_str("Hi Apache")?)
        .await?;

    // Wait a bit for all messages to arrive.
    tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
    println!("Stop the message stream and shutdown iggy client");
    client.shutdown().await?;
    Ok(())
}

If you already have a custom iggy client, you can use that one by replacing the with_client_from_url constructor with the following:

    let producer = IggyStreamProducer::build(&client, &config).await?;

You find sample utils to build a customized iggy client in the examples folder of the iggy repository.

Producer configuration

The IggyProducerConfig gives you a way to configure the producer in sufficient detail. Please note, if you have questions about any of those settings, please ask in the community discord. For basic customization, the from_stream_topic constructor lets you set a custom stream and topic name as well as the maximum batch length and linger time between sends. The producer resolves destinations by name; keep the corresponding identifier fields consistent with those names.

use iggy::prelude::*;
use std::str::FromStr;

let config = IggyProducerConfig::from_stream_topic(
    "test_stream",
    "test_topic",
    100,
    IggyDuration::from_str("5ms").unwrap(),
)?;

The remaining configuration fields are set to default values applicable to the most common use case. This should be sufficient for a simple application, or proof of concept without getting lost in details. However, for more complex applications, you might want to configure more details and for that, see below a commented example of the full producer configuration:

use iggy::prelude::*;
use std::str::FromStr;

#[tokio::main]
async fn main() -> Result<(), IggyError> {
    let stream = "test_stream";
    let topic = "test_topic";

    // The builder simplifies the IggyProducer configuration.
    let config = IggyProducerConfig::builder()
        // Set the stream identifier and name.
        .stream_id(Identifier::from_str_value(stream)?)
        .stream_name(stream)
        // Set the topic identifier and name
        .topic_id(Identifier::from_str_value(topic)?)
        .topic_name(topic)
        // Sets the number of partitions to create for the topic.
        // The more clients are reading concurrently,
        // the more partitions you should create.
        // i.e. if you have 10 clients, you should create 10 partitions
        .topic_partitions_count(10)
        // The max number of messages to send in a batch.
        // The greater the batch length, the higher the bulk throughput.
        // Note, there is a tradeoff between batch length and latency,
        // so you want to benchmark your setup.
        .batch_length(100)
        // Sets the interval between sending the messages.
        // Affects latency so you want to benchmark.
        .linger_time(IggyDuration::from_str("5ms").unwrap())
        // `Partitioning` specifies to which partition the messages
        // should be sent.
        // It has the following options:
        // - `Balanced` - the client picks the next partition with a
        //    local round-robin cursor.
        // - `PartitionId` - the partition ID is provided by the client.
        // - `MessagesKey` - the client hashes the provided messages key
        //    to a partition ID.
        // Binary clients cache the partition count for their lifetime;
        // reconnecting does not refresh it.
        .partitioning(Partitioning::balanced())
        // Sets the retry policy (maximum number of retries and interval)
        // in case of messages sending failure.
        // The error can be related either to disconnecting from the server
        // or to the server rejecting the messages.
        // Default is 3 retries with 1 second interval between them.
        .send_retries_count(3)
        .send_retries_interval(NonZeroIggyDuration::ONE_SECOND)
        // Optionally, set a custom client side encryptor for encrypting
        // the messages' payloads. Currently only Aes256Gcm is supported.
        // Note, this is independent of server side encryption.
        // You can add client encryption, server encryption, or both.
        // .encryptor(std::sync::Arc::new(EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(&[1; 32])?)))
        .build();
    Ok(())
}

Note, when your requirements exceed this configuration, you can still use the underlying low level SDK for fine grained control over every detail of the producer.

IggyStreamConsumer Builder

When you have an existing server with multiple producers, you may have to implement a consumer for a specific stream or topic. In some cases you may already have a configured iggy client, in others you may need to create one. In either case, you can use the IggyStreamConsumer to create the consumer.

Build consumer and client

use iggy::prelude::*;
use iggy_examples::shared::stream::PrintEventConsumer;
use tokio::sync::oneshot;

const IGGY_URL: &str = "iggy://iggy:iggy@127.0.0.1:8090";

#[tokio::main]
async fn main() -> Result<(), IggyError> {
    let config = IggyConsumerConfig::default();
    let (client, mut consumer) =
        IggyStreamConsumer::with_client_from_url(IGGY_URL, &config).await?;

    let (tx, rx) = oneshot::channel();
    tokio::spawn(async move {
        match consumer.consume_messages(&PrintEventConsumer {}, rx).await {
            Ok(_) => {}
            Err(err) => eprintln!("Failed to consume messages: {err}"),
        }
    });

    // Wait a bit for all messages to arrive.
    tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
    println!("Stop the message stream and shutdown iggy client");
    tx.send(()).expect("Failed to send shutdown signal");
    client.shutdown().await?;
    Ok(())
}

Similar to the previous StreamBuilder, the IggyConsumerMessageExt is used to process messages using the PrintEventConsumer implementation of the MessageConsumer trait.

Build consumer from an existing client

In the event that you have an existing iggy client, you can use the IggyStreamConsumer::build method to create the consumer. To do so, just replace the with_client_from_url with the following:

    let mut consumer = IggyStreamConsumer::build(&client, &config).await?;

Notice, you find some utils to build a customized iggy client in the examples folder of the iggy repository.

Consumer configuration

The default configuration only exists to get you started quickly i.e. to test out an idea, draft an proof of concept or similar. In practice, you often want fine grained control over the configuration. The IggyConsumerConfig gives you a way to configure the consumer in sufficient detail without distracting with low level details. Please note, if you have questions about any of those settings, please ask in the community discord. See below a commented configuration example:

use iggy::prelude::*;
use std::str::FromStr;

#[tokio::main]
async fn main() -> Result<(), IggyError> {
    let stream = "test_stream";
    let topic = "test_topic";

    let config = IggyConsumerConfig::builder()
        // Set the stream identifier and name.
        .stream_id(Identifier::from_str_value(stream)?)
        .stream_name(stream)
        // Set the topic identifier and name
        .topic_id(Identifier::from_str_value(topic)?)
        .topic_name(topic)
        // The auto-commit configuration for storing the message offset.
        // See: https://github.com/apache/iggy/blob/master/core/sdk/src/clients/consumer.rs
        .auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages))
        // The max number of messages to poll in a batch.
        // The greater the batch length, the higher the bulk throughput.
        // Note, there is a tradeoff between batch length and latency,
        // so you want to benchmark your configuration.
        .batch_length(100)
        // Create the stream if it doesn't exist.
        .create_stream_if_not_exists(true)
        // Create the topic if it doesn't exist.
        .create_topic_if_not_exists(true)
        // Members of the same consumer group use the same name.
        .consumer_name("test_consumer".to_string())
        // The type of consumer. It can be either `Consumer` or `ConsumerGroup`.
        // ConsumerGroup is default.
        .consumer_kind(ConsumerKind::ConsumerGroup)
        // Partition count when creating a topic. Also used as the partition
        // ID for an ordinary consumer; group assignment ignores this value.
        .partitions_count(1)
        // The polling interval for messages.
        .polling_interval(IggyDuration::from_str("5ms").unwrap())
        // `PollingStrategy` specifies from where to start polling messages.
        // It has the following kinds:
        // - `Offset` - start polling from the specified offset.
        // - `Timestamp` - start at or after the broker append timestamp.
        // - `First` - start polling from the first message in the partition.
        // - `Last` - read up to batch_length messages ending at the committed offset.
        // - `Next` - start polling from the next message after the
        // last polled message based on the stored consumer offset.
        .polling_strategy(PollingStrategy::last())
        // Sets the polling retry interval in case of server disconnection.
        .polling_retry_interval(NonZeroIggyDuration::ONE_SECOND)
        // Sets the number of retries and the interval when initializing
        // the consumer if the stream or topic is not found.
        // Useful when the stream or topic is created dynamically
        // by the producer.
        // The default is 5 retries with a 3 second interval,
        // so init retry is enabled out of the box.
        .init_retries(5)
        .init_interval(NonZeroIggyDuration::ONE_SECOND)
        // Optionally, configure Aes256Gcm payload and user-header decryption.
        // Replace PollingMessages auto-commit with a compatible mode first.
        // Key must be identical to the one used by the producer;
        // thus ensure secure key exchange.
        // Note, this is independent of server side encryption.
        // you can add client encryption, server encryption, or both.
        // .encryptor(std::sync::Arc::new(EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(&[1; 32])?)))
        .build();

    Ok(())
}

partitions_count also selects the partition ID for an ordinary consumer. When it creates a topic with n partitions, the valid IDs are 0..n-1, so using the same n as an ordinary consumer's ID fails. The sample uses a consumer group. For an ordinary consumer, create the topic separately and use client.consumer(name, stream, topic, partition_id) to select a valid partition.

Add consumers dynamically at runtime.

When you create consumers on demand at application runtime, the consumer has to survive the window where the stream or topic does not exist yet. The init_retries and init_interval settings cover exactly that, and they are enabled by default (5 retries with a 3 second interval), so a consumer created shortly before its producer usually just works.

However, when you cannot guarantee that the stream or topic ever gets created by someone else, you have to set create_stream_if_not_exists and create_topic_if_not_exists to true in the IggyConsumerConfig (both are false by default) to ensure the consumer creates them and starts correctly. Other than that, you just call the IggyStreamConsumer constructors as before i.e.:

    let (client, mut consumer) =
        IggyStreamConsumer::with_client_from_url(IGGY_URL, &config).await?;

Here config is the IggyConsumerConfig built above, which selects the stream and topic, initialization retries, and automatic creation behavior.

Add producers dynamically at runtime.

Adding new producers to the iggy server at runtime is as simple as creating a new config and calling the IggyStreamProducer::build method (or with_client_from_url when you need a new client too). Unlike the consumer, the producer always creates missing streams or topics by default. The IggyProducerConfig has a convenient constructor to create a new configuration from just the stream, topic, batch length and linger time. See the example below.

    let config = IggyProducerConfig::from_stream_topic(
        "new_stream",
        "new_topic",
        100,
        IggyDuration::from_str("5ms").unwrap(),
    )?;

    let (client, producer) =
        IggyStreamProducer::with_client_from_url(IGGY_URL, &config).await?;

If you encounter a problem with any of the examples show on this page, please ask in the community discord.

Stream Builder

On this page