Producer and consumer samples for the C# SDK, built on the high-level publisher and consumer.

These samples use the High-level SDK - the recommended way to build producers and consumers. For the low-level, per-call equivalents, see the Guide.

Producer

A publisher that creates the stream and topic if missing, batches sends in the background, and retries failures:

using System.Text;
using Apache.Iggy;
using Apache.Iggy.Configuration;
using Apache.Iggy.Enums;
using Apache.Iggy.Extensions;
using Apache.Iggy.Factory;
using Apache.Iggy.Messages;

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

await client.ConnectAsync();
await client.LoginUserAsync("iggy", "iggy");

var publisher = client.CreatePublisherBuilder(
        Identifier.String("dev"),
        Identifier.String("events"))
    .CreateStreamIfNotExists("dev")
    .CreateTopicIfNotExists("events", topicPartitionsCount: 2)
    .WithBackgroundSending(batchSize: 100, flushInterval: TimeSpan.FromMilliseconds(100))
    .WithRetry(maxAttempts: 3)
    .Build();

await publisher.InitAsync();

for (var i = 0; i < 100; i++)
{
    var payload = Encoding.UTF8.GetBytes($"Event #{i}");
    await publisher.SendMessagesAsync(new List<Message> { new(Guid.NewGuid(), payload) });
}

// Drain the background queue before exiting
await publisher.WaitUntilAllSendsAsync();
await publisher.DisposeAsync();

Console.WriteLine("Sent 100 messages");

Consumer group

A consumer that creates and joins a consumer group, commits offsets after each received message, and surfaces polling errors:

using System.Text;
using Apache.Iggy;
using Apache.Iggy.Configuration;
using Apache.Iggy.Consumers;
using Apache.Iggy.Enums;
using Apache.Iggy.Extensions;
using Apache.Iggy.Factory;
using Apache.Iggy.Kinds;

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

await client.ConnectAsync();
await client.LoginUserAsync("iggy", "iggy");

var consumer = client.CreateConsumerBuilder(
        Identifier.String("dev"),
        Identifier.String("events"),
        Consumer.Group("event-processors"))
    .WithConsumerGroup("event-processors", createIfNotExists: true, joinGroup: true)
    .WithPollingStrategy(PollingStrategy.Next())
    .WithBatchSize(20)
    .WithAutoCommitMode(AutoCommitMode.AfterReceive)
    .SubscribeOnPollingError(e =>
    {
        Console.WriteLine($"Polling error: {e.Exception.Message}");
        return Task.CompletedTask;
    })
    .Build();

await consumer.InitAsync();

await foreach (var message in consumer.ReceiveAsync())
{
    var payload = Encoding.UTF8.GetString(message.Message.Payload);
    Console.WriteLine($"Partition {message.PartitionId}, offset {message.CurrentOffset}: {payload}");
}

Run several instances of the consumer to see the group load-balance partitions between members.

A typed publisher/consumer pair that (de)serializes a record as JSON. The typed builders are configured via statements rather than one fluent chain - the With* methods return the untyped base builder, so Build() must be called on the typed builder variable (see Typed consumer):

using System.Buffers;
using System.Text.Json;
using Apache.Iggy;
using Apache.Iggy.Configuration;
using Apache.Iggy.Consumers;
using Apache.Iggy.Enums;
using Apache.Iggy.Factory;
using Apache.Iggy.Kinds;
using Apache.Iggy.Publishers;

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

await client.ConnectAsync();
await client.LoginUserAsync("iggy", "iggy");

// Publish typed messages
var publisherBuilder = IggyPublisherBuilder<OrderEvent>.Create(
    client,
    Identifier.String("orders"),
    Identifier.String("created"),
    new OrderSerializer()
);
publisherBuilder.CreateStreamIfNotExists("orders");
publisherBuilder.CreateTopicIfNotExists("created");

var publisher = publisherBuilder.Build();
await publisher.InitAsync();

await publisher.SendAsync(new OrderEvent(Guid.NewGuid(), 99.90m));
await publisher.DisposeAsync();

// Consume them
var consumerBuilder = IggyConsumerBuilder<OrderEvent>.Create(
    client,
    Identifier.String("orders"),
    Identifier.String("created"),
    Consumer.New(1),
    new OrderDeserializer()
);
consumerBuilder.WithPollingStrategy(PollingStrategy.Next());
consumerBuilder.WithAutoCommitMode(AutoCommitMode.AfterReceive);

var consumer = consumerBuilder.Build();
await consumer.InitAsync();

await foreach (var message in consumer.ReceiveDeserializedAsync())
{
    if (message.Status == MessageStatus.Success)
    {
        Console.WriteLine($"Order {message.Data!.OrderId}, amount {message.Data.Amount}");
    }
}

record OrderEvent(Guid OrderId, decimal Amount);

class OrderSerializer : ISerializer<OrderEvent>
{
    public void Serialize(OrderEvent data, IBufferWriter<byte> writer) =>
        writer.Write(JsonSerializer.SerializeToUtf8Bytes(data));
}

class OrderDeserializer : IDeserializer<OrderEvent>
{
    public OrderEvent Deserialize(ReadOnlyMemory<byte> data) =>
        JsonSerializer.Deserialize<OrderEvent>(data.Span)!;
}

More examples

The examples/csharp directory in the Iggy repository contains complete, runnable projects:

Basic / GettingStarted - low-level producer and consumer
NewSdk - high-level IggyPublisher / IggyConsumer (like the samples above)
MessageEnvelope - envelope pattern (message type + JSON payload) over the low-level client
MessageHeaders - user-defined message headers
TcpTls - TLS-encrypted TCP connection

Examples

Producer

Consumer group

Typed messages

More examples

On this page