# Stream Builder

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

Rendered page: https://iggy.apache.org/docs/sdk/rust/stream-builder/

Source: https://github.com/apache/iggy-website/blob/main/content/docs/sdk/rust/stream-builder.mdx

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:

1) The consumer and producer operate on the same stream and topic.
2) The consumer operates on different streams and topics.
3) The producers operate on different streams and topics.
4) Add consumers dynamically at runtime.
5) 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**](https://github.com/apache/iggy/tree/master/examples/rust/src/stream-builder) of the iggy repository. Also,
if you encounter a problem with any of the examples below, please ask in the [**community discord**](https://discord.gg/apache-iggy).

The examples below connect with the `iggy`/`iggy` root credentials, which **only exist** when the server was started with:

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

Setting `IGGY_ROOT_USERNAME`/`IGGY_ROOT_PASSWORD` works too. Otherwise root gets a generated password and the examples fail with `InvalidCredentials`.

## IggyStream Builder

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

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

const IGGY_URL: &str = "iggy://iggy:iggy@localhost: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:

```rust
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:

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

const IGGY_URL: &str = "iggy://iggy:iggy@localhost: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:

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

You find sample utils to build a customized iggy client in
the [**examples folder**](https://github.com/apache/iggy/blob/master/examples/rust/src/shared/client.rs) 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.

```rust
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:

```rust
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.
        // The partition count is cached per topic and not refreshed
        // while the client stays connected.
        .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(IggyDuration::new_from_secs(1))
        // 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(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**](https://github.com/apache/iggy/blob/master/examples/rust/src/basic/producer/main.rs) 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

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

const IGGY_URL: &str = "iggy://iggy:iggy@localhost: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:

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

Notice, you find some utils to build a customized iggy client in
the [**examples folder**](https://github.com/apache/iggy/blob/master/examples/rust/src/shared/client.rs) 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](https://discord.gg/apache-iggy). See below a commented configuration example:

```rust
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)
        // The name of the consumer. Must be unique.
        .consumer_name("test_consumer".to_string())
        // The type of consumer. It can be either `Consumer` or `ConsumerGroup`.
        // ConsumerGroup is default.
        .consumer_kind(ConsumerKind::ConsumerGroup)
        // Sets the number of partitions for ConsumerKind `Consumer`.
        // Does not apply to `ConsumerGroup`.
        .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 polling from the specified timestamp.
        // - `First` - start polling from the first message in the partition.
        // - `Last` - start polling from the last message in the partition.
        // - `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(IggyDuration::new_from_secs(1))
        // 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(IggyDuration::new_from_secs(1))
        // Optionally, set a custom client side encryptor for encrypting
        // the messages' payloads. Currently only Aes256Gcm is supported.
        // 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(Arc::new(EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(&[1; 32])?)))
        .build();

    Ok(())
}
```

## 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.:

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

Where `get_my_custom_iggy_consumer_config` refers to a function that returns an `IggyConsumerConfig`
that specifies the stream and topic to consume as well the init retry or whether to create the stream and topic.

## 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.

```rust
    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**](https://discord.gg/apache-iggy).