This is an automated email from the ASF dual-hosted git repository. lukaszzborek pushed a commit to branch dotnet-docs in repository https://gitbox.apache.org/repos/asf/iggy-website.git
commit 8ca72df5b8686939d99cb64ccc09137502ac41a6 Author: Łukasz Zborek <[email protected]> AuthorDate: Sat Jul 25 11:09:05 2026 +0200 docs: update csharp client docs --- content/docs/sdk/csharp/examples.mdx | 203 +++++++--- content/docs/sdk/csharp/guide.mdx | 622 +++++++++++++++++++++++++++++ content/docs/sdk/csharp/high-level-sdk.mdx | 313 +++++++++++++++ content/docs/sdk/csharp/intro.mdx | 157 +++++--- content/docs/sdk/csharp/meta.json | 2 +- 5 files changed, 1170 insertions(+), 127 deletions(-) diff --git a/content/docs/sdk/csharp/examples.mdx b/content/docs/sdk/csharp/examples.mdx index a0155935..87005329 100644 --- a/content/docs/sdk/csharp/examples.mdx +++ b/content/docs/sdk/csharp/examples.mdx @@ -2,111 +2,186 @@ title: Examples --- -## Basic producer +These samples use the [High-level SDK](/docs/sdk/csharp/high-level-sdk) — the recommended way to build producers and consumers. For the low-level, per-call equivalents, see the [Guide](/docs/sdk/csharp/guide). + +## Producer + +A publisher that creates the stream and topic if missing, batches sends in the background, and retries failures: ```csharp using System.Text; using Apache.Iggy; using Apache.Iggy.Configuration; using Apache.Iggy.Enums; -using Apache.Iggy.Exceptions; +using Apache.Iggy.Extensions; using Apache.Iggy.Factory; -using Apache.Iggy.Kinds; using Apache.Iggy.Messages; -const string StreamName = "dev"; -const string TopicName = "events"; - -var client = IggyClientFactory.CreateClient(new IggyClientConfigurator() +var client = IggyClientFactory.CreateClient(new IggyClientConfigurator { BaseAddress = "127.0.0.1:8090", - Protocol = Protocol.Tcp, + Protocol = Protocol.Tcp }); await client.ConnectAsync(); -await client.LoginUser("iggy", "iggy"); +await client.LoginUserAsync("iggy", "iggy"); -// Create stream and topic if they don't exist -try -{ - await client.CreateStreamAsync(StreamName); -} -catch (InvalidResponseException) -{ - Console.WriteLine("Stream already exists."); -} +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(); -try -{ - await client.CreateTopicAsync( - Identifier.String(StreamName), - TopicName, - 2, - CompressionAlgorithm.None - ); -} -catch (InvalidResponseException) -{ - Console.WriteLine("Topic already exists."); -} +await publisher.InitAsync(); -// Send messages -var partitioning = Partitioning.PartitionId(0); -for (int i = 0; i < 100; i++) +for (var i = 0; i < 100; i++) { var payload = Encoding.UTF8.GetBytes($"Event #{i}"); - var messages = new List<Message> { new Message(Guid.NewGuid(), payload) }; - await client.SendMessagesAsync( - Identifier.String(StreamName), - Identifier.String(TopicName), - partitioning, - messages - ); + 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"); ``` -## Basic consumer +## Consumer group + +A consumer that creates and joins a consumer group, commits offsets after each received message, and surfaces polling errors: ```csharp using System.Text; using Apache.Iggy; using Apache.Iggy.Configuration; -using Apache.Iggy.Contracts; +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. + +## Typed messages + +A typed publisher/consumer pair that (de)serializes a record as JSON. Note that 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](/docs/sdk/csharp/high-level-sdk#typed-consumer)): + +```csharp +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() +var client = IggyClientFactory.CreateClient(new IggyClientConfigurator { BaseAddress = "127.0.0.1:8090", - Protocol = Protocol.Tcp, + Protocol = Protocol.Tcp }); await client.ConnectAsync(); -await client.LoginUser("iggy", "iggy"); - -// Poll messages -var consumer = Consumer.New(1); -var offset = 0ul; -uint messagesPerBatch = 100; - -var polledMessages = await client.PollMessagesAsync( - Identifier.String("dev"), - Identifier.String("events"), - 0, - consumer, - PollingStrategy.Offset(offset), - messagesPerBatch, - false +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); -foreach (var message in polledMessages.Messages) +class OrderSerializer : ISerializer<OrderEvent> { - var payload = Encoding.UTF8.GetString(message.Payload); - Console.WriteLine($"Offset: {message.Header.Offset}, Payload: {payload}"); + 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)!; } ``` -For the full source code, see the [examples/csharp](https://github.com/apache/iggy/tree/master/examples/csharp) directory. +## More examples + +The [examples/csharp](https://github.com/apache/iggy/tree/master/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 diff --git a/content/docs/sdk/csharp/guide.mdx b/content/docs/sdk/csharp/guide.mdx new file mode 100644 index 00000000..b1379e9b --- /dev/null +++ b/content/docs/sdk/csharp/guide.mdx @@ -0,0 +1,622 @@ +--- +title: 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](/docs/sdk/csharp/high-level-sdk). + +All examples assume you already have a connected, authenticated client (see [Creating a client](/docs/sdk/csharp/intro#creating-a-client)). + +## Client configuration + +`IggyClientConfigurator` exposes the full set of connection options — buffer sizes, TLS, automatic reconnection with exponential backoff, and auto-login: + +```csharp +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](#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: + +```csharp +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): + +```csharp +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: + +```csharp +var byId = Identifier.Numeric(0); +var byName = Identifier.String("my-stream"); +``` + +## Authentication + +### User login + +Begin with the root account (`iggy` / `iggy`): + +```csharp +var response = await client.LoginUserAsync("iggy", "iggy"); + +// Log out the currently authenticated user +await client.LogoutUserAsync(); +``` + +### Creating users + +Create new users with customizable permissions: + +```csharp +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: + +```csharp +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 + +```csharp +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: + +```csharp +// 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 + +```csharp +// 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: + +```csharp +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`. + +```csharp +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: + +```csharp +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: + +```csharp +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](/docs/sdk/csharp/high-level-sdk#publishing-with-rented-batches). + +### Partitioning strategies + +Control which partition receives each message: + +```csharp +// 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: + +```csharp +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): + +```csharp +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: + +```csharp +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: + +```csharp +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: + +```csharp +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](/docs/sdk/csharp/high-level-sdk#consuming-with-rented-buffers). + +### Polling strategies + +Control where consumption starts: + +```csharp +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: + +```csharp +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. + +```csharp +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. + +```csharp +await client.JoinConsumerGroupAsync(streamId, topicId, Identifier.String("my-consumer-group")); +await client.LeaveConsumerGroupAsync(streamId, topicId, Identifier.String("my-consumer-group")); +``` + +## System operations + +```csharp +// 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: + +```csharp +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. + +```csharp +await client.DeleteSegmentsAsync( + Identifier.String("my-stream"), + Identifier.String("my-topic"), + partitionId: 1, + segmentsCount: 2 +); +``` diff --git a/content/docs/sdk/csharp/high-level-sdk.mdx b/content/docs/sdk/csharp/high-level-sdk.mdx new file mode 100644 index 00000000..04eb51c8 --- /dev/null +++ b/content/docs/sdk/csharp/high-level-sdk.mdx @@ -0,0 +1,313 @@ +--- +title: High-level SDK +--- + +The per-call [`IIggyClient`](/docs/sdk/csharp/guide) 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 — no per-message allocation + +Both are 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: + +```csharp +using Apache.Iggy; +using Apache.Iggy.Kinds; +using Apache.Iggy.Messages; +using Apache.Iggy.Publishers; + +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(); +await publisher.DisposeAsync(); +``` + +### Publisher builder options + +| Method | Description | +|--------|-------------| +| `WithConnection(protocol, address, login, password, receiveBufferSize = 4096, sendBufferSize = 4096, reconnectionSettings = null)` | Connection settings — only used when the builder creates its own client (i.e. the `Create(streamId, topicId)` overload, without an existing client) | +| `WithPartitioning(partitioning)` | Routing strategy for produced messages (default: balanced) | +| `CreateStreamIfNotExists(name)` | Auto-create the stream on `InitAsync` if missing | +| `CreateTopicIfNotExists(name, topicPartitionsCount = 1, compressionAlgorithm = None, replicationFactor = null, messageExpiry = TimeSpan.Zero, maxTopicSize = 0)` | Auto-create the topic on `InitAsync` if missing | +| `WithRetry(enabled = true, maxAttempts = 3, initialDelay = 100ms, maxDelay = 10s, backoffMultiplier = 2.0)` | Retry failed sends with exponential backoff | +| `WithBackgroundSending(enabled = true, queueCapacity = 10000, batchSize = 100, flushInterval = 100ms, disposalTimeout = 5s)` | Queue and flush sends in the background for higher throughput | +| `WithEncryptor(encryptor)` | Client-side payload encryption. Only valid on a builder-created client — for an external client, set `IggyClientConfigurator.MessageEncryptor` instead | +| `SubscribeOnBackgroundError(handler)` | Observe background-processing errors (only fires when background sending is enabled) | +| `SubscribeOnMessageBatchFailed(handler)` | Fires when a batch fails after all retries are exhausted (only with background sending — direct sends throw to the caller instead) | +| `WithLogger(loggerFactory)` | Logger factory for diagnostics | + +**`WithRetry` parameters** + +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `enabled` | `bool` | `true` | Whether retry is enabled | +| `maxAttempts` | `int` | `3` | Maximum retry attempts | +| `initialDelay` | `TimeSpan?` | `100ms` | Delay before the first retry | +| `maxDelay` | `TimeSpan?` | `10s` | Maximum delay between retries | +| `backoffMultiplier` | `double` | `2.0` | Exponential backoff multiplier | + +**`WithBackgroundSending` parameters** + +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `enabled` | `bool` | `true` | Whether background sending is enabled | +| `queueCapacity` | `int` | `10000` | Max queued send calls (one slot per `SendMessagesAsync` call, regardless of batch size) | +| `batchSize` | `int` | `100` | Messages sent per batch | +| `flushInterval` | `TimeSpan?` | `100ms` | Interval at which pending messages are flushed | +| `disposalTimeout` | `TimeSpan?` | `5s` | How long `DisposeAsync` waits for the background processor to drain | + +A batch also flushes once its accumulated payload reaches `IggyPublisherConfig.BackgroundMaxBatchBytes` (default 256 KB, `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. + +### Publishing with rented batches + +`SendMessagesAsync` takes messages whose payloads each live in their own `byte[]`. 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`: + +```csharp +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); +``` + +`RentedMessageBatchBuilder`: + +| Member | Description | +|--------|-------------| +| `new RentedMessageBatchBuilder(sizeHint = 1024)` | Initial capacity hint (bytes) for the rented buffer | +| `Add(payload, id = null, userHeaders = null)` | Copies a `ReadOnlySpan<byte>` payload into the shared buffer | +| `Add(state, writePayload, id = null, userHeaders = null)` | Writes the payload via an `Action<TState, IBufferWriter<byte>>` callback directly into the buffer | +| `Count` | Number of payloads added so far | +| `Build()` | Materializes the `RentedMessageBatch`; the builder must not be used afterwards | + +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>`: + +```csharp +class OrderSerializer : ISerializer<Order> +{ + public void Serialize(Order data, IBufferWriter<byte> writer) => + writer.Write(JsonSerializer.SerializeToUtf8Bytes(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> { /* ... */ }); +``` + +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`: + +```csharp +var publisher = IggyPublisherBuilder<Order>.Create( + client, + Identifier.String("orders-stream"), + Identifier.String("orders-topic"), + new SystemTextJsonSerializer<Order>() +).Build(); +``` + +There is no built-in deserializer counterpart — consumers implement `IDeserializer<T>` themselves (see [Typed consumer](#typed-consumer)). + +## IggyConsumer + +Configure a consumer, initialize it, then iterate. `ReceiveAsync` returns an `IAsyncEnumerable<ReceivedMessage>`: + +```csharp +using System.Text; +using Apache.Iggy; +using Apache.Iggy.Consumers; +using Apache.Iggy.Kinds; + +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 + +| Method | Description | +|--------|-------------| +| `WithConnection(protocol, address, login, password, receiveBufferSize = 4096, sendBufferSize = 4096, reconnectionSettings = null)` | Connection settings — only used when the builder creates its own client | +| `WithPartitionId(partitionId)` | Consume from a specific partition | +| `WithPollingStrategy(pollingStrategy)` | Where to start consuming (default: `Offset(0)`). An `Offset(...)` strategy is advanced client-side after each poll; other strategies (e.g. `Next()`) are sent as-is and rely on server-side offset tracking | +| `WithBatchSize(batchSize)` | Messages fetched per poll (default: `100`) | +| `WithAutoCommitMode(mode)` | Offset auto-commit behavior (see below) | +| `WithConsumerGroup(groupName, createIfNotExists = true, joinGroup = true)` | Create and/or join a consumer group | +| `WithPollingInterval(interval)` | Delay between polls to throttle server requests (default: `100ms`); `TimeSpan.Zero` disables throttling | +| `WithEncryptor(encryptor)` | Client-side payload decryption. Only valid on a builder-created client — for an external client, set `IggyClientConfigurator.MessageEncryptor` instead. Cannot be combined with `AutoCommitMode.Auto` (the builder throws — a decryption failure would silently skip an already-committed batch) | +| `SubscribeOnPollingError(handler)` | Observe polling errors | +| `WithLogger(loggerFactory)` | Logger factory for diagnostics | + +Auto-commit modes (`AutoCommitMode`): + +| Mode | Description | +|------|-------------| +| `Auto` | Commit the offset while polling | +| `AfterReceive` | Commit after each message is received | +| `Disabled` | Commit manually | + +### Manual offset control + +With `AutoCommitMode.Disabled`, commit offsets yourself via the consumer's `StoreOffsetAsync`: + +```csharp +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; pass `resetLastPolled: true` to also move the consumer's cached last-polled position so the next poll resumes past the stored offset. `DeleteOffsetAsync(partitionId)` clears the stored offset. + +### Consumer groups + +Pass a `Consumer.Group(...)` and let the builder create and join the group for load-balanced consumption: + +```csharp +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}"); +} + +await consumer.DisposeAsync(); +``` + +### Consuming with rented buffers + +`ReceiveAsync` copies each payload into its own `byte[]`. `ReceiveRentedAsync` is the high-level counterpart of the low-level [rented poll API](/docs/sdk/csharp/guide#polling-with-rented-buffers): 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: + +```csharp +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`](/docs/sdk/csharp/guide#polling-with-rented-buffers)), `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`. Forgetting to dispose a message never corrupts data — the batch buffer simply isn't returned to the pool and is reclaimed by the GC as an ordinary allocation. + +### 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: + +```csharp +class OrderDeserializer : IDeserializer<OrderEvent> +{ + public OrderEvent Deserialize(ReadOnlyMemory<byte> data) => + JsonSerializer.Deserialize<OrderEvent>(data.Span)!; +} + +var builder = IggyConsumerBuilder<OrderEvent>.Create( + client, + Identifier.String("orders-stream"), + Identifier.String("orders-topic"), + Consumer.Group("order-processors"), + new OrderDeserializer() +); +builder.WithAutoCommitMode(AutoCommitMode.AfterReceive); + +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}"); + } +} +``` + +> **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. diff --git a/content/docs/sdk/csharp/intro.mdx b/content/docs/sdk/csharp/intro.mdx index 441d8d53..ab6761bb 100644 --- a/content/docs/sdk/csharp/intro.mdx +++ b/content/docs/sdk/csharp/intro.mdx @@ -2,7 +2,9 @@ title: C# SDK --- -The Iggy SDK for C# is a library that allows you to interact with the Iggy API from your .NET applications. It supports TCP and HTTP transports. The package is available on [NuGet](https://www.nuget.org/packages/Apache.Iggy/) and the source code can be found on [GitHub](https://github.com/apache/iggy/tree/master/foreign/csharp). +The Iggy SDK for C# is a modern, async-first client library for interacting with an Iggy message streaming server from your .NET applications. It supports TCP and HTTP transports. The package is available on [NuGet](https://www.nuget.org/packages/Apache.Iggy/) and the source code lives on [GitHub](https://github.com/apache/iggy/tree/master/foreign/csharp). + +The SDK is built around the `IIggyClient` interface, which aggregates every feature (publishing, consuming, stream/topic management, users, offsets, consumer groups, and system operations). For the low-level per-call API and the full configuration reference, see the [Guide](/docs/sdk/csharp/guide). For the ergonomic, batteries-included producer/consumer abstractions, see the [High-level SDK](/docs/sdk/csharp/high-level-sdk). ## Installation @@ -10,8 +12,60 @@ The Iggy SDK for C# is a library that allows you to interact with the Iggy API f dotnet add package Apache.Iggy ``` +## Supported protocols + +The SDK supports two transport protocols: + +- **TCP** — binary protocol for optimal performance and lower latency (recommended) +- **HTTP** — RESTful JSON API for stateless operations + +Some operations are TCP-only and throw `FeatureUnavailableException` on HTTP: joining/leaving a consumer group, `GetMeAsync`, and `DeleteSegmentsAsync`. + +## Creating a client + +Create a client with `IggyClientFactory.CreateClient`, then call `ConnectAsync`: + +```csharp +using Apache.Iggy.Configuration; +using Apache.Iggy.Enums; +using Apache.Iggy.Factory; + +var client = IggyClientFactory.CreateClient(new IggyClientConfigurator +{ + BaseAddress = "127.0.0.1:8090", + Protocol = Protocol.Tcp +}); + +await client.ConnectAsync(); +await client.LoginUserAsync("iggy", "iggy"); +``` + +Optionally, provide an `ILoggerFactory` for diagnostics (defaults to `NullLoggerFactory.Instance`): + +```csharp +using Microsoft.Extensions.Logging; + +var loggerFactory = LoggerFactory.Create(builder => +{ + builder + .AddFilter("Apache.Iggy", LogLevel.Information) + .AddConsole(); +}); + +var client = IggyClientFactory.CreateClient(new IggyClientConfigurator +{ + BaseAddress = "127.0.0.1:8090", + Protocol = Protocol.Tcp, + LoggerFactory = loggerFactory +}); +``` + +`IggyClientConfigurator` also exposes buffer sizes, TLS, automatic reconnection with exponential backoff, auto-login (so you can skip the explicit `LoginUserAsync` call), and client-side message encryption. See [Client configuration](/docs/sdk/csharp/guide#client-configuration) for the full reference. + ## Quick start +These samples use the [High-level SDK](/docs/sdk/csharp/high-level-sdk) — the recommended way to build producers and consumers. For the equivalent low-level, per-call flow, see the [Guide](/docs/sdk/csharp/guide). + ### Producer ```csharp @@ -19,42 +73,35 @@ 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.Kinds; using Apache.Iggy.Messages; -const string StreamName = "sample-stream"; -const string TopicName = "sample-topic"; - -var client = IggyClientFactory.CreateClient(new IggyClientConfigurator() +var client = IggyClientFactory.CreateClient(new IggyClientConfigurator { BaseAddress = "127.0.0.1:8090", - Protocol = Protocol.Tcp, + Protocol = Protocol.Tcp }); await client.ConnectAsync(); -await client.LoginUser("iggy", "iggy"); - -await client.CreateStreamAsync(StreamName); -await client.CreateTopicAsync( - Identifier.String(StreamName), - TopicName, - 1, - CompressionAlgorithm.None -); - -var partitioning = Partitioning.PartitionId(0); -for (int i = 0; i < 10; i++) +await client.LoginUserAsync("iggy", "iggy"); + +var publisher = client.CreatePublisherBuilder( + Identifier.String("sample-stream"), + Identifier.String("sample-topic")) + .CreateStreamIfNotExists("sample-stream") + .CreateTopicIfNotExists("sample-topic") + .Build(); + +await publisher.InitAsync(); + +for (var i = 0; i < 10; i++) { var payload = Encoding.UTF8.GetBytes($"message-{i}"); - var messages = new List<Message> { new Message(Guid.NewGuid(), payload) }; - await client.SendMessagesAsync( - Identifier.String(StreamName), - Identifier.String(TopicName), - partitioning, - messages - ); + await publisher.SendMessagesAsync(new List<Message> { new(Guid.NewGuid(), payload) }); } + +await publisher.DisposeAsync(); ``` ### Consumer @@ -63,56 +110,42 @@ for (int i = 0; i < 10; i++) using System.Text; using Apache.Iggy; using Apache.Iggy.Configuration; -using Apache.Iggy.Contracts; +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() +var client = IggyClientFactory.CreateClient(new IggyClientConfigurator { BaseAddress = "127.0.0.1:8090", - Protocol = Protocol.Tcp, + Protocol = Protocol.Tcp }); await client.ConnectAsync(); -await client.LoginUser("iggy", "iggy"); +await client.LoginUserAsync("iggy", "iggy"); -var consumer = Consumer.New(1); -var offset = 0ul; -uint messagesPerBatch = 10; - -while (true) -{ - var polledMessages = await client.PollMessagesAsync( +var consumer = client.CreateConsumerBuilder( Identifier.String("sample-stream"), Identifier.String("sample-topic"), - 0, - consumer, - PollingStrategy.Offset(offset), - messagesPerBatch, - false - ); - - if (!polledMessages.Messages.Any()) break; - - offset += (ulong)polledMessages.Messages.Count; - foreach (var message in polledMessages.Messages) - { - var payload = Encoding.UTF8.GetString(message.Payload); - Console.WriteLine($"Offset: {message.Header.Offset}, Payload: {payload}"); - } + Consumer.New(1)) + .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($"Offset {message.CurrentOffset}: {payload}"); } ``` -## Examples - -Working examples are available in the [examples/csharp](https://github.com/apache/iggy/tree/master/examples/csharp) directory. The following example sets are included: +`ReceiveAsync` polls indefinitely — pass a `CancellationToken` or `break` out of the loop to stop. -- **GettingStarted** - basic producer and consumer -- **Basic** - producer and consumer with settings -- **NewSdk** - new high-level SDK API patterns -- **MessageEnvelope** - working with message envelopes -- **MessageHeaders** - custom message headers -- **TcpTls** - TLS-encrypted TCP connections +## Next steps -The solution can be opened with Visual Studio or built with `dotnet build`. +- [Guide](/docs/sdk/csharp/guide) — client configuration reference and the full API surface: auth, streams, topics, partitions, publishing, consuming, offsets, consumer groups, system operations +- [High-level SDK](/docs/sdk/csharp/high-level-sdk) — `IggyPublisher` / `IggyConsumer` with background sending, retries, auto-commit, typed (de)serialization, and pooled (rented) buffers for allocation-free hot paths +- [Examples](/docs/sdk/csharp/examples) — producer, consumer-group, and typed-message samples, plus links to runnable projects diff --git a/content/docs/sdk/csharp/meta.json b/content/docs/sdk/csharp/meta.json index 6c1e2caa..358882d9 100644 --- a/content/docs/sdk/csharp/meta.json +++ b/content/docs/sdk/csharp/meta.json @@ -1 +1 @@ -{"title": "C#", "pages": ["intro", "examples"]} +{"title": "C#", "pages": ["intro", "guide", "high-level-sdk", "examples"]}
