Guide

using System.Text;
using Apache.Iggy;
using Apache.Iggy.Configuration;
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 Partitioning = Apache.Iggy.Kinds.Partitioning;

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

    // Null preserves the operating system socket defaults.
    ReceiveBufferSize = null,
    SendBufferSize = null,

    // Upper bound on a reply frame the server sends, 64 MiB by default (TCP)
    MaxResponseFrameSize = 64 * 1024 * 1024,

    // Idle ping keeping the session alive and consumer-group assignments fresh (TCP).
    // Default 5 seconds; 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
    },

    // Configured credentials are reused when TCP reconnects.
    AutoLoginSettings = AutoLoginSettings.For("iggy", "iggy")
    // or AutoLoginSettings.ForPersonalAccessToken("your_token")
});

await client.ConnectAsync();

using System.Security.Cryptography;
using Apache.Iggy.Encryption;

var key = RandomNumberGenerator.GetBytes(32);
using var encryptor = new AesMessageEncryptor(key);

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

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

client.SubscribeConnectionEvents(handler);

// Later
client.UnsubscribeConnectionEvents(handler);

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

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

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

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

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<uint, StreamPermissions>
    {
        [0] = new StreamPermissions
        {
            ManageStream = false,
            ReadStream = true,
            ManageTopics = false,
            ReadTopics = true,
            PollMessages = true,
            SendMessages = true,
            Topics = new Dictionary<uint, TopicPermissions>
            {
                [0] = new TopicPermissions
                {
                    ManageTopic = false,
                    ReadTopic = true,
                    PollMessages = true,
                    SendMessages = false
                }
            }
        }
    }
};

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);
userId = Identifier.String("renamed_user");

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

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

// Delete
await client.DeleteUserAsync(userId);

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

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

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

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

await client.CreateTopicAsync(
    streamId,
    name: "durable-topic",
    partitionsCount: 3,
    options: new TopicOptions
    {
        SegmentSize = 128 * 1024 * 1024,                  // segment size in bytes (multiple of 512)
        Durability = Durability.Persisted,
        ConsumerOffsetDurability = Durability.Persisted,
        MessagesRequiredToSave = 1000,                    // attempt a flush after this many messages
        SizeOfMessagesRequiredToSave = 32 * 1024 * 1024,  // or after this many bytes, whichever trips first
        PreallocateSegments = true                        // request filesystem preallocation
    }.ToDictionary()
);

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 optional)
await client.UpdateTopicAsync(streamId, topicId, "renamed-topic",
    messageExpiry: TimeSpan.FromDays(7), maxTopicSize: 1024 * 1024 * 1024);

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

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

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

var response = await client.SendMessagesAsync(
    streamId,
    topicId,
    Partitioning.None(),  // balanced partitioning
    messages
);

// The confirmations report where each committed batch landed; this is the only
// way to learn the partition and offset a message was assigned
foreach (var confirmation in response.Confirmations)
{
    Console.WriteLine($"Partition {confirmation.PartitionId}, base offset {confirmation.BaseOffset}");
}

// Balanced - the client round-robins across partitions (default)
Partitioning.None();

// Send to a specific partition (partition ids are 0-based)
Partitioning.PartitionId(0);

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

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(0), messages);

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

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

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.

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

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

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

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

// Health check
await client.PingAsync();

// Address of the node the client currently talks to
var address = client.GetCurrentAddress();

// 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

var specs = await client.DescribeOptionsAsync(OptionsScope.Topic);

foreach (var spec in specs)
{
    Console.WriteLine($"{spec.Key} ({spec.Kind}): {spec.Description}");
}

const uint PingCommand = 1;
var responseBytes = await client.SendBinaryRequestAsync(PingCommand, Array.Empty<byte>());

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

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