Website | Getting started | Documentation | Examples | Discord
Official Rust client SDK for Apache Iggy, the persistent message streaming platform written in Rust. The SDK ships a low-level transport client (QUIC, TCP, HTTP, WebSocket) for direct command access and a high-level producer/consumer API with batching, consumer groups, and auto-commit.
IggyClient API across all four.iggy:// (TCP default), iggy+tcp://, iggy+quic://, iggy+http://, iggy+ws://. Binary transports apply credentials on connect(); HTTP requires an explicit login. Option keys and reconnection support differ by transport.IggyClient: producer(stream, topic), consumer(name, stream, topic, partition), and consumer_group(name, stream, topic).direct (awaited send) and background (buffered with parallel shard workers using OrderedSharding or BalancedSharding). Configurable batch length / size and linger time.balanced, messages_key, or explicit partition_id. Custom Partitioner is pluggable.Stream. Polling strategies: next, offset, timestamp, first, last.Interval, When, After, IntervalOrWhen, IntervalOrAfter, or disabled.IggyStream, IggyStreamProducer, IggyStreamConsumer) for declarative producer + consumer setup on shared or separate stream/topic.HeaderKey / HeaderValue), client-side AES-256-GCM encryption (via Aes256GcmEncryptor), topic compression metadata (None and Gzip; no runtime compression yet), server-honored message expiry, and server-side deduplication.Run from your application crate. Use a release compatible with your server; for unreleased changes, build the SDK and server from the same source checkout.
Cluster auto-commit polling requires servers that support consumer session attachment and primary poll routing (binary commands 14, 103 and 104). Rust manual and interval offset writes also require command 123. It discovers the primary while allowing final commits for partitions awaiting handoff; stores and deletes retain their existing wire formats and deduplication keys. Pause binary auto-commit consumers during this upgrade, upgrade every server first, then upgrade the SDKs and restart the consumers so they join their groups again. Older SDKs can lose group membership when a backup refuses an offset commit; the new SDK does not fall back to that path on an older server. HTTP clients use server-side forwarding and need no new routing commands.
cargo add iggy
All four transports are included; this crate declares no optional Cargo features.
Start a source server from the repository root in a separate terminal:
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
Use disposable replica data with --fresh. Environment credentials override the flag, and recovered credentials are not replaced. The sample expects iggy/iggy and requires iggy, Tokio and futures-util in the application.
use std::error::Error; use std::str::FromStr; use futures_util::StreamExt; use iggy::prelude::*; const STREAM: &str = "telemetry"; const TOPIC: &str = "device-events"; const CONSUMER_GROUP: &str = "telemetry-ingester"; #[tokio::main] async fn main() -> Result<(), Box<dyn Error>> { let client = IggyClient::from_connection_string( "iggy://iggy:iggy@localhost:8090", )?; client.connect().await?; let producer = client .producer(STREAM, TOPIC)? .direct( DirectConfig::builder() .batch_length(1000) .linger_time(IggyDuration::from_str("1ms")?) .build(), ) .partitioning(Partitioning::balanced()) .build(); producer.init().await?; producer .send(vec![IggyMessage::from_str("Hello Apache Iggy")?]) .await?; let mut consumer = client .consumer_group(CONSUMER_GROUP, STREAM, TOPIC)? .auto_commit(AutoCommit::IntervalOrWhen( NonZeroIggyDuration::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(); consumer.init().await?; while let Some(message) = consumer.next().await { match message { Ok(message) => { let payload = std::str::from_utf8(&message.message.payload) .unwrap_or("<non-utf8>"); println!( "offset={} partition={} current_offset={} payload={payload}", message.message.header.offset, message.partition_id, message.current_offset, ); if let Some(headers) = message.message.user_headers_map()? { for (key, value) in headers { println!(" header {key} = {value:?}"); } } } Err(error) => eprintln!("poll error: {error}"), } } Ok(()) }
For lower-level control over individual commands (login, stream/topic management, raw send, polling by offset or timestamp), use the transport-specific clients directly. See the examples and the Rust SDK docs.
For IggyConsumerConfig, partitions_count controls topic creation only. An ordinary consumer uses partition 0 unless the builder‘s partition_id or the config’s with_partition_id selects another partition. Code that previously used partitions_count to select an existing partition must set partition_id explicitly. Consumer-group assignment ignores partition_id.
Stable releases follow semver (x.y.z). Edge releases (x.y.z-edge.N) are cut from master between stable versions and may include unreleased fixes; pin to a stable version for production.
iggy_common: shared types and traits.iggy_binary_protocol: wire protocol codec.iggy-cli: command-line tool, cargo install iggy-cli.Licensed under the Apache License, Version 2.0. See LICENSE.