High-level SDK

The per-call IIggyClient API is explicit but verbose - you manage partitioning, batching, retries, and offsets yourself. The high-level IggyPublisher and IggyConsumer wrap that surface with fluent builders that handle:

• Background sending with buffering, batching, and retries
• Automatic stream/topic/consumer-group creation
• Automatic offset commits (on each poll, after each received message, or manual)
• Consuming a topic as an async stream (IAsyncEnumerable)
• Typed message (de)serialization
• Sending and receiving over rented, pooled buffers to avoid separate payload copies on TCP

The fragments below reuse an authenticated client and the imports in the first example. The named streams/topics must already exist unless the snippet creates them. Both abstractions can be built from an existing connected client via IggyPublisherBuilder.Create(...) / IggyConsumerBuilder.Create(...), or equivalently via the client.CreatePublisherBuilder(...) / client.CreateConsumerBuilder(...) extension methods from Apache.Iggy.Extensions.

IggyPublisher

Configure a publisher, initialize it, then send. InitAsync validates (and optionally creates) the stream and topic:

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

Publisher builder options

WithRetry parameters

WithBackgroundSending parameters

A batch also flushes once its accumulated payload reaches IggyPublisherConfig.BackgroundMaxBatchBytes (default 256 KiB, 0 disables the byte gate) - whichever of batchSize or the byte limit is hit first. This knob has no builder method. Set it on the config directly if needed. Await WaitUntilAllSendsAsync before disposal to finish queued sends; disposal alone can discard them. Background send calls return an empty confirmation list before delivery, and failures are reported through the subscribed events.

Publishing with rented batches

SendMessagesAsync accepts messages with ReadOnlyMemory<byte> payloads, which can share backing storage. On hot paths, build a RentedMessageBatch instead - every payload is written into a single buffer rented from the shared array pool - and hand it to SendAsync:

using System.Text.Json;
using Apache.Iggy.Messages;

using var builder = new RentedMessageBatchBuilder(sizeHint: 4096);

foreach (var evt in events)
{
    // Serialize straight into the pooled buffer (static lambda + state = no closure allocation)
    builder.Add(evt, static (e, writer) =>
    {
        using var json = new Utf8JsonWriter(writer);
        JsonSerializer.Serialize(json, e);
    }, Guid.NewGuid());
}

var batch = builder.Build();
await publisher.SendAsync(batch);
await publisher.WaitUntilAllSendsAsync();

RentedMessageBatchBuilder:

A null message id sends the message with id 0, letting the server assign one.

Ownership rules:

• publisher.SendAsync(batch) takes ownership - do not dispose the batch yourself. Its buffer returns to the pool once the batch is sent (immediately, or after the background flush).
• For a direct low-level send, pass batch.Messages to client.SendMessagesAsync(...) yourself, await it, then dispose the batch (using var batch = builder.Build()).
• After disposal, batch.Messages and every payload's Span throw ObjectDisposedException instead of reading recycled pool memory.
• Dispose the builder only when abandoning it before Build(); a using on the builder is safe either way (it becomes a no-op after Build()).

Typed publisher

For automatic object serialization, use IggyPublisherBuilder<T> with an ISerializer<T>:

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

Besides the collection overload, SendAsync also accepts a single item (SendAsync(order, messageId: null, userHeaders: null)) or a collection of (data, messageId, userHeaders) tuples when you need per-message ids or headers.

For JSON you don't need a custom serializer - the built-in SystemTextJsonSerializer<T> (in Apache.Iggy.Publishers) writes System.Text.Json output directly into the send buffer, with optional JsonSerializerOptions:

await using var publisher = IggyPublisherBuilder<Order>.Create(
    client,
    Identifier.String("orders-stream"),
    Identifier.String("orders-topic"),
    new SystemTextJsonSerializer<Order>()
).Build();

With background sending enabled, typed values are serialized at flush time. Keep them immutable or pass a snapshot until WaitUntilAllSendsAsync completes.

There is no built-in deserializer counterpart - consumers implement IDeserializer<T> themselves (see Typed consumer).

IggyConsumer

Configure a consumer, initialize it, then iterate. ReceiveAsync returns an IAsyncEnumerable<ReceivedMessage>:

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

Consumer builder options

For consumer groups, use Next() with offset commits to track each partition independently. An Offset(...) strategy advances one shared cursor across the group's partitions and can skip messages from another partition.

Auto-commit modes (AutoCommitMode):

Manual offset control

With AutoCommitMode.Disabled, commit offsets yourself via the consumer's StoreOffsetAsync:

await foreach (var message in consumer.ReceiveAsync())
{
    Process(message);
    await consumer.StoreOffsetAsync(message.CurrentOffset, message.PartitionId);
}

StoreOffsetAsync(offset, partitionId, resetLastPolled = false) stores the offset for a partition. resetLastPolled: true also changes its duplicate-filter position; it does not change the configured polling strategy or remove buffered messages. DeleteOffsetAsync(partitionId) clears the server offset while preserving that local state. Create a new consumer to restart with a different position.

Consumer groups

Pass a Consumer.Group(...) and let the builder create and join the group for load-balanced consumption:

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

Consuming with rented buffers

On TCP, ReceiveAsync copies each payload into its own byte[]. ReceiveRentedAsync is the high-level counterpart of the low-level rented poll API: payloads are slices of a pooled buffer shared by all messages from the same poll. Each yielded ReceivedRentedMessage must be disposed - the buffer returns to the pool once the last message of its batch is disposed:

await foreach (var message in consumer.ReceiveRentedAsync())
{
    using (message)
    {
        var text = Encoding.UTF8.GetString(message.Message.Payload.Span);
        Console.WriteLine($"Offset {message.CurrentOffset}: {text}");
    }
}

ReceivedRentedMessage (IDisposable) exposes Message (a RentedMessageResponse), CurrentOffset, PartitionId, Status, and Error. This path performs no deserialization, so Status is always Success. Payload and raw-header memory are only valid until the message is disposed - copy out anything you need to keep (e.g. message.Message.Payload.ToArray()).

Auto-commit modes apply exactly as with ReceiveAsync. If a message is not disposed, the batch buffer is not returned to the pool; it becomes eligible for garbage collection once no references remain.

Typed consumer

For automatic deserialization, use IggyConsumerBuilder<T> with an IDeserializer<T> and iterate with ReceiveDeserializedAsync. Each ReceivedMessage<T> carries a Status you should check:

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

Note: On the typed builders, the fluent With* methods are inherited from the untyped base builder and return the base type, while the typed Build() hides the base one. Chaining Create(...).WithAutoCommitMode(...).Build() therefore resolves to the base Build() and returns an untyped consumer/publisher. Keep the typed builder in a variable (as above) and call Build() on it - the With* calls mutate the builder, so their return value can be ignored.