Guide

This guide covers client configuration and the full IIggyClient API surface — the low-level, per-call operations. For the ergonomic producer/consumer abstractions built on top of these, see the High-level SDK.

All examples assume you already have a connected, authenticated client (see Creating a client).

Client configuration

IggyClientConfigurator exposes the full set of connection options — buffer sizes, TLS, automatic reconnection with exponential backoff, and auto-login:

var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
    BaseAddress = "127.0.0.1:8090",
    Protocol = Protocol.Tcp,

    // Buffer sizes (optional, default: 4096)
    ReceiveBufferSize = 4096,
    SendBufferSize = 4096,

    // TLS/SSL configuration
    TlsSettings = new TlsSettings
    {
        Enabled = true,
        Hostname = "iggy",
        CertificatePath = "/path/to/cert"
    },

    // Automatic reconnection with exponential backoff
    ReconnectionSettings = new ReconnectionSettings
    {
        Enabled = true,
        MaxRetries = 3,              // 0 = infinite retries
        InitialDelay = TimeSpan.FromSeconds(5),
        MaxDelay = TimeSpan.FromSeconds(30),
        WaitAfterReconnect = TimeSpan.FromSeconds(1),
        UseExponentialBackoff = true,
        BackoffMultiplier = 2.0
    },

    // Auto-login after connection
    AutoLoginSettings = new AutoLoginSettings
    {
        Enabled = true,
        Username = "iggy",
        Password = "iggy"
    }
});

await client.ConnectAsync();

With AutoLoginSettings.Enabled = true, the client logs in automatically once the connection is established, so you can skip the explicit LoginUserAsync call.

`IggyClientConfigurator`

Property	Type	Default	Description
`BaseAddress`	`string`	required	Server address, e.g. `127.0.0.1:8090`
`Protocol`	`Protocol`	required	Transport: `Protocol.Tcp` or `Protocol.Http`
`ReceiveBufferSize`	`int`	`4096`	Receive buffer size in bytes
`SendBufferSize`	`int`	`4096`	Send buffer size in bytes
`TlsSettings`	`TlsSettings`	disabled	TLS/SSL configuration (see below)
`ReconnectionSettings`	`ReconnectionSettings`	see below	Automatic reconnection behavior
`AutoLoginSettings`	`AutoLoginSettings`	disabled	Automatic login on connect
`LoggerFactory`	`ILoggerFactory`	`NullLoggerFactory.Instance`	Logger factory for diagnostics (currently applied to TCP clients only)
`MessageEncryptor`	`IMessageEncryptor?`	`null`	Client-side payload encryptor (encrypts on send, decrypts on poll — see Message encryption)
`AllowAutoCommitWithEncryptor`	`bool`	`false`	Allow auto-commit while an encryptor is configured

`TlsSettings`

Property	Type	Default	Description
`Enabled`	`bool`	`false`	Whether TLS is enabled
`Hostname`	`string`	`""`	Server name for the TLS handshake
`CertificatePath`	`string`	`""`	Path to the certificate (CA / self-signed) file

`ReconnectionSettings`

Property	Type	Default	Description
`Enabled`	`bool`	`false`	Enable automatic reconnection when the connection drops
`MaxRetries`	`int`	`3`	Maximum reconnection attempts (`0` = infinite)
`InitialDelay`	`TimeSpan`	`5s`	Delay before the first reconnection attempt
`MaxDelay`	`TimeSpan`	`30s`	Maximum delay between attempts
`WaitAfterReconnect`	`TimeSpan`	`1s`	Pause after a successful reconnect (e.g. to rejoin a consumer group)
`UseExponentialBackoff`	`bool`	`true`	Use exponential backoff for delays
`BackoffMultiplier`	`double`	`2.0`	Multiplier for exponential backoff

`AutoLoginSettings`

Property	Type	Default	Description
`Enabled`	`bool`	`false`	Log in automatically once connected
`Username`	`string`	`""`	Username for auto-login
`Password`	`string`	`""`	Password for auto-login

Message encryption

Set IggyClientConfigurator.MessageEncryptor to encrypt message payloads and user headers client-side — they are encrypted on send and decrypted on poll, so the server only ever sees ciphertext. The built-in AesMessageEncryptor uses AES-GCM (confidentiality + authenticity) and takes a 16-, 24-, or 32-byte key for AES-128/192/256:

using Apache.Iggy.Encryption;

using var encryptor = new AesMessageEncryptor(key); // 16, 24, or 32 bytes

var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
    BaseAddress = "127.0.0.1:8090",
    Protocol = Protocol.Tcp,
    MessageEncryptor = encryptor
});

For a custom scheme, implement IMessageEncryptor.

Things to know:

The encryptor applies to every message on the connection — topics mixing encrypted and plaintext messages are not supported.
You own the encryptor and must dispose it once the client is done with it.
With an encryptor configured, polling with autoCommit: true throws InvalidOperationException by default: the server commits the offset before the client decrypts, so a decryption failure would silently skip the whole batch. Poll with autoCommit: false and store offsets after processing, or opt in with AllowAutoCommitWithEncryptor = true. This guard does not affect the high-level IggyConsumer commit modes.
On the high-level builders, WithEncryptor(...) is only valid when the builder creates its own client; for an external client, set MessageEncryptor on the configurator instead.

Connection events

Subscribe to connection state changes — useful for reacting to reconnects (e.g. rejoining a consumer group):

Func<ConnectionStateChangedEventArgs, Task> handler = async args =>
{
    Console.WriteLine($"Current connection state: {args.CurrentState}");
    await Task.CompletedTask;
};

client.SubscribeConnectionEvents(handler);

// Later
client.UnsubscribeConnectionEvents(handler);

Identifiers

Streams, topics, users, and consumer groups are referenced by an Identifier, which can be either numeric or a name:

var byId = Identifier.Numeric(0);
var byName = Identifier.String("my-stream");

Authentication

Begin with the root account (iggy / iggy):

var response = await client.LoginUserAsync("iggy", "iggy");

// Log out the currently authenticated user
await client.LogoutUserAsync();

Creating users

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);

var loginResponse = await client.LoginUserAsync("test_user", "secure_password");

Besides Global, Permissions.Streams scopes permissions to specific streams (keyed by numeric stream id), and each StreamPermissions can in turn scope down to specific topics:

var permissions = new Permissions
{
    Global = new GlobalPermissions
    {
        ManageServers = false,
        ManageUsers = false,
        ManageStreams = false,
        ManageTopics = false,
        PollMessages = false,
        ReadServers = false,
        ReadStreams = false,
        ReadTopics = false,
        ReadUsers = false,
        SendMessages = false
    },
    Streams = new Dictionary<int, StreamPermissions>
    {
        [1] = new StreamPermissions
        {
            ManageStream = false,
            ReadStream = true,
            ManageTopics = false,
            ReadTopics = true,
            PollMessages = true,
            SendMessages = true,
            Topics = new Dictionary<int, TopicPermissions>
            {
                [1] = new TopicPermissions
                {
                    ManageTopic = false,
                    ReadTopic = true,
                    PollMessages = true,
                    SendMessages = false
                }
            }
        }
    }
};

Managing users

var userId = Identifier.String("test_user");

// Get / list
var user = await client.GetUserAsync(userId);
var users = await client.GetUsersAsync();

// Update name and/or status (both optional)
await client.UpdateUserAsync(userId, userName: "renamed_user", status: UserStatus.Inactive);

// Replace permissions
await client.UpdatePermissionsAsync(userId, permissions);

// Change password
await client.ChangePasswordAsync(userId, "secure_password", "new_password");

// Delete
await client.DeleteUserAsync(userId);

Personal access tokens

Create and use Personal Access Tokens (PAT) for programmatic access:

// Create a PAT (expiry is optional; null = server default, TimeSpan.MaxValue = never expires)
var pat = await client.CreatePersonalAccessTokenAsync("api-token", TimeSpan.FromHours(1));

// Login with the raw token value
await client.LoginWithPersonalAccessTokenAsync(pat!.Token);

// List / delete (by name)
var tokens = await client.GetPersonalAccessTokensAsync();
await client.DeletePersonalAccessTokenAsync("api-token");

Streams

// Create
await client.CreateStreamAsync("my-stream");

// Get / list
var stream = await client.GetStreamByIdAsync(Identifier.String("my-stream"));
var streams = await client.GetStreamsAsync();

// Update / purge / delete
await client.UpdateStreamAsync(Identifier.String("my-stream"), "renamed-stream");
await client.PurgeStreamAsync(Identifier.String("renamed-stream"));
await client.DeleteStreamAsync(Identifier.String("renamed-stream"));

Topics

Every stream contains topics that organize messages into partitions:

var streamId = Identifier.String("my-stream");

await client.CreateTopicAsync(
    streamId,
    name: "my-topic",
    partitionsCount: 3,
    compressionAlgorithm: CompressionAlgorithm.None,
    replicationFactor: 1,
    messageExpiry: TimeSpan.Zero,  // null or TimeSpan.Zero = server default; TimeSpan.MaxValue = never expire
    maxTopicSize: 0                // 0 = unlimited
);

compressionAlgorithm, replicationFactor, messageExpiry, and maxTopicSize are optional. messageExpiry is a nullable TimeSpan.

var topicId = Identifier.String("my-topic");

// Get / list
var topic = await client.GetTopicByIdAsync(streamId, topicId);
var topics = await client.GetTopicsAsync(streamId);

// Update (name required; compression, expiry, size, replication optional)
await client.UpdateTopicAsync(streamId, topicId, "renamed-topic");

// Purge (delete all messages, keep the topic) / delete
await client.PurgeTopicAsync(streamId, Identifier.String("renamed-topic"));
await client.DeleteTopicAsync(streamId, Identifier.String("renamed-topic"));

Partitions

Add partitions to or remove them from an existing topic:

await client.CreatePartitionsAsync(streamId, topicId, partitionsCount: 2);
await client.DeletePartitionsAsync(streamId, topicId, partitionsCount: 2);

Publishing messages

Sending messages

A Message takes an id (Guid or UInt128) and a payload:

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
);

A single-message overload is also available: SendMessagesAsync(streamId, topicId, partitioning, message).

To send many payloads without a byte[] allocation per message, build them into a single pooled buffer with RentedMessageBatchBuilder — see Publishing with rented batches.

Partitioning strategies

Control which partition receives each message:

// Balanced — the server selects the partition (default)
Partitioning.None()

// Send to a specific partition
Partitioning.PartitionId(1)

// Key-based routing — messages with the same key land on the same partition
Partitioning.EntityIdString("user-123")
Partitioning.EntityIdInt(12345)
Partitioning.EntityIdUlong(12345)
Partitioning.EntityIdGuid(Guid.NewGuid())
Partitioning.EntityIdBytes(new byte[] { 1, 2, 3 })

User-defined headers

Add typed custom headers to messages. Build keys with HeaderKey.FromString and values with the HeaderValue.From* factories:

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);

Available value factories: FromString, FromBool, FromBytes, FromInt32, FromInt64, FromInt128, FromUInt32, FromUInt64, FromUInt128, FromFloat, FromDouble, FromGuid.

Flushing the unsaved buffer

Force a flush of the in-memory buffer to disk for a specific partition. When fsync is true, data is both flushed and synchronized (durable):

await client.FlushUnsavedBufferAsync(
    Identifier.String("my-stream"),
    Identifier.String("my-topic"),
    partitionId: 1,
    fsync: true
);

Consuming messages

Fetching messages

Poll a batch of messages. The partitionId may be null to consume from any partition:

var polledMessages = await client.PollMessagesAsync(
    streamId,
    topicId,
    partitionId: 0,
    Consumer.New(1),               // or Consumer.Group("my-group")
    PollingStrategy.Next(),
    count: 10,
    autoCommit: true
);

foreach (var message in polledMessages.Messages)
{
    Console.WriteLine($"Message: {Encoding.UTF8.GetString(message.Payload)}");
}

A convenience overload accepts a MessageFetchRequest if you prefer named fields:

var polledMessages = await client.PollMessagesAsync(new MessageFetchRequest
{
    StreamId = streamId,
    TopicId = topicId,
    Consumer = Consumer.New(1),
    Count = 10,
    PartitionId = 0,
    PollingStrategy = PollingStrategy.Next(),
    AutoCommit = true
});

Polling with rented buffers

PollMessagesAsync copies each payload into its own byte[]. On hot paths that allocation adds up. PollMessagesRentedAsync instead returns a PolledMessagesRental whose payloads and raw headers are slices over a single buffer rented from a shared pool — no per-message allocation.

The rental owns that buffer, so you must dispose it, and the payload/header memory is only valid until you do. Wrap it in using and never hold a Payload/RawUserHeaders reference past the block:

using var rental = await client.PollMessagesRentedAsync(
    streamId,
    topicId,
    partitionId: 0,
    Consumer.New(1),
    PollingStrategy.Next(),
    count: 100,
    autoCommit: true
);

foreach (var message in rental.Messages)
{
    // message.Payload is ReadOnlyMemory<byte> backed by the rented buffer.
    // Process it in place; copy out only what you need to keep.
    var text = Encoding.UTF8.GetString(message.Payload.Span);
    Console.WriteLine($"Offset {message.Header.Offset}: {text}");
}
// Buffer returns to the pool here. Payload/RawUserHeaders are invalid after this point.

A MessageFetchRequest overload (PollMessagesRentedAsync(request)) is also available.

PolledMessagesRental (IDisposable) exposes:

Member	Type	Description
`PartitionId`	`int`	Partition the messages came from
`CurrentOffset`	`ulong`	Current offset for the partition
`Messages`	`IReadOnlyList<RentedMessageResponse>`	The rented messages

Each RentedMessageResponse:

Member	Type	Description
`Header`	`MessageHeader`	Message header (offset, timestamp, id, …)
`Payload`	`ReadOnlyMemory<byte>`	Payload backed by rented memory — valid only until the rental is disposed
`RawUserHeaders`	`ReadOnlyMemory<byte>`	Raw user-header bytes backed by rented memory
`UserHeaders`	`Dictionary<HeaderKey, HeaderValue>?`	User headers, parsed lazily and cached on first access

Warning: Do not store Payload, RawUserHeaders, or a RentedMessageResponse beyond the using scope. To retain data, copy it out (e.g. message.Payload.ToArray()) before disposal.

The high-level IggyConsumer exposes the same pooled path as an async stream via ReceiveRentedAsync — see Consuming with rented buffers.

Polling strategies

Control where consumption starts:

PollingStrategy.Offset(1000)              // from a specific offset
PollingStrategy.Timestamp(1699564800000000)  // from a timestamp (microseconds since epoch)
PollingStrategy.First()                   // from the earliest message
PollingStrategy.Last()                    // from the latest message
PollingStrategy.Next()                    // from the next unread message

Offset management

Consumer offsets are keyed by consumer + stream + topic + partition. Note the argument order — the Consumer comes first:

var consumer = Consumer.New(1);
var streamId = Identifier.String("my-stream");
var topicId = Identifier.String("my-topic");

// Store the current position
await client.StoreOffsetAsync(consumer, streamId, topicId, offset: 42, partitionId: 0);

// Retrieve the stored offset
var offsetInfo = await client.GetOffsetAsync(consumer, streamId, topicId, partitionId: 0);
Console.WriteLine($"Stored offset: {offsetInfo!.StoredOffset}");

// Clear the stored offset
await client.DeleteOffsetAsync(consumer, streamId, topicId, partitionId: 0);

Consumer groups

Consumer groups coordinate message consumption across multiple consumers, load-balancing partitions between members.

var streamId = Identifier.String("my-stream");
var topicId = Identifier.String("my-topic");

// Create
await client.CreateConsumerGroupAsync(streamId, topicId, "my-consumer-group");

// Inspect
var groups = await client.GetConsumerGroupsAsync(streamId, topicId);
var group = await client.GetConsumerGroupByIdAsync(streamId, topicId, Identifier.String("my-consumer-group"));

// Delete
await client.DeleteConsumerGroupAsync(streamId, topicId, Identifier.String("my-consumer-group"));

Joining and leaving

Note: Join/Leave are TCP-only and throw FeatureUnavailableException on HTTP.

await client.JoinConsumerGroupAsync(streamId, topicId, Identifier.String("my-consumer-group"));
await client.LeaveConsumerGroupAsync(streamId, topicId, Identifier.String("my-consumer-group"));

System operations

// Health check
await client.PingAsync();

// Server statistics
var stats = await client.GetStatsAsync();

// Cluster metadata and node information
var metadata = await client.GetClusterMetadataAsync();

// Connected clients
var clients = await client.GetClientsAsync();
var clientById = await client.GetClientByIdAsync(clientId: 1);
var currentClient = await client.GetMeAsync();  // TCP-only

Snapshots

Capture a system snapshot as a compressed 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 }
);

Compression methods: Stored, Deflated, Bzip2, Zstd, Lzma, Xz. Snapshot types: FilesystemOverview, ProcessList, ResourceUsage, Test, ServerLogs, ServerConfig, All.

Segment management

Delete the last N segments from a partition:

Note: TCP-only — throws FeatureUnavailableException on HTTP.

await client.DeleteSegmentsAsync(
    Identifier.String("my-stream"),
    Identifier.String("my-topic"),
    partitionId: 1,
    segmentsCount: 2
);

On this page