The Apache Iggy C# SDK provides a comprehensive client library for interacting with Iggy message streaming servers. It offers a modern, async-first API with support for multiple transport protocols and comprehensive message streaming capabilities.
Install the NuGet package:
dotnet add package Apache.Iggy --version 0.9.0
The examples target server 0.9.0. For source builds, use the server and SDK from the same checkout. The SDK targets .NET 8 and .NET 10; repository examples require .NET 10. 0.9.0 includes the independent message and consumer-offset durability options.
Cluster auto-commit polling over TCP/TLS keeps group membership on the coordinator and uses separate connections to partition primaries. It requires server support for binary commands 14, 103 and 104. Pause binary auto-commit consumers for the whole upgrade: upgrade every server first, then the SDKs, and restart consumers so they rejoin their groups. Older SDKs can lose membership when a backup refuses an offset commit; the new SDK does not fall back to legacy polling.
The SDK supports two transport protocols:
Over TCP the SDK speaks the VSR consensus framing, which is the only wire protocol the server accepts.
See Viewstamped Replication (VSR) for what that means for the client API.
The SDK is built around the IIggyClient interface. The API fragments below reuse this client and these imports. Create the named resources before read, update, delete, send, or poll operations:
using System.Buffers; using System.Text; using System.Text.Json; using Apache.Iggy; using Apache.Iggy.Configuration; using Apache.Iggy.Consumers; using Apache.Iggy.Contracts; using Apache.Iggy.Contracts.Auth; using Apache.Iggy.Enums; using Apache.Iggy.Factory; using Apache.Iggy.Headers; using Apache.Iggy.IggyClient; using Apache.Iggy.Kinds; using Apache.Iggy.Messages; using Apache.Iggy.Publishers; using Microsoft.Extensions.Logging; using Partitioning = Apache.Iggy.Kinds.Partitioning; using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator { BaseAddress = "127.0.0.1:8090", Protocol = Protocol.Tcp }); await client.ConnectAsync(); await client.LoginUserAsync("iggy", "iggy");
Optionally, you can provide an ILoggerFactory for diagnostics and debugging (defaults to NullLoggerFactory.Instance). Add Microsoft.Extensions.Logging.Console to use AddConsole:
using var loggerFactory = LoggerFactory.Create(builder => { builder .AddFilter("Apache.Iggy", LogLevel.Information) .AddConsole(); }); using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator { BaseAddress = "127.0.0.1:8090", Protocol = Protocol.Tcp, LoggerFactory = loggerFactory }); await client.ConnectAsync();
The IggyClientConfigurator provides comprehensive configuration options:
using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator { BaseAddress = "127.0.0.1:8090", Protocol = Protocol.Tcp, // Socket buffer sizes in bytes (optional, null = OS default) ReceiveBufferSize = null, SendBufferSize = null, // Idle ping keeping the session alive under server-side heartbeat verification (TCP only). // Default 5 seconds, must be between 1 millisecond and about 49 days. Init-only. HeartbeatInterval = TimeSpan.FromSeconds(5), // Automatic reconnection with exponential backoff (enabled by default, infinite retries) ReconnectionSettings = new ReconnectionSettings { Enabled = true, MaxRetries = 0, // 0 = infinite retries InitialDelay = TimeSpan.FromSeconds(5), MaxDelay = TimeSpan.FromSeconds(30), WaitAfterReconnect = TimeSpan.FromSeconds(1), UseExponentialBackoff = true, BackoffMultiplier = 2.0 }, // Auto-login after connection. Optional for reconnection: a client that signs in with // LoginUserAsync has that sign-in replayed on a reconnect too. Without either, a reconnect // cannot restore the session and a lost connection fails the request AutoLoginSettings = AutoLoginSettings.For("iggy", "iggy"), // or AutoLoginSettings.ForPersonalAccessToken("your_token") // Optional: logging LoggerFactory = loggerFactory }); await client.ConnectAsync();
TCP applies the connection, heartbeat, TLS, and auto-login settings. HTTP ConnectAsync does no network work and requires explicit login. For TLS configuration, see the TcpTls example.
Over TCP every request is wrapped in a 256-byte consensus header, the client registers a consensus session at login, and replicated writes complete after the required quorum acknowledges them. Durability.Persisted additionally waits for persistence on that quorum; explicit offset stores have an independent ConsumerOffsetDurability policy. The IIggyClient surface is unchanged, with the few exceptions listed under Limitations.
using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator { BaseAddress = "127.0.0.1:8090", Protocol = Protocol.Tcp, // Upper bound on a reply frame the server announces, 64 MiB by default. MaxResponseFrameSize = 64 * 1024 * 1024, AutoLoginSettings = AutoLoginSettings.For("iggy", "iggy") }); await client.ConnectAsync();
LoginUserAsync / LoginWithPersonalAccessTokenAsync run the register handshake at login, and the session lives for as long as the connection. Logging out, being evicted or losing the connection ends it, and the next login registers a fresh one.JoinConsumerGroupAsync makes this client a member; the assignment is synced on demand and refreshed on every PingAsync. Partition counts are cached for 30 seconds, so a topic another client widens is picked up without waiting for a ping.PingAsync costs more than a ping. Besides the ping it re-syncs the assignment of every consumer group this client has joined, so it makes one extra round trip per joined group. The TCP client pings on its own every HeartbeatInterval (5 seconds by default) while connected, so an idle session survives the server's heartbeat verification and assignments stay fresh; a lost session is repaired by the regular reconnect and auto login.Within its request deadline, the SDK can replay a request the server reports as never admitted. Replay-safe operations can also be retried after a lost connection. Two cases surface to the caller:
IggyInvalidStatusCodeException carries the server status code, with FromServer telling apart a verdict the cluster reported from a failure the client raised itself.VsrRequestOutcomeUnknownException means no server verdict arrived after the request was written - the connection was lost, the call was cancelled, or the server evicted the session while the request was in flight - so the cluster may or may not have committed it. The SDK will not replay it on a new session, because that would bypass server-side deduplication - re-issuing it is the caller‘s decision. Background IggyPublisher sends report it through the message-batch-failed event without retrying; direct sends throw it to the caller, and IggyConsumer rethrows it rather than swallowing it, because an auto-committing poll may have advanced the offset already. Rethrowing ends the consumer’s polling loop: catch it around the enumeration, decide whether the operation is safe to re-issue, and start consuming again.GetMeAsync, segment deletion, and raw binary requests are TCP-only and throw FeatureUnavailableException on HTTP.StoreOffsetAsync / DeleteOffsetAsync need an explicit partition id under VSR: the broker does not resolve a null partition for a consumer-offset request, so passing one throws client-side.FlushUnsavedBufferAsync always throws FeatureUnavailableException; configure topic durability instead.IggyInvalidStatusCodeException. An existing topic with no available messages returns an empty poll.MaxResponseFrameSize bounds the reply frames the VSR reader accepts. A reply larger than the 64 MiB default is refused and the connection is dropped, so raise it if a single response legitimately exceeds thatGetSnapshotAsync is the usual case.IggyConsumerBuilder / IggyPublisherBuilder now auto-login with the credentials passed to WithConnection. Before, a builder-created client came back from a reconnect unauthenticated; now the credentials are held for the lifetime of the connection and replayed.HeartbeatInterval (5 seconds) and reconnects by default with unlimited failed roster passes. Each pass tries the known addresses before consuming one retry. Connection and TLS handshake failures can be retried; an invalid local CA file stops after the pass, and rejected credentials are surfaced. Set MaxRetries, pass a cancellation token, or disable reconnection to bound attempts. Request replay has its own deadline and outcome-safety rules. Configured auto-login credentials take precedence over credentials remembered from successful username/password or PAT login. Without either, reconnecting cannot restore the authenticated session.AutoLoginSettings properties are now init-only, as is IggyClientConfigurator.HeartbeatInterval. Build them with an object initializer or the AutoLoginSettings.For / AutoLoginSettings.ForPersonalAccessToken factories instead of assigning after construction.IggyConsumerBuilder / IggyPublisherBuilder accept a personal access token through the WithConnection(protocol, address, personalAccessToken, ...) overload, as an alternative to a username and password.System.IO.Hashing, used for the client-side message-key partitioner.NoDelay. The protocol is request/reply, so a write is always the last one before the client waits for the answer and Nagle has nothing to coalesce it with - it only held back the trailing segment of a large request until the previous one was acked.Use the credentials configured for the server. The setup below initializes a new server with iggy / iggy:
var response = await client.LoginUserAsync("iggy", "iggy");
Create new users with customizable permissions:
var permissions = new Permissions { Global = new GlobalPermissions { ManageServers = true, ManageUsers = true, ManageStreams = true, ManageTopics = true, PollMessages = true, ReadServers = true, ReadStreams = true, ReadTopics = true, ReadUsers = true, SendMessages = true } }; await client.CreateUserAsync("test_user", "secure_password", UserStatus.Active, permissions); // Login with the new user var loginResponse = await client.LoginUserAsync("test_user", "secure_password");
Create and use Personal Access Tokens (PAT) for programmatic access:
// Create a PAT var patResponse = await client.CreatePersonalAccessTokenAsync("api-token", TimeSpan.FromHours(1)); // Login with PAT await client.LoginWithPersonalAccessTokenAsync(patResponse!.Token);
await client.CreateStreamAsync("my-stream");
You can reference streams by either numeric ID or name:
var streamById = Identifier.Numeric(0); var streamByName = Identifier.String("my-stream");
Every stream contains topics for organizing messages:
var streamId = Identifier.String("my-stream"); await client.CreateTopicAsync( streamId, name: "my-topic", partitionsCount: 3, compressionAlgorithm: CompressionAlgorithm.None, messageExpiry: TimeSpan.MaxValue, // never expire maxTopicSize: 1024 * 1024 * 1024 // 1 GiB );
Note: Stream and topic names use hyphens instead of spaces. Iggy automatically replaces spaces with hyphens.
Send messages using the publisher interface:
var streamId = Identifier.String("my-stream"); var topicId = Identifier.String("my-topic"); var messages = new List<Message> { new(Guid.NewGuid(), "Hello, Iggy!"u8.ToArray()), new(1, "Another message"u8.ToArray()) }; await client.SendMessagesAsync( streamId, topicId, Partitioning.None(), // balanced partitioning messages );
Control which partition receives each message:
// Balanced partitioning (default) Partitioning.None(); // Send to specific partition Partitioning.PartitionId(1); // Key-based partitioning (string) Partitioning.EntityIdString("user-123"); // Key-based partitioning (integer) Partitioning.EntityIdInt(12345); // Key-based partitioning (GUID) Partitioning.EntityIdGuid(Guid.NewGuid());
Add custom headers to messages with typed values:
var headers = new Dictionary<HeaderKey, HeaderValue> { { HeaderKey.FromString("correlation_id"), HeaderValue.FromString("req-123") }, { HeaderKey.FromString("priority"), HeaderValue.FromInt32(1) }, { HeaderKey.FromString("timeout"), HeaderValue.FromInt64(5000) }, { HeaderKey.FromString("confidence"), HeaderValue.FromFloat(0.95f) }, { HeaderKey.FromString("is_urgent"), HeaderValue.FromBool(true) }, { HeaderKey.FromString("request_id"), HeaderValue.FromGuid(Guid.NewGuid()) } }; var messages = new List<Message> { new(Guid.NewGuid(), "Message with headers"u8.ToArray(), headers) }; await client.SendMessagesAsync( streamId, topicId, Partitioning.PartitionId(1), messages );
There is no on-demand flush command. Set TopicOptions.Durability = Durability.Persisted when creating a topic to wait for message persistence on the required quorum. ConsumerOffsetDurability controls explicit offset-store completion independently. Flush thresholds alone do not provide that guarantee.
Coordinate message consumption across multiple consumers:
var groupResponse = await client.CreateConsumerGroupAsync( Identifier.String("my-stream"), Identifier.String("my-topic"), "my-consumer-group" );
Note: Join/Leave operations are only supported on TCP protocol and will throw FeatureUnavailableException on HTTP.
// Join a consumer group await client.JoinConsumerGroupAsync( Identifier.String("my-stream"), Identifier.String("my-topic"), Identifier.String("my-consumer-group") ); // Leave a consumer group await client.LeaveConsumerGroupAsync( Identifier.String("my-stream"), Identifier.String("my-topic"), Identifier.String("my-consumer-group") );
Fetch a batch of messages:
var polledMessages = await client.PollMessagesAsync(new MessageFetchRequest { StreamId = streamId, TopicId = topicId, Consumer = Consumer.New(1), // or Consumer.Group("my-consumer-group") for consumer group Count = 10, PartitionId = 0, // optional, null for consumer group PollingStrategy = PollingStrategy.Next(), AutoCommit = true }); foreach (var message in polledMessages.Messages) { Console.WriteLine($"Message: {Encoding.UTF8.GetString(message.Payload)}"); }
With AutoCommit = true, the server advances/submits the offset before application processing. The poll reply does not wait for durable completion of that store. Use explicit offset stores after processing when that distinction matters.
Control where message consumption starts:
// Start from a specific offset PollingStrategy.Offset(1000); // Start from a specific timestamp (microseconds since epoch) PollingStrategy.Timestamp(1699564800000000); // Start from the first message PollingStrategy.First(); // Start from the last message PollingStrategy.Last(); // Start from the next unread message PollingStrategy.Next();
Store the current consumer position:
await client.StoreOffsetAsync( Consumer.New(1), Identifier.String("my-stream"), Identifier.String("my-topic"), offset: 42, partitionId: 0 );
Get the current stored offset:
var offsetInfo = await client.GetOffsetAsync( Consumer.New(1), Identifier.String("my-stream"), Identifier.String("my-topic"), partitionId: 0 ); Console.WriteLine($"Stored offset: {offsetInfo!.StoredOffset}");
Clear stored offsets:
await client.DeleteOffsetAsync( Consumer.New(1), Identifier.String("my-stream"), Identifier.String("my-topic"), partitionId: 0 );
Get cluster metadata and node information:
var metadata = await client.GetClusterMetadataAsync();
Retrieve server performance metrics:
var stats = await client.GetStatsAsync();
Verify server connectivity:
await client.PingAsync();
Get information about connected clients:
var clients = await client.GetClientsAsync(); var currentClient = await client.GetMeAsync();
Capture a system snapshot as a compressed ZIP archive:
var snapshotBytes = await client.GetSnapshotAsync( SnapshotCompression.Zstd, new List<SystemSnapshotType> { SystemSnapshotType.ServerLogs, SystemSnapshotType.ServerConfig, SystemSnapshotType.ResourceUsage } ); // Or capture everything var fullSnapshot = await client.GetSnapshotAsync( SnapshotCompression.Deflated, new List<SystemSnapshotType> { SystemSnapshotType.All } );
Available compression methods: Stored, Deflated, Bzip2, Zstd, Lzma, Xz.
Available snapshot types: FilesystemOverview, ProcessList, ResourceUsage, Test, ServerLogs, ServerConfig, All.
Delete up to N of the oldest sealed segments from a partition; the active segment is retained:
await client.DeleteSegmentsAsync( Identifier.String("my-stream"), Identifier.String("my-topic"), partitionId: 1, segmentsCount: 2 );
Subscribe to connection events:
Func<ConnectionStateChangedEventArgs, Task> handler = async args => { Console.WriteLine($"Current connection state: {args.CurrentState}"); await Task.CompletedTask; }; client.SubscribeConnectionEvents(handler); // Later client.UnsubscribeConnectionEvents(handler);
High-level publisher with background sending and retries. Retry settings apply to background sends; maxAttempts includes the first send. Await the queue drain before disposal:
using System.Buffers; using System.Text; using System.Text.Json; using Apache.Iggy; using Apache.Iggy.Consumers; using Apache.Iggy.Kinds; using Apache.Iggy.Messages; using Apache.Iggy.Publishers; await using var publisher = IggyPublisherBuilder.Create( client, Identifier.String("my-stream"), Identifier.String("my-topic") ) .WithPartitioning(Partitioning.None()) .WithBackgroundSending(enabled: true, batchSize: 100) .WithRetry(maxAttempts: 3) .Build(); await publisher.InitAsync(); var messages = new List<Message> { new(Guid.NewGuid(), "Message 1"u8.ToArray()), new(0, "Message 2"u8.ToArray()) }; await publisher.SendMessagesAsync(messages); // Drain the background queue, then dispose await publisher.WaitUntilAllSendsAsync();
For automatic object serialization, use the typed variant:
await using var publisher = IggyPublisherBuilder<Order>.Create( client, Identifier.String("orders-stream"), Identifier.String("orders-topic"), new OrderSerializer() ).Build(); await publisher.InitAsync(); await publisher.SendAsync(new List<Order> { new(Guid.NewGuid(), 99.90m) }); record Order(Guid OrderId, decimal Amount); class OrderSerializer : ISerializer<Order> { public void Serialize(Order data, IBufferWriter<byte> writer) => writer.Write(JsonSerializer.SerializeToUtf8Bytes(data)); }
High-level consumer with automatic offset management and consumer groups:
using System.Text; using Apache.Iggy; using Apache.Iggy.Consumers; using Apache.Iggy.Kinds; await using var consumer = IggyConsumerBuilder.Create( client, Identifier.String("my-stream"), Identifier.String("my-topic"), Consumer.New(1) ) .WithPollingStrategy(PollingStrategy.Next()) .WithBatchSize(10) .WithAutoCommitMode(AutoCommitMode.Auto) .Build(); await consumer.InitAsync(); await foreach (var message in consumer.ReceiveAsync()) { var payload = Encoding.UTF8.GetString(message.Message.Payload); Console.WriteLine($"Offset {message.CurrentOffset}: {payload}"); }
For consumer groups (load-balanced across multiple consumers):
await using var consumer = IggyConsumerBuilder.Create( client, Identifier.String("my-stream"), Identifier.String("my-topic"), Consumer.Group("my-group") ) .WithConsumerGroup("my-group", createIfNotExists: true, joinGroup: true) .WithPollingStrategy(PollingStrategy.Next()) .WithAutoCommitMode(AutoCommitMode.AfterReceive) .Build(); await consumer.InitAsync(); await foreach (var message in consumer.ReceiveAsync()) { var payload = Encoding.UTF8.GetString(message.Message.Payload); Console.WriteLine($"Partition {message.PartitionId}: {payload}"); }
For automatic deserialization:
var builder = IggyConsumerBuilder<OrderEvent>.Create( client, Identifier.String("orders-stream"), Identifier.String("orders-topic"), Consumer.Group("order-processors"), new OrderDeserializer() ); builder.WithPollingStrategy(PollingStrategy.Next()); builder.WithAutoCommitMode(AutoCommitMode.AfterReceive); await using var consumer = builder.Build(); await consumer.InitAsync(); await foreach (var message in consumer.ReceiveDeserializedAsync()) { if (message.Status == MessageStatus.Success) { Console.WriteLine($"Order: {message.Data?.OrderId}"); } } record OrderEvent(Guid OrderId, decimal Amount); class OrderDeserializer : IDeserializer<OrderEvent> { public OrderEvent Deserialize(ReadOnlyMemory<byte> data) => JsonSerializer.Deserialize<OrderEvent>(data.Span)!; }
The SDK provides the following main interfaces:
Additionally, builder-based APIs are available:
Examples are located in examples/csharp/src/ in root iggy directory. Available examples:
Start the matching source server from the repository root (use configured credentials for an existing data directory):
IGGY_ROOT_USERNAME=iggy IGGY_ROOT_PASSWORD=iggy cargo run --bin iggy-server
Run an example from the examples/csharp/ directory. Run TcpTls from the repository root so its certificate paths resolve:
dotnet run -c Release --project src/GettingStarted/Iggy_SDK.Examples.GettingStarted.Producer dotnet run -c Release --project src/GettingStarted/Iggy_SDK.Examples.GettingStarted.Consumer
Integration tests are located in Iggy_SDK.Tests.Integration/. Tests can run against:
IGGY_SERVER_HOST environment variable)The suite runs against iggy-server. TCP only: the SDK frames TCP with the VSR wire protocol, the cluster serves reads from the primary, and the HTTP surface has no equivalent path to route them through.
cargo build --bin iggy-server --bin iggy docker build --no-cache -f core/server/Dockerfile --platform linux/amd64 --target runtime-prebuilt --build-arg PREBUILT_IGGY_SERVER=target/debug/iggy-server --build-arg PREBUILT_IGGY_CLI=target/debug/iggy -t iggy-server:test .
dotnet build foreign/csharp/Iggy_SDK.Tests.Integration
cd foreign/csharp export IGGY_SERVER_DOCKER_IMAGE=iggy-server:test dotnet test -f net10.0 --project Iggy_SDK.Tests.Integration --no-build --verbosity diagnostic
IGGY_SERVER_DOCKER_IMAGE defaults to iggy-server:test, so the export above is only needed to point at a different image. Rider and Visual Studio need nothing configured.
ASP.NET Core Dependency Injection