High-level SDK

If you've read through the getting started guide, you might have noticed that it's quite verbose and requires a lot of boilerplate code to get started. This is where the High-level SDK comes in, as it does provide a more user-friendly interface to interact with the Iggy API for both, producer and consumer. Let's consider the following features:

• Automatically creating & joining the consumer groups
• Committing the offset depending on the particular mode (e.g. in the background based on some interval, after polling N messages etc.)
• Batching the messages, whether it's about producing or consuming
• Processing the messages as if the stream was an async iterator
• Reusing the same client for both, producing and consuming on the same topic without repeating the configuration
• And more...

Connection string

Instead of providing the configuration for the client, you can use the connection string. It's a string that contains all the necessary information to connect to the Iggy API, and it works with all four transports:

iggy://iggy:iggy@localhost:8090             (TCP, default)
iggy+tcp://iggy:iggy@localhost:8090         (TCP, explicit)
iggy+quic://iggy:iggy@localhost:8080        (QUIC)
iggy+http://iggy:iggy@localhost:3000        (HTTP)
iggy+ws://iggy:iggy@localhost:8092          (WebSocket)

Which can be used to create the client like this:

let client = IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;

Options are appended as query parameters. For TCP:

iggy://iggy:iggy@localhost:8090?tls=true&tls_domain=test.com&reconnection_retries=5&reconnection_interval=5s&reestablish_after=10s&heartbeat_interval=3s&nodelay=true

The option keys differ per transport (for example QUIC uses reconnection_max_retries and reconnection_reestablish_after where TCP uses reconnection_retries and reestablish_after). Unknown keys are rejected as hard errors. See Connection Strings for the full per-transport option tables and defaults.

Unless you need to provide a specific implementation of the client-side Encryptor, Partitioner or adjust some other settings, you should be good to go with the connection string. On the other hand, you can always make use of IggyClientBuilder::from_connection_string() to extend the options on top of the provided connection string.

The IggyClient internally wraps its transport in an IggyRwLock, so a single instance is cheap to clone and can be shared across multiple tasks. Typically, it's a good idea to create a separate connection for producing and consuming, but it's not a requirement.

Producer

The producer is a high-level abstraction that allows you to send messages to the topic. It's quite simple to use and doesn't require you to handle the offsets, partitions or any other low-level details. To begin with, you can simply invoke client.producer() to get the IggyProducerBuilder allowing you to configure the producer. Let's take a look at the basic example:

let producer = client
    .producer("my-stream", "my-topic")?
    .direct(
        DirectConfig::builder()
            .batch_length(1000)
            .linger_time(IggyDuration::from_str("5ms")?)
            .build(),
    )
    .partitioning(Partitioning::balanced())
    .build();
producer.init().await?;

The code above will result in creating the producer that will try to send the messages in batches of 1000 every 5 milliseconds. You can choose between the direct (an instant producer) or the background (which will send the messages in the background by buffering them). The partitioning is set to balanced which means that the producer will try to distribute the messages evenly across all the partitions. The init() method is used to ensure that the producer is ready to send the messages by validating the existence of the stream, topic etc.

Finally, you can use the send() method to send the messages to the topic. The producer never needs to be a mutable binding: both init() and send() take &self, so a plain let producer is enough for its whole life. Here's how you can send the messages:

let messages = vec![IggyMessage::from_str("hello")?, IggyMessage::from_str("world")?];
producer.send(messages).await?;

Of course, you can provide the message headers, custom binary serialization etc. as it's the same IggyMessage type as the one used in the low-level API (use IggyMessage::builder() to set headers or a custom ID). The producer will take care of the rest, including the retries, partitioning, batching etc.

Besides send(), the producer also offers send_one(message), send_with_partitioning(messages, partitioning) to override the partitioning per call, and send_to(stream, topic, messages, partitioning) to target a different stream and topic.

Background producer configuration

The background mode buffers messages and sends them from shard workers, trading latency for throughput and backpressure control. BackgroundConfig::builder() exposes:

batch_size limits bytes while batch_length limits the message count. Whichever limit fires first (including linger_time) triggers the flush.

Call producer.shutdown().await to flush the background buffer before exiting. Dropping the producer without it silently discards unflushed messages.

Send confirmations

Every send method returns a SendMessagesResponse carrying the commit confirmations of the chunks the send was split into. The list is empty whenever the server sends no confirmation payload, and always for a background producer, which hands the messages to a dispatcher and returns before the send happens. Branch on confirmations.is_empty() instead of indexing. Delivery is at-least-once: a retried chunk may already have been committed by an earlier attempt, so an offset in a confirmation never implies uniqueness.

Consumer

The consumer is a high-level abstraction that allows you to receive the messages from the topic. It's quite simple to use and doesn't require you to handle the offsets, partitions or any other low-level details. To begin with, you can simply invoke client.consumer() to get the IggyConsumerBuilder allowing you to configure the consumer. Let's take a look at the basic example:

let mut consumer = client
    .consumer_group("my-consumer-group", "my-stream", "my-topic")?
    .auto_commit(AutoCommit::IntervalOrWhen(
        IggyDuration::from_str("1s")?,
        AutoCommitWhen::ConsumingAllMessages,
    ))
    .create_consumer_group_if_not_exists()
    .auto_join_consumer_group()
    .polling_strategy(PollingStrategy::next())
    .poll_interval(IggyDuration::from_str("1ms")?)
    .batch_length(1000)
    .build();

The code above will result in creating the consumer that will try to consume the messages in batches of 1000 every 1 millisecond. The auto-commit is set to commit the offset every second or when all the messages are consumed (fetched). The polling strategy is set to next which means that the consumer will try to consume the next available message from the partition currently assigned to the consumer group (you can also invoke a regular consumer() builder if you do not plan to use the consumer groups). The build() method is used to create the consumer.

Finally, you can use the next() method to receive the messages from the topic. Unlike the producer, the consumer must stay a mutable binding for its whole life: init() takes &mut self, and next() comes from the futures Stream implementation which also requires mutable access. The init() is used to ensure that the consumer is ready to receive the messages by validating the existence of the stream, topic, consumer group etc. Here's how you can consume the messages:

consumer.init().await?;

// Start consuming the messages
while let Some(message) = consumer.next().await {
    match message {
        Ok(received) => {
            // received.message is the IggyMessage
        }
        Err(error) => eprintln!("Error while receiving message: {error}"),
    }
}

In order to use the async iterator extension, add futures-util dependency and import futures_util::StreamExt.

Consumer builder options

Defaults and lesser-known knobs on IggyConsumerBuilder:

Auto-commit matrix

AutoCommit decides when the consumer offset is stored on the server:

AutoCommitWhen triggers fire while receiving: PollingMessages, ConsumingAllMessages, ConsumingEachMessage, ConsumingEveryNthMessage(n). AutoCommitAfter triggers fire after processing: ConsumingAllMessages, ConsumingEachMessage, ConsumingEveryNthMessage(n).

The After and IntervalOrAfter variants only work with consume_messages() from the IggyConsumerMessageExt trait (see Stream Builder). The plain next() loop cannot observe when your processing finished.

Raw requests

For custom commands the IggyClient exposes two escape hatches below the typed API:

• send_binary_request(code, payload) sends a raw binary command and returns the raw response. Binary transports only. On HTTP it returns FeatureUnavailable. Login and logout codes are rejected with InvalidCommand, use login_user/logout_user so the SDK session state stays correct. Custom codes are forwarded to the server, which decides whether it implements them.
• send_http_request(method, path, body) invokes an arbitrary HTTP endpoint and returns the raw response body. HTTP transport only. Binary transports return FeatureUnavailable.