This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 81f282647 docs(csharp): update csharp README.md file and bump
dependencies (#2356)
81f282647 is described below
commit 81f28264758657e77a336c87cd65d5c2b5d901ef
Author: Łukasz Zborek <[email protected]>
AuthorDate: Mon Nov 17 08:49:35 2025 +0100
docs(csharp): update csharp README.md file and bump dependencies (#2356)
---
foreign/csharp/DEPENDENCIES.md | 4 +-
foreign/csharp/Directory.Packages.props | 4 +-
foreign/csharp/README.md | 764 ++++++++++++++++++++------------
3 files changed, 490 insertions(+), 282 deletions(-)
diff --git a/foreign/csharp/DEPENDENCIES.md b/foreign/csharp/DEPENDENCIES.md
index 2114743dd..5e8657912 100644
--- a/foreign/csharp/DEPENDENCIES.md
+++ b/foreign/csharp/DEPENDENCIES.md
@@ -9,10 +9,10 @@ Microsoft.NET.Test.Sdk: "18.0.0", "MIT",
Microsoft.Testing.Extensions.CodeCoverage: "18.0.4", "MIT",
Microsoft.Testing.Extensions.TrxReport: "1.9.0", "MIT",
Moq: "4.20.72", "BSD-3-Clause",
-Reqnroll.xUnit: "3.1.2", "BSD-3-Clause",
+Reqnroll.xUnit: "3.2.1", "BSD-3-Clause",
Shouldly: "4.3.0", "BSD-3-Clause",
System.IO.Hashing: "8.0.0", "MIT",
-Testcontainers: "4.7.0", "MIT",
+Testcontainers: "4.8.1", "MIT",
TUnit: "0.70.0", "MIT",
xunit: "2.9.3", "Apache-2.0",
xunit.runner.visualstudio: "3.1.5", "Apache-2.0",
diff --git a/foreign/csharp/Directory.Packages.props
b/foreign/csharp/Directory.Packages.props
index c6056884a..9ab1be837 100644
--- a/foreign/csharp/Directory.Packages.props
+++ b/foreign/csharp/Directory.Packages.props
@@ -13,10 +13,10 @@
<PackageVersion Include="Microsoft.Testing.Extensions.CodeCoverage"
Version="18.0.4" />
<PackageVersion Include="Microsoft.Testing.Extensions.TrxReport"
Version="1.9.0" />
<PackageVersion Include="Moq" Version="4.20.72" />
- <PackageVersion Include="Reqnroll.xUnit" Version="3.1.2" />
+ <PackageVersion Include="Reqnroll.xUnit" Version="3.2.1" />
<PackageVersion Include="Shouldly" Version="4.3.0" />
<PackageVersion Include="System.IO.Hashing" Version="8.0.0" />
- <PackageVersion Include="Testcontainers" Version="4.7.0" />
+ <PackageVersion Include="Testcontainers" Version="4.8.1" />
<PackageVersion Include="TUnit" Version="0.70.0" />
<PackageVersion Include="xunit" Version="2.9.3" />
<PackageVersion Include="xunit.runner.visualstudio" Version="3.1.5">
diff --git a/foreign/csharp/README.md b/foreign/csharp/README.md
index 24eaef7f0..6e4a76228 100644
--- a/foreign/csharp/README.md
+++ b/foreign/csharp/README.md
@@ -6,407 +6,615 @@
</div>
+## Overview
+
+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.
+
## Getting Started
-Currently supported transfer protocols
+### Installation
+
+Install the NuGet package:
+
+```bash
+dotnet add package Apache.Iggy
+```
+
+### Supported Protocols
-- TCP
-- HTTP
+The SDK supports two transport protocols:
-The whole SDK revolves around `IIggyClient` interface to create an instance of
it, use following code
+- **TCP** - Binary protocol for optimal performance and lower latency
(recommended)
+- **HTTP** - RESTful JSON API for stateless operations
+
+### Creating a Client
+
+The SDK is built around the `IIggyClient` interface. To create a client
instance:
```c#
var loggerFactory = LoggerFactory.Create(builder =>
{
builder
- .AddFilter("Iggy_SDK.MessageStream.Implementations;", LogLevel.Trace)
+ .AddFilter("Apache.Iggy", LogLevel.Information)
.AddConsole();
});
-var bus = MessageStreamFactory.CreateMessageStream(options =>
+
+var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
- options.BaseAdress = "127.0.0.1:8090";
- options.Protocol = Protocol.Tcp;
- options.TlsSettings = x =>
- {
- x.Enabled = false;
- x.Hostname = "iggy";
- x.Authenticate = false;
- };
+ BaseAddress = "127.0.0.1:8090",
+ Protocol = Protocol.Tcp
}, loggerFactory);
+
+await client.ConnectAsync();
```
-Iggy necessitates the use of `ILoggerFactory` to generate logs from locations
that are inaccessible to the user.
+The `ILoggerFactory` is required and used throughout the SDK for diagnostics
and debugging.
+
+### Configuration
-In addition to the basic configuration settings, Iggy provides support for
batching send/poll messages at intervals,
-which effectively decreases the frequency of network calls, this option is
enabled by default.
+The `IggyClientConfigurator` provides comprehensive configuration options:
```c#
-//---Snip---
-var bus = MessageStreamFactory.CreateMessageStream(options =>
+var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
- options.BaseAdress = "127.0.0.1:8090";
- options.Protocol = protocol;
- options.TlsSettings = x =>
+ BaseAddress = "127.0.0.1:8090",
+ Protocol = Protocol.Tcp,
+
+ // Buffer sizes (optional, default: 4096)
+ ReceiveBufferSize = 4096,
+ SendBufferSize = 4096,
+
+ // TLS/SSL configuration
+ TlsSettings = new TlsConfiguration
{
- x.Enabled = false;
- x.Hostname = "iggy";
- x.Authenticate = false;
- };
+ Enabled = true,
+ Hostname = "iggy",
+ Authenticate = true
+ },
- options.IntervalBatchingConfig = x =>
+ // Automatic reconnection with exponential backoff
+ ReconnectionSettings = new ReconnectionSettings
{
- x.Enabled = true;
- x.Interval = TimeSpan.FromMilliseconds(100);
- x.MaxMessagesPerBatch = 1000;
- x.MaxRequests = 4096;
- };
- options.MessagePollingSettings = x =>
+ 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
{
- x.Interval = TimeSpan.FromMilliseconds(100);
- x.StoreOffsetStrategy = StoreOffset.AfterProcessingEachMessage;
- };
+ Enabled = true,
+ Username = "your_username",
+ Password = "your_password"
+ }
}, loggerFactory);
+
+await client.ConnectAsync();
```
-### Creating and logging in a user
+## Authentication
+
+### User Login
-To begin, utilize the root account (note that the root account cannot be
removed or updated).
+Begin by using the root account (note: the root account cannot be removed or
updated):
```c#
-var response = await bus.LoginUser(new LoginUserRequest
-{
- Username = "iggy",
- Password = "iggy",
-});
+var response = await client.LoginUser("iggy", "iggy");
```
-Furthermore, after logging in, you have the option to create an account with
customizable `Permissions`.
+### Creating Users
+
+Create new users with customizable permissions:
```c#
-//---Snip---
-await bus.CreateUser(new CreateUserRequest
+var permissions = new Permissions
{
- Username = "test_user",
- Password = "pa55w0rD!@",
- Status = UserStatus.Active,
- Permissions = new Permissions
+ Global = new GlobalPermissions
{
- Global = new GlobalPermissions
- {
- ManageServers = true,
- ManageUsers = true,
- ManageStreams = true,
- ManageTopics = true,
- PollMessages = true,
- ReadServers = true,
- ReadStreams = true,
- ReadTopics = true,
- ReadUsers = true,
- SendMessages = true
- },
- Streams = new Dictionary<int, StreamPermissions>
- {
- {
- streamId, new StreamPermissions
- {
- ManageStream = true,
- ReadStream = true,
- SendMessages = true,
- PollMessages = true,
- ManageTopics = true,
- ReadTopics = true,
- Topics = new Dictionary<int, TopicPermissions>
- {
- {
- topicId, new TopicPermissions
- {
- ManageTopic = true,
- ReadTopic = true,
- PollMessages = true,
- SendMessages = true
- }
- }
- }
- }
- }
- }
+ ManageServers = true,
+ ManageUsers = true,
+ ManageStreams = true,
+ ManageTopics = true,
+ PollMessages = true,
+ ReadServers = true,
+ ReadStreams = true,
+ ReadTopics = true,
+ ReadUsers = true,
+ SendMessages = true
}
-});
+};
-var response = await bus.LoginUser(new LoginUserRequest
-{
- Username = "test_user",
- Password = "pa55w0rD!@",
-});
+await client.CreateUser("test_user", "secure_password", UserStatus.Active,
permissions);
+
+// Login with the new user
+var loginResponse = await client.LoginUser("test_user", "secure_password");
```
-Alternatively, once you've logged in, you can create a `Personal Access Token`
that can be reused for further logins.
+### Personal Access Tokens
+
+Create and use Personal Access Tokens (PAT) for programmatic access:
```c#
-var response = await bus.LoginUser(new LoginUserRequest
-{
- Username = "your_username",
- Password = "your_password",
-});
+// Create a PAT
+var patResponse = await client.CreatePersonalAccessTokenAsync("api-token",
3600);
-var patResponse = await bus.CreatePersonalAccessTokenAsync(new
CreatePersonalAccessTokenRequest
-{
- Name = "first-pat",
- Expiry = 60, // seconds from creation time
-});
-await bus.LoginWithPersonalAccessToken(new LoginWithPersonalAccessToken
-{
- Token = patResponse.Token
-});
+// Login with PAT
+await client.LoginWithPersonalAccessToken(patResponse.Token);
```
-### Creating first stream and topic
+## Streams and Topics
-In order to create stream use `CreateStreamAsync` method.
+### Creating Streams
```c#
-await bus.CreateStreamAsync(new StreamRequest
-{
- StreamId = 1,
- Name = "first-stream",
-});
+await client.CreateStreamAsync("my-stream");
```
-Every stream has a topic to which you can broadcast messages, for the purpose
of create one
-use `CreateTopicAsync` method.
+You can reference streams by either numeric ID or name:
```c#
-var streamId = Identifier.Numeric(1);
-await bus.CreateTopicAsync(streamId, new TopicRequest
-{
- Name = "first-topic",
- PartitionsCount = 3,
- TopicId = 1
-});
+var streamById = Identifier.Numeric(0);
+var streamByName = Identifier.String("my-stream");
```
-Notice that both Stream aswell as Topic use `-` instead of space in its name,
Iggy will replace any spaces in
-name with `-` instead, so keep that in mind.
+### Creating Topics
+
+Every stream contains topics for organizing messages:
+
+```c#
+var streamId = Identifier.String("my-stream");
+
+await client.CreateTopicAsync(
+ streamId,
+ name: "my-topic",
+ partitionsCount: 3,
+ compressionAlgorithm: CompressionAlgorithm.None,
+ replicationFactor: 1,
+ messageExpiry: 0, // 0 = never expire
+ maxTopicSize: 0 // 0 = unlimited
+);
+```
-### Sending messages
+Note: Stream and topic names use hyphens instead of spaces. Iggy automatically
replaces spaces with hyphens.
-To send messages you can use `SendMessagesAsync` method.
+## Publishing Messages
+
+### Sending Messages
+
+Send messages using the publisher interface:
```c#
-Func<byte[], byte[]> encryptor = static payload =>
+var streamId = Identifier.String("my-stream");
+var topicId = Identifier.String("my-topic");
+
+var messages = new List<Message>
{
- string aes_key = "AXe8YwuIn1zxt3FPWTZFlAa14EHdPAdN9FaZ9RQWihc=";
- string aes_iv = "bsxnWolsAyO7kCfWuyrnqg==";
+ new(Guid.NewGuid(), "Hello, Iggy!"u8.ToArray()),
+ new(1, "Another message"u8.ToArray())
+};
+
+await client.SendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.None(), // balanced partitioning
+ messages
+);
+```
- var key = Convert.FromBase64String(aes_key);
- var iv = Convert.FromBase64String(aes_iv);
+### Partitioning Strategies
- using Aes aes = Aes.Create();
- ICryptoTransform encryptor = aes.CreateEncryptor(key, iv);
+Control which partition receives each message:
- using MemoryStream memoryStream = new MemoryStream();
- using CryptoStream cryptoStream = new CryptoStream(memoryStream,
encryptor, CryptoStreamMode.Write);
- using BinaryWriter streamWriter = new BinaryWriter(cryptoStream);
- streamWriter.Write(payload);
+```c#
+// Balanced partitioning (default)
+Partitioning.None()
- return memoryStream.ToArray();
-};
+// Send to specific partition
+Partitioning.PartitionId(1)
-var messages = new List<Message>(); // your messages
-var streamId = Identifier.Numeric(1);
-var topicId = Identifier.Numeric(1);
-await bus.SendMessagesAsync(new MessageSendRequest
-{
- Messages = new List<Message>(),
- Partitioning = Partitioning.PartitionId(1),
- StreamId = streamId,
- TopicId = topicId,
-}, encryptor); //encryptor is optional
+// Key-based partitioning (string)
+Partitioning.EntityIdString("user-123")
+
+// Key-based partitioning (integer)
+Partitioning.EntityIdInt(12345)
+
+// Key-based partitioning (GUID)
+Partitioning.EntityIdGuid(Guid.NewGuid())
```
-The `Message` struct has two fields `Id` and `Payload`.
+### User-Defined Headers
+
+Add custom headers to messages with typed values:
```c#
-struct Message
+var headers = new Dictionary<HeaderKey, HeaderValue>
{
- public required MessageHeader Header { get; init; }
- public required byte[] Payload { get; init; }
- public Dictionary<HeaderKey, HeaderValue>? UserHeaders { get; init; }
-}
+ { new HeaderKey { Value = "correlation_id" },
HeaderValue.FromString("req-123") },
+ { new HeaderKey { Value = "priority" }, HeaderValue.FromInt32(1) },
+ { new HeaderKey { Value = "timeout" }, HeaderValue.FromInt64(5000) },
+ { new HeaderKey { Value = "confidence" }, HeaderValue.FromFloat(0.95f) },
+ { new HeaderKey { Value = "is_urgent" }, HeaderValue.FromBool(true) },
+ { new HeaderKey { Value = "request_id" },
HeaderValue.FromGuid(Guid.NewGuid()) }
+};
-public readonly struct MessageHeader
+var messages = new List<Message>
{
- public ulong Checksum { get; init; }
- public UInt128 Id { get; init; }
- public ulong Offset { get; init; }
- public DateTimeOffset Timestamp { get; init; }
- public ulong OriginTimestamp { get; init; }
- public int UserHeadersLength { get; init; }
- public int PayloadLength { get; init; }
-}
+ new(Guid.NewGuid(), "Message with headers"u8.ToArray(), headers)
+};
+
+await client.SendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.PartitionId(1),
+ messages
+);
```
-Furthermore, there's a generic overload for this method that takes binary
serializer as argument.
+## Consumer Groups
+
+### Creating Consumer Groups
+
+Coordinate message consumption across multiple consumers:
```c#
-//---Snip---
-Func<Envelope, byte[]> serializer = static envelope =>
-{
- Span<byte> buffer = stackalloc byte[envelope.MessageType.Length + 4 +
envelope.Payload.Length];
+var groupResponse = await client.CreateConsumerGroupAsync(
+ Identifier.String("my-stream"),
+ Identifier.String("my-topic"),
+ "my-consumer-group"
+);
+```
- BinaryPrimitives.WriteInt32LittleEndian(buffer[..4],
envelope.MessageType.Length);
-
Encoding.UTF8.GetBytes(envelope.MessageType).CopyTo(buffer[4..(envelope.MessageType.Length
+ 4)]);
-
Encoding.UTF8.GetBytes(envelope.Payload).CopyTo(buffer[(envelope.MessageType.Length
+ 4)..]);
+### Joining and Leaving Groups
- return buffer.ToArray();
-};
+**Note:** Join/Leave operations are only supported on TCP protocol and will
throw `FeatureUnavailableException` on HTTP.
+
+```c#
+// Join a consumer group
+await client.JoinConsumerGroupAsync(
+ Identifier.String("my-stream"),
+ Identifier.String("my-topic"),
+ Identifier.Numeric(1) // consumer ID
+);
+
+// Leave a consumer group
+await client.LeaveConsumerGroupAsync(
+ Identifier.String("my-stream"),
+ Identifier.String("my-topic"),
+ Identifier.Numeric(1) // consumer ID
+);
+```
-var messages = new List<Envelope>(); // your messages
-await bus.SendMessagesAsync(new MessageSendRequest<Envelope>
+## Consuming Messages
+
+### Fetching Messages
+
+Fetch a batch of messages:
+
+```c#
+var polledMessages = await client.PollMessagesAsync(new MessageFetchRequest
{
StreamId = streamId,
TopicId = topicId,
- Partitioning = Partitioning.PartitionId(1),
- Messages = messages
-},
-serializer,
-encryptor);
+ 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)}");
+}
```
-Both generic and non generic method accept optional `Headers` dictionary.
+### Polling Strategies
+
+Control where message consumption starts:
```c#
-//---Snip---
-var headers = new Dictionary<HeaderKey, HeaderValue>
-{
- { new HeaderKey { Value = "key_1".ToLower() },
HeaderValue.FromString("test-value-1") },
- { new HeaderKey { Value = "key_2".ToLower() }, HeaderValue.FromInt32(69) },
- { new HeaderKey { Value = "key_3".ToLower() },
HeaderValue.FromFloat(420.69f) },
- { new HeaderKey { Value = "key_4".ToLower() }, HeaderValue.FromBool(true)
},
- { new HeaderKey { Value = "key_5".ToLower() },
HeaderValue.FromBytes(byteArray) },
- { new HeaderKey { Value = "key_6".ToLower() }, HeaderValue.FromInt128(new
Int128(6969696969, 420420420)) },
- { new HeaderKey { Value = "key7".ToLower() },
HeaderValue.FromGuid(Guid.NewGuid()) }
-};
+// Start from a specific offset
+PollingStrategy.Offset(1000)
-await bus.SendMessagesAsync<Envelope>(new MessageSendRequest<Envelope>
-{
- StreamId = streamId,
- TopicId = topicId,
- Partitioning = Partitioning.PartitionId(1),
- Messages = messages
-},
-serializer,
-encryptor,
-headers);
+// 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()
```
-### Fetching Messages
+## Offset Management
-Fetching messages is done with `FetchMessagesAsync`.
+### Storing Offsets
+
+Store the current consumer position:
```c#
-Func<byte[], byte[]> decryptor = static payload =>
-{
- string aes_key = "AXe8YwuIn1zxt3FPWTZFlAa14EHdPAdN9FaZ9RQWihc=";
- string aes_iv = "bsxnWolsAyO7kCfWuyrnqg==";
+await client.StoreOffsetAsync(
+ Identifier.String("my-stream"),
+ Identifier.String("my-topic"),
+ Identifier.Numeric(1), // consumer ID
+ 0, // partition ID
+ 42 // offset value
+);
+```
- var key = Convert.FromBase64String(aes_key);
- var iv = Convert.FromBase64String(aes_iv);
+### Retrieving Offsets
- using Aes aes = Aes.Create();
- ICryptoTransform decryptor = aes.CreateDecryptor(key, iv);
+Get the current stored offset:
- using MemoryStream memoryStream = new MemoryStream(payload);
- using CryptoStream cryptoStream = new CryptoStream(memoryStream,
decryptor, CryptoStreamMode.Read);
- using BinaryReader binaryReader = new BinaryReader(cryptoStream);
+```c#
+var offsetInfo = await client.GetOffsetAsync(
+ Identifier.String("my-stream"),
+ Identifier.String("my-topic"),
+ Identifier.Numeric(1), // consumer ID
+ 0 // partition ID
+);
+
+Console.WriteLine($"Current offset: {offsetInfo.Offset}");
+```
- return binaryReader.ReadBytes(payload.Length);
-};
+### Deleting Offsets
+
+Clear stored offsets:
+
+```c#
+await client.DeleteOffsetAsync(
+ Identifier.String("my-stream"),
+ Identifier.String("my-topic"),
+ Identifier.Numeric(1), // consumer ID
+ 0 // partition ID
+);
+```
+
+## System Operations
+
+### Cluster Information
+
+Get cluster metadata and node information:
+
+```c#
+var metadata = await client.GetClusterMetadataAsync();
+```
+
+### Server Statistics
+
+Retrieve server performance metrics:
+
+```c#
+var stats = await client.GetStatsAsync();
+```
+
+### Health Checks
+
+Verify server connectivity:
-var messages = await bus.FetchMessagesAsync(new MessageFetchRequest
+```c#
+await client.PingAsync();
+```
+
+### Client Information
+
+Get information about connected clients:
+
+```c#
+var clients = await client.GetClientsAsync();
+var currentClient = await client.GetMeAsync();
+```
+
+## Event Subscription
+
+Subscribe to connection events:
+
+```c#
+// Subscribe to connection events
+client.SubscribeConnectionEvents(async connectionState =>
{
- StreamId = streamId,
- TopicId = topicId,
- Consumer = Consumer.New(1),
- Count = 1,
- PartitionId = 1,
- PollingStrategy = PollingStrategy.Next(),
- AutoCommit = true
-},
-decryptor);
+ Console.WriteLine($"Current connection state:
{connectionState.CurrentState}");
+
+ await SaveConnectionStateLog(connectionState.CurrentState);
+});
+
+// Unsubscribe
+client.UnsubscribeConnectionEvents(handler);
```
-Similarly, as with `SendMessagesAsync`, there's a generic overload that
accepts a binary deserializer.
+## Advanced: IggyPublisher
+
+High-level publisher with background sending, retries, and encryption:
```c#
-//---Snip---
-Func<byte[], Envelope> deserializer = serializedData =>
+var publisher = IggyPublisherBuilder.Create(
+ client,
+ Identifier.String("my-stream"),
+ Identifier.String("my-topic")
+)
+.WithBackgroundSending(enabled: true, batchSize: 100)
+.WithRetry(maxAttempts: 3)
+.Build();
+
+await publisher.InitAsync();
+
+var messages = new List<Message>
{
- Envelope envelope = new Envelope();
- int messageTypeLength = BitConverter.ToInt32(serializedData, 0);
- envelope.MessageType = Encoding.UTF8.GetString(serializedData, 4,
messageTypeLength);
- envelope.Payload = Encoding.UTF8.GetString(serializedData, 4 +
messageTypeLength, serializedData.Length - (4 + messageTypeLength));
- return envelope;
+ new(Guid.NewGuid(), "Message 1"u8.ToArray()),
+ new(0, "Message 2"u8.ToArray())
};
-var messages = await bus.FetchMessagesAsync<Envelope>(new MessageFetchRequest
+await publisher.SendMessages(messages);
+
+// Wait for all messages to be sent
+await publisher.WaitUntilAllSends();
+await publisher.DisposeAsync();
+```
+
+For automatic object serialization, use the typed variant:
+
+```c#
+class OrderSerializer : ISerializer<Order>
{
- StreamId = streamId,
- TopicId = topicId,
- Consumer = Consumer.New(1),
- Count = 1,
- PartitionId = 1,
- PollingStrategy = PollingStrategy.Next(),
- AutoCommit = true
-}, deserializer, decryptor);
+ public byte[] Serialize(Order data) =>
+ Encoding.UTF8.GetBytes(JsonSerializer.Serialize(data));
+}
+
+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> { /* ... */ });
```
-Beyond the `FetchMessagesAsync` functionality, there's also a
`PollMessagesAsync` method that spawns new thread which
-polls messages in background.
+## Advanced: IggyConsumer
+
+High-level consumer with automatic offset management and consumer groups:
```c#
-//---Snip---
-await foreach (var messageResponse in bus.PollMessagesAsync<Envelope>(new
PollMessagesRequest
+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())
{
- Consumer = Consumer.New(consumerId),
- Count = 1,
- TopicId = topicId,
- StreamId = streamId,
- PartitionId = 1,
- PollingStrategy = PollingStrategy.Next(),
-}, deserializer, decryptor))
+ var payload = Encoding.UTF8.GetString(message.Message.Payload);
+ Console.WriteLine($"Offset {message.CurrentOffset}: {payload}");
+}
+```
+
+For consumer groups (load-balanced across multiple consumers):
+
+```c#
+var consumer = IggyConsumerBuilder.Create(
+ client,
+ Identifier.String("my-stream"),
+ Identifier.String("my-topic"),
+ Consumer.Group("my-group")
+)
+.WithConsumerGroup("my-group", createIfNotExists: true)
+.WithPollingStrategy(PollingStrategy.Next())
+.WithAutoCommitMode(AutoCommitMode.AfterReceive)
+.Build();
+
+await consumer.InitAsync();
+
+await foreach (var message in consumer.ReceiveAsync())
{
- //handle the message response
+ Console.WriteLine($"Partition {message.PartitionId}:
{message.Message.Payload}");
}
+await consumer.DisposeAsync();
```
-It is worth noting that every method (except `PollMessagesAsync`) will throw
an `InvalidResponseException` when
-encountering an error.
+For automatic deserialization:
-If you register `IIggyClient` in a dependency injection container, you will
have access to interfaces
-that encapsulate smaller parts of the system `IIggyStream` `IIggyTopic`
`IIggyPublisher` `IIggyConsumer`
-`IIggyConsumerGroup` `IIggyOffset`
-`IIggyPartition` `IIggyUsers` `IIggyUtils`
+```c#
+class OrderDeserializer : IDeserializer<OrderEvent>
+{
+ public OrderEvent Deserialize(byte[] data) =>
+ JsonSerializer.Deserialize<OrderEvent>(Encoding.UTF8.GetString(data))!;
+}
-For more information about how Iggy works check its
[documentation](https://iggy.apache.org/docs/)
+var consumer = IggyConsumerBuilder<OrderEvent>.Create(
+ client,
+ Identifier.String("orders-stream"),
+ Identifier.String("orders-topic"),
+ Consumer.Group("order-processors"),
+ new OrderDeserializer()
+)
+.WithAutoCommitMode(AutoCommitMode.Auto)
+.Build();
-## Producer / Consumer Sample
+await consumer.InitAsync();
-To run the samples, first get [Iggy](https://github.com/apache/iggy), Run the
server with `cargo run --bin iggy-server`,
-then get the SDK, cd into `Iggy_SDK`
-and run following commands: `dotnet run -c Release --project
Iggy_Sample_Producer` for producer,
-`dotnet run -c Release --project Iggy_Sample_Consumer`
-for consumer.
+await foreach (var message in consumer.ReceiveDeserializedAsync())
+{
+ if (message.Status == MessageStatus.Success)
+ {
+ Console.WriteLine($"Order: {message.Data?.OrderId}");
+ }
+}
+```
+
+## API Reference
+
+The SDK provides the following main interfaces:
+
+- **IIggyClient** - Main client interface (aggregates all features)
+- **IIggyPublisher** - High-level message publishing interface
+- **IIggyConsumer** - High-level message consumption interface
+- **IIggyStream** - Stream management
+- **IIggyTopic** - Topic management
+- **IIggyOffset** - Offset management
+- **IIggyConsumerGroup** - Consumer group operations
+- **IIggyPartition** - Partition operations
+- **IIggyUsers** - User and authentication management
+- **IIggySystem** - System and cluster operations
+- **IIggyPersonalAccessToken** - Personal access token management
+
+Additionally, builder-based APIs are available:
+
+- **IggyPublisherBuilder** / **IggyPublisherBuilder<T>** - Fluent publisher
configuration
+- **IggyConsumerBuilder** / **IggyConsumerBuilder<T>** - Fluent consumer
configuration
+
+## Running Examples
+
+Examples are located in `examples/csharp/` in root iggy directory.
+
+- Start the Iggy server:
+
+```bash
+cargo run --bin iggy-server
+```
+
+- Run the producer example:
+
+```bash
+dotnet run -c Release --project Iggy_SDK.Examples.GettingStarted.Producer
+```
+
+- Run the consumer example:
+
+```bash
+dotnet run -c Release --project Iggy_SDK.Examples.GettingStarted.Consumer
+```
## Integration Tests
-Integration tests are located in
`Iggy_SDK/Iggy_Sample_Producer/IntegrationTests` folder.
-Tests can be run against a dockerized Iggy server with TestContainers or local
Iggy server.
-To run with a local Iggy server, the environment variable `IGGY_SERVER_HOST`
needs to be set.
+Integration tests are located in `Iggy_SDK.Tests.Integration/`. Tests can run
against:
+
+- A dockerized Iggy server with TestContainers
+- A local Iggy server (set `IGGY_SERVER_HOST` environment variable)
+
+## Useful Resources
+
+- [Iggy Documentation](https://iggy.apache.org/docs/)
+- [NuGet Package](https://www.nuget.org/packages/Apache.Iggy)
## ROADMAP - TODO
-- [ ] Consumer/Publisher client - WIP
- [ ] Error handling with status codes and descriptions
- [ ] Add support for `ASP.NET Core` Dependency Injection
