You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/12 10:25:27 UTC
(iggy-website) 08/13: fix(docs): align SDK guides with server 0.9.0
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch docs/align-with-topic-durability
in repository https://gitbox.apache.org/repos/asf/iggy-website.git
commit 443e2fc89ebc23d270435248853fa53540bd64a3
Author: hubcio <h....@gmail.com>
AuthorDate: Fri Sep 11 12:59:55 2026 +0200
fix(docs): align SDK guides with server 0.9.0
SDK setup and examples used incompatible releases, stale APIs and
incorrect authentication, offset and ownership guarantees.
Align the language guides with verified source or edge packages.
Correct runnable samples and describe transport, retry, durability
and example lifecycle limits.
---
content/docs/sdk/connection-strings.mdx | 96 ++++++++++++---------
content/docs/sdk/cpp/intro.mdx | 15 ++--
content/docs/sdk/csharp/examples.mdx | 47 +++++++---
content/docs/sdk/csharp/guide.mdx | 132 ++++++++++++++++-------------
content/docs/sdk/csharp/high-level-sdk.mdx | 90 +++++++++++---------
content/docs/sdk/csharp/intro.mdx | 82 +++++++++++-------
content/docs/sdk/go/examples.mdx | 14 +--
content/docs/sdk/go/intro.mdx | 36 +++++---
content/docs/sdk/introduction.mdx | 35 ++++----
content/docs/sdk/java/examples.mdx | 20 +++--
content/docs/sdk/java/intro.mdx | 52 +++++++++---
content/docs/sdk/node/examples.mdx | 47 +++++-----
content/docs/sdk/node/intro.mdx | 30 ++++---
content/docs/sdk/php/intro.mdx | 32 ++++---
content/docs/sdk/python/examples.mdx | 16 ++--
content/docs/sdk/python/intro.mdx | 33 +++++---
content/docs/sdk/rust/examples.mdx | 17 +++-
content/docs/sdk/rust/high-level-sdk.mdx | 76 ++++++++++-------
content/docs/sdk/rust/intro.mdx | 24 ++++--
content/docs/sdk/rust/stream-builder.mdx | 48 ++++++-----
20 files changed, 568 insertions(+), 374 deletions(-)
diff --git a/content/docs/sdk/connection-strings.mdx b/content/docs/sdk/connection-strings.mdx
index fbc92bb0..e242ff85 100644
--- a/content/docs/sdk/connection-strings.mdx
+++ b/content/docs/sdk/connection-strings.mdx
@@ -13,7 +13,9 @@ Connection strings are supported by the Rust SDK and the SDKs that wrap it (Pyth
iggy[+transport]://credentials@host:port[?option=value&option=value]
```
-The host and port are **always required**. Options are appended as `key=value` pairs separated by `&`.
+The host and port are **always required**. Use a hostname or IPv4 address. The parser rejects IPv6 forms such as `[::1]:8090`. Options are appended as `key=value` pairs separated by `&`. Scheme names and option keys are case-sensitive.
+
+The parser does not percent-decode credentials or option values. Reserved separators such as `:` and `@` cannot be escaped inside a username or password; use a PAT for credentials that contain them. Duplicate option keys use the last value. Boolean options enable only for the literal `true`; use `false` to disable them.
**Unknown option keys are hard errors.** The parser rejects the whole string with `InvalidConnectionString` instead of ignoring the key, so a typo fails fast rather than silently falling back to a default.
@@ -36,68 +38,76 @@ Two forms are accepted:
**Username and password**, separated by a colon. Both parts must be non-empty:
```
-iggy://username:password@localhost:8090
+iggy://username:password@127.0.0.1:8090
```
**Personal Access Token**: any credential without a colon is treated as a PAT. Server-minted tokens are plain base64 strings with no prefix, so paste the token exactly as the server returned it:
```
-iggy://<personal-access-token>@localhost:8090
+iggy://<personal-access-token>@127.0.0.1:8090
```
**Do not prepend anything** to the token. Strings like `iggypat-...` appear only in Iggy's own test code. A real token with a prefix added will fail to log in.
-The default `iggy`/`iggy` root credentials **only exist** when the server was started with `--with-default-root-credentials` (or with `IGGY_ROOT_USERNAME`/`IGGY_ROOT_PASSWORD` set) on its first boot. Otherwise the root user gets a generated password and samples using `iggy:iggy` fail with `InvalidCredentials`:
+The examples use `iggy`/`iggy`. For a new local development instance, with no `IGGY_ROOT_USERNAME` or `IGGY_ROOT_PASSWORD` overrides, use:
```bash
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
+The environment takes precedence over the flag. Stored or recovered root credentials are not replaced by bootstrap settings. `--fresh` deletes this replica's local data; in a cluster it can recover committed state, including credentials, from another replica. Use this reset only for disposable development data. A new standalone server without explicit credentials generates a root password; a new cluster requires explicit root credentials.
+
+TCP, QUIC, and WebSocket apply connection-string credentials during `connect()`. HTTP requires an explicit login, as described below.
+
## Duration values
-Options typed as durations take human-readable values such as `5s`, `500ms`, or `1m`, unless a table below says the value is a plain number.
+Options typed as durations take human-readable values such as `5s`, `500ms`, or `1m`, unless a table below says the value is a plain number. Heartbeat and reconnection intervals must be greater than zero; the reconnection cooldown can be zero.
## TCP options
| Key | Description | Default |
|-----|-------------|---------|
| `tls` | Enable TLS (`true`/`false`) | `false` |
-| `tls_domain` | Domain name for TLS validation | empty |
-| `tls_ca_file` | Path to a CA certificate file | none |
-| `reconnection_retries` | Number of reconnection attempts, or `unlimited` | `unlimited` |
-| `reconnection_interval` | Duration between reconnection attempts | `1s` |
-| `reestablish_after` | Duration to wait before reestablishing the connection | `5s` |
+| `tls_domain` | Name for TLS validation; when empty, use the dialed hostname or IP | empty |
+| `tls_ca_file` | PEM file of trusted CA certificates, replacing the built-in roots | none |
+| `reconnection_retries` | Retry passes over known endpoints after the initial pass, or `unlimited` | `unlimited` |
+| `reconnection_interval` | Duration between retry passes | `1s` |
+| `reestablish_after` | Cooldown measured from the last connection establishment | `5s` |
| `heartbeat_interval` | Duration between heartbeats | `5s` |
| `nodelay` | Enable `TCP_NODELAY` (`true`/`false`) | `false` |
```
-iggy://iggy:iggy@localhost:8090?tls=true&tls_domain=example.com&reconnection_retries=5&heartbeat_interval=3s&nodelay=true
+iggy://iggy:iggy@127.0.0.1:8090?tls=true&tls_domain=localhost&tls_ca_file=core/certs/iggy_ca_cert.pem&reconnection_retries=5&heartbeat_interval=3s&nodelay=true
```
+With `tls=true`, TCP always validates the server certificate. Without `tls_ca_file`, it uses the built-in Mozilla roots. An explicit CA file can trust a private CA. The example above runs from the repository root against a TLS-enabled listener using the test certificate and key in `core/certs/iggy_cert.pem` and `core/certs/iggy_key.pem`. Those certificates are for local testing; see [Security](/docs/server/security) for TLS configuration.
+
## QUIC options
The QUIC reconnection keys **differ from TCP**: `reconnection_max_retries` (not `reconnection_retries`) and `reconnection_reestablish_after` (not `reestablish_after`). Using the TCP names in a QUIC string is a hard error.
| Key | Description | Default |
|-----|-------------|---------|
-| `response_buffer_size` | Response buffer size in bytes | `10000000` |
-| `max_concurrent_bidi_streams` | Maximum concurrent bidirectional streams | `10000` |
+| `response_buffer_size` | Maximum bytes read for one response | `10000000` |
+| `max_concurrent_bidi_streams` | Maximum concurrent peer-initiated bidirectional streams | `10000` |
| `datagram_send_buffer_size` | Datagram send buffer size in bytes | `100000` |
-| `initial_mtu` | Initial MTU | `1200` |
-| `send_window` | Send window size | `100000` |
-| `receive_window` | Receive window size | `100000` |
-| `keep_alive_interval` | Keep-alive interval in milliseconds (number) | `5000` |
-| `max_idle_timeout` | Maximum idle timeout in milliseconds (number) | `10000` |
+| `initial_mtu` | Initial UDP payload size in bytes, clamped to at least `1200` | `1200` |
+| `send_window` | Send window size in bytes | `100000` |
+| `receive_window` | Connection receive window size in bytes | `100000` |
+| `keep_alive_interval` | Keep-alive interval in milliseconds; `0` disables it | `5000` |
+| `max_idle_timeout` | Idle timeout in milliseconds; `0` leaves Quinn's `30000` ms default | `10000` |
| `validate_certificate` | Validate the server certificate (`true`/`false`) | `false` |
| `heartbeat_interval` | Duration between heartbeats | `5s` |
| `reconnection_max_retries` | Number of reconnection attempts, or `unlimited` | `unlimited` |
| `reconnection_interval` | Duration between reconnection attempts | `1s` |
-| `reconnection_reestablish_after` | Duration to wait before reestablishing the connection | `5s` |
+| `reconnection_reestablish_after` | Cooldown measured from the last connection establishment | `5s` |
```
-iggy+quic://iggy:iggy@localhost:8080?validate_certificate=false&reconnection_max_retries=5
+iggy+quic://iggy:iggy@127.0.0.1:8080?validate_certificate=false&reconnection_max_retries=5
```
+QUIC certificate validation is disabled by default. When enabled, it uses the platform verifier and the connection-string configuration's server name, `localhost`. For a different certificate name, use the Rust QUIC configuration builder's `with_server_name()` method. Values above QUIC's 62-bit limit for `max_concurrent_bidi_streams`, `receive_window`, or `max_idle_timeout` are rejected with `InvalidConfiguration`.
+
## WebSocket options
| Key | Description | Default |
@@ -105,54 +115,62 @@ iggy+quic://iggy:iggy@localhost:8080?validate_certificate=false&reconnection_max
| `heartbeat_interval` | Duration between heartbeats | `5s` |
| `reconnection_retries` | Number of reconnection attempts, or `unlimited` | `unlimited` |
| `reconnection_interval` | Duration between reconnection attempts | `1s` |
-| `reestablish_after` | Duration to wait before reestablishing the connection | `5s` |
-| `read_buffer_size` | Read buffer size in bytes | transport default |
-| `write_buffer_size` | Write buffer size in bytes | transport default |
-| `max_write_buffer_size` | Maximum write buffer size in bytes | transport default |
-| `max_message_size` | Maximum WebSocket message size in bytes | transport default |
-| `max_frame_size` | Maximum WebSocket frame size in bytes | transport default |
-| `accept_unmasked_frames` | Accept unmasked frames (`true`/`false`) | transport default |
+| `reestablish_after` | Cooldown measured from the last connection establishment | `5s` |
+| `read_buffer_size` | Read buffer size in bytes | `131072` |
+| `write_buffer_size` | Target write buffer size in bytes | `131072` |
+| `max_write_buffer_size` | Maximum write buffer size in bytes; must exceed `write_buffer_size` | `usize::MAX` |
+| `max_message_size` | Maximum incoming WebSocket message size in bytes | `67108864` |
+| `max_frame_size` | Maximum incoming frame payload size in bytes | `16777216` |
+| `accept_unmasked_frames` | Has no effect on client connections | `false` |
| `tls` | Enable TLS (`true`/`false`) | `false` |
-| `tls_domain` | Domain name for TLS validation | empty |
-| `tls_ca_file` | Path to a CA certificate file | none |
+| `tls_domain` | Name for TLS validation; when empty, use the connected IP | empty |
+| `tls_ca_file` | PEM file of trusted CA certificates when validation is enabled | none |
| `tls_validate_certificate` | Validate the server certificate (`true`/`false`) | `false` |
```
-iggy+ws://iggy:iggy@localhost:8092?heartbeat_interval=5s&max_message_size=1048576
+iggy+ws://iggy:iggy@127.0.0.1:8092?heartbeat_interval=5s&max_message_size=1048576
```
+WebSocket TLS does not validate the server certificate unless `tls_validate_certificate=true`. With validation enabled, `tls_ca_file` replaces the built-in Mozilla roots. Size limits cannot be removed through connection strings; use the WebSocket configuration object for an unlimited message or frame size.
+
## HTTP options
| Key | Description | Default |
|-----|-------------|---------|
| `heartbeat_interval` | Duration between heartbeats | `5s` |
-| `retries` | Number of request retries | `3` |
+| `retries` | Maximum retries for transient request failures | `3` |
```
-iggy+http://iggy:iggy@localhost:3000?retries=5
+iggy+http://iggy:iggy@127.0.0.1:3000?retries=5
```
+The HTTP scheme constructs a plain `http://` API URL. Use the Rust HTTP builder's `with_api_url()` for HTTPS. HTTP parses the credential field but does not log in with it. In Rust, call `login_user()` or `login_with_personal_access_token()` explicitly before protected requests.
+
## Usage
+In Rust, construct clients inside a Tokio runtime, which QUIC needs even before `connect()`. The following snippet assumes an async function returning `Result<_, IggyError>` and the matching [Rust SDK setup](/docs/sdk/rust/intro).
+
```rust
use iggy::prelude::*;
// TCP with default options
-let client = IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+let client = IggyClient::from_connection_string("iggy://iggy:iggy@127.0.0.1:8090")?;
// QUIC
-let client = IggyClient::from_connection_string("iggy+quic://iggy:iggy@localhost:8080")?;
+let client = IggyClient::from_connection_string("iggy+quic://iggy:iggy@127.0.0.1:8080")?;
// WebSocket
-let client = IggyClient::from_connection_string("iggy+ws://iggy:iggy@localhost:8092")?;
+let client = IggyClient::from_connection_string("iggy+ws://iggy:iggy@127.0.0.1:8092")?;
// HTTP
-let client = IggyClient::from_connection_string("iggy+http://iggy:iggy@localhost:3000")?;
+let client = IggyClient::from_connection_string("iggy+http://iggy:iggy@127.0.0.1:3000")?;
+client.login_user("iggy", "iggy").await?;
```
```python
-# Python
-client = IggyClient.from_connection_string("iggy://iggy:iggy@localhost:8090")
+from apache_iggy import IggyClient
+
+client = IggyClient.from_connection_string("iggy://iggy:iggy@127.0.0.1:8090")
```
-When you need settings a connection string cannot express (a custom `Encryptor`, a `Partitioner`, and similar), start from `IggyClientBuilder::from_connection_string()` and extend the builder on top.
+When you need settings a connection string cannot express (client-side encryption, a custom `Partitioner`, and similar), start from `IggyClientBuilder::from_connection_string()` and extend the builder on top.
diff --git a/content/docs/sdk/cpp/intro.mdx b/content/docs/sdk/cpp/intro.mdx
index cc883334..78b60957 100644
--- a/content/docs/sdk/cpp/intro.mdx
+++ b/content/docs/sdk/cpp/intro.mdx
@@ -9,7 +9,7 @@ This SDK isn't published to any package registry. Consume it from the monorepo a
## Building from source
-The project builds with Bazel (`BUILD.bazel` / `MODULE.bazel`). The build drives the system-provided cargo toolchain to compile the Rust bridge, so concurrent runs can race. Run builds serially.
+The project builds with the Bazel version pinned in `.bazelversion` (`BUILD.bazel` / `MODULE.bazel`). Its Rust bridge uses the Cargo and Rust toolchain provisioned by `rules_rust`, with a separate target directory in the Bazel output tree. Use a compatible Java runtime for Bazel; the pinned Bazel 9.2.0 requires Java 21 or newer.
```bash
cd foreign/cpp
@@ -26,7 +26,7 @@ bazel test //:e2e
## Quick start
-The snippet below mirrors the SDK's own end-to-end tests in `foreign/cpp/tests/e2e`:
+The snippet below mirrors the SDK's own end-to-end tests in `foreign/cpp/tests/e2e`. Start a [local server](/docs/introduction/getting-started) on `127.0.0.1:8090` with the `iggy`/`iggy` credentials first. The stream name must not already exist.
```cpp
#include "lib.rs.h"
@@ -34,6 +34,7 @@ The snippet below mirrors the SDK's own end-to-end tests in `foreign/cpp/tests/e
#include <cstdint>
#include <iostream>
#include <string>
+#include <utility>
iggy::ffi::Identifier id(const std::string &name) {
iggy::ffi::Identifier identifier;
@@ -56,9 +57,7 @@ rust::Vec<std::uint8_t> payload(const std::string &text) {
}
int main() {
- // Empty string = TCP to the default local address. Connection strings
- // (iggy://..., iggy+quic://..., ...) are also accepted.
- iggy::ffi::Client *client = iggy::ffi::new_connection("");
+ iggy::ffi::Client *client = iggy::ffi::new_connection({});
client->connect();
client->login_user("iggy", "iggy");
@@ -105,7 +104,7 @@ iggy+http://user:pass@host:port (HTTP)
iggy+ws://user:pass@host:port (WebSocket)
```
-When an empty string or plain address is passed to `new_connection()`, it defaults to TCP.
+`iggy::ffi::from_connection_string()` accepts these strings. `new_connection()` instead takes an `IggyClientConfig`: `{}` selects TCP at `127.0.0.1:8090` without automatic login, and its `server_address` field overrides that address. Binary clients authenticate from their connection string during `connect()`; HTTP requires an explicit `login_user()` call. See [connection strings](/docs/sdk/connection-strings) for the supported options and credential syntax.
## Helper types
@@ -113,6 +112,6 @@ Beyond the generated `lib.rs.h` FFI surface, `foreign/cpp/include/iggy.hpp` ship
## Current status
-The FFI bridge exposes the full command-level client: streams, topics, partitions, and segments; message sending and polling with user headers; consumer groups and consumer offsets; stats, client info, and cluster metadata; snapshots; permission updates and password changes; topic option discovery (`describe_options`); and raw binary requests (`send_binary_request`).
+The FFI bridge exposes command-level operations for streams, topics, partitions, and segments; message sending and polling with user headers; consumer groups and consumer offsets; stats, client info, and cluster metadata; snapshots; user create/get/update/delete, permission updates and password changes; topic option discovery (`describe_options`); and raw binary requests (`send_binary_request`).
-Not yet exposed: the high-level `Producer`/`Consumer` wrappers and user management (create/get/update/delete users, personal access tokens). Contributions are welcome.
+Not yet exposed: the high-level `Producer`/`Consumer` wrappers and personal access token management. Contributions are welcome.
diff --git a/content/docs/sdk/csharp/examples.mdx b/content/docs/sdk/csharp/examples.mdx
index 43302389..34a1cf6f 100644
--- a/content/docs/sdk/csharp/examples.mdx
+++ b/content/docs/sdk/csharp/examples.mdx
@@ -5,6 +5,8 @@ description: "Producer and consumer samples for the C# SDK, built on the high-le
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).
+Use the [installation and server setup](/docs/sdk/csharp/intro) first. The snippets share their named resources: run a producer before its corresponding consumer.
+
## Producer
A publisher that creates the stream and topic if missing, batches sends in the background, and retries failures:
@@ -18,7 +20,7 @@ using Apache.Iggy.Extensions;
using Apache.Iggy.Factory;
using Apache.Iggy.Messages;
-var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
+using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp
@@ -27,7 +29,7 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
await client.ConnectAsync();
await client.LoginUserAsync("iggy", "iggy");
-var publisher = client.CreatePublisherBuilder(
+await using var publisher = client.CreatePublisherBuilder(
Identifier.String("dev"),
Identifier.String("events"))
.CreateStreamIfNotExists("dev")
@@ -46,14 +48,13 @@ for (var i = 0; i < 100; i++)
// Drain the background queue before exiting
await publisher.WaitUntilAllSendsAsync();
-await publisher.DisposeAsync();
-Console.WriteLine("Sent 100 messages");
+Console.WriteLine("Queued 100 messages");
```
## Consumer group
-A consumer that creates and joins a consumer group, commits offsets after each received message, and surfaces polling errors:
+A consumer that creates and joins a consumer group, stores each offset after handling the yielded message and requesting the next one, and surfaces polling errors:
```csharp
using System.Text;
@@ -65,7 +66,7 @@ using Apache.Iggy.Extensions;
using Apache.Iggy.Factory;
using Apache.Iggy.Kinds;
-var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
+using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp
@@ -74,7 +75,7 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
await client.ConnectAsync();
await client.LoginUserAsync("iggy", "iggy");
-var consumer = client.CreateConsumerBuilder(
+await using var consumer = client.CreateConsumerBuilder(
Identifier.String("dev"),
Identifier.String("events"),
Consumer.Group("event-processors"))
@@ -115,7 +116,7 @@ using Apache.Iggy.Factory;
using Apache.Iggy.Kinds;
using Apache.Iggy.Publishers;
-var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
+using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp
@@ -134,11 +135,10 @@ var publisherBuilder = IggyPublisherBuilder<OrderEvent>.Create(
publisherBuilder.CreateStreamIfNotExists("orders");
publisherBuilder.CreateTopicIfNotExists("created");
-var publisher = publisherBuilder.Build();
+await using 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(
@@ -151,7 +151,7 @@ var consumerBuilder = IggyConsumerBuilder<OrderEvent>.Create(
consumerBuilder.WithPollingStrategy(PollingStrategy.Next());
consumerBuilder.WithAutoCommitMode(AutoCommitMode.AfterReceive);
-var consumer = consumerBuilder.Build();
+await using var consumer = consumerBuilder.Build();
await consumer.InitAsync();
await foreach (var message in consumer.ReceiveDeserializedAsync())
@@ -186,3 +186,28 @@ The [examples/csharp](https://github.com/apache/iggy/tree/master/examples/csharp
- **MessageEnvelope** - envelope pattern (message type + JSON payload) over the low-level client
- **MessageHeaders** - user-defined message headers
- **TcpTls** - TLS-encrypted TCP connection
+
+Repository examples require .NET 10 and reference the local SDK. Build and run from the same checkout as the server:
+
+```bash
+dotnet build examples/csharp/Iggy_SDK.Examples.sln --configuration Release
+dotnet run --no-build --configuration Release --project examples/csharp/src/GettingStarted/Iggy_SDK.Examples.GettingStarted.Producer
+dotnet run --no-build --configuration Release --project examples/csharp/src/GettingStarted/Iggy_SDK.Examples.GettingStarted.Consumer
+```
+
+The getting-started producer sends 50 messages to partition 0; its consumer reads five nonempty batches and then exits. Envelope and header examples also use five batches. NewSdk uses four partitions and its group consumer runs until stopped.
+
+For TcpTls, run from the repository root so `core/certs/iggy_ca_cert.pem` resolves. Start a separate matching server with the example TLS certificate and root credentials:
+
+```bash
+IGGY_ROOT_USERNAME=iggy IGGY_ROOT_PASSWORD=iggy \
+IGGY_TCP_TLS_ENABLED=true \
+IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
+IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
+cargo run --bin iggy-server
+```
+
+```bash
+dotnet run --no-build --configuration Release --project examples/csharp/src/TcpTls/Iggy_SDK.Examples.TcpTls.Producer
+dotnet run --no-build --configuration Release --project examples/csharp/src/TcpTls/Iggy_SDK.Examples.TcpTls.Consumer
+```
diff --git a/content/docs/sdk/csharp/guide.mdx b/content/docs/sdk/csharp/guide.mdx
index ac5e5d71..550177b7 100644
--- a/content/docs/sdk/csharp/guide.mdx
+++ b/content/docs/sdk/csharp/guide.mdx
@@ -5,33 +5,38 @@ description: "Client configuration and the full low-level IIggyClient API surfac
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)).
+The API fragments below reuse a connected, authenticated `client` (see [Creating a client](/docs/sdk/csharp/intro#creating-a-client)) and the imports in the configuration example. Use existing streams and topics for read, update, delete, publish, and poll operations; the create examples show how to prepare them.
## Client configuration
`IggyClientConfigurator` exposes the full set of connection options - buffer sizes, TLS, heartbeat, automatic reconnection with exponential backoff, and auto-login:
```csharp
-var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
+using System.Text;
+using Apache.Iggy;
+using Apache.Iggy.Configuration;
+using Apache.Iggy.Contracts;
+using Apache.Iggy.Contracts.Auth;
+using Apache.Iggy.Enums;
+using Apache.Iggy.Factory;
+using Apache.Iggy.Headers;
+using Apache.Iggy.IggyClient;
+using Apache.Iggy.Kinds;
+using Apache.Iggy.Messages;
+using Partitioning = Apache.Iggy.Kinds.Partitioning;
+
+using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp,
- // Buffer sizes (optional, default: 4096)
- ReceiveBufferSize = 4096,
- SendBufferSize = 4096,
+ // Null preserves the operating system socket defaults.
+ ReceiveBufferSize = null,
+ SendBufferSize = null,
// Upper bound on a reply frame the server sends, 64 MiB by default (TCP)
MaxResponseFrameSize = 64 * 1024 * 1024,
- // TLS/SSL configuration
- TlsSettings = new TlsSettings
- {
- Enabled = true,
- Hostname = "iggy",
- CertificatePath = "/path/to/cert"
- },
-
// Idle ping keeping the session alive and consumer-group assignments fresh (TCP).
// Default 5 seconds; init-only
HeartbeatInterval = TimeSpan.FromSeconds(5),
@@ -48,8 +53,7 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
BackoffMultiplier = 2.0
},
- // Auto-login after connection. Reconnection needs it: without credentials
- // to replay, a reconnect cannot restore the session
+ // Configured credentials are reused when TCP reconnects.
AutoLoginSettings = AutoLoginSettings.For("iggy", "iggy")
// or AutoLoginSettings.ForPersonalAccessToken("your_token")
});
@@ -57,7 +61,7 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
await client.ConnectAsync();
```
-With auto-login configured, the client logs in automatically once the connection is established, so you can skip the explicit `LoginUserAsync` call.
+With TCP auto-login configured, the client logs in once connected. Successful manual username/password or PAT login is also remembered for reconnects. HTTP `ConnectAsync` does no network work and does not auto-login; call `LoginUserAsync` or `LoginWithPersonalAccessTokenAsync` explicitly.
#### `IggyClientConfigurator`
@@ -66,12 +70,12 @@ With auto-login configured, the client logs in automatically once the connection
| `BaseAddress` | `string` | *required* | Server address. TCP takes `host:port` (e.g. `127.0.0.1:8090`); HTTP needs a full URI (e.g. `http://127.0.0.1:3000`) |
| `Protocol` | `Protocol` | *required* | Transport: `Protocol.Tcp` or `Protocol.Http` |
| `MaxResponseFrameSize` | `int` | 64 MiB | Largest reply frame accepted over TCP. A reply above the bound is refused and the connection dropped; raise it if a single response legitimately exceeds it (a large [snapshot](#snapshots) is the usual case) |
-| `ReceiveBufferSize` | `int` | `4096` | Receive buffer size in bytes |
-| `SendBufferSize` | `int` | `4096` | Send buffer size in bytes |
+| `ReceiveBufferSize` | `int?` | `null` | TCP receive buffer override in bytes; null preserves the OS default |
+| `SendBufferSize` | `int?` | `null` | TCP send buffer override in bytes; null preserves the OS default |
| `HeartbeatInterval` | `TimeSpan` | `5s` | Interval between the automatic pings the TCP client sends while connected (see [Heartbeat](#heartbeat)). Init-only; must be between 1 millisecond and about 49 days |
-| `TlsSettings` | `TlsSettings` | disabled | TLS/SSL configuration (see below) |
-| `ReconnectionSettings` | `ReconnectionSettings` | enabled | Automatic reconnection behavior (see below) |
-| `AutoLoginSettings` | `AutoLoginSettings` | disabled | Automatic login on connect |
+| `TlsSettings` | `TlsSettings` | disabled | TCP TLS configuration (see below); HTTP uses its HTTPS URI and standard certificate validation |
+| `ReconnectionSettings` | `ReconnectionSettings` | enabled | TCP reconnection behavior (see below) |
+| `AutoLoginSettings` | `AutoLoginSettings` | disabled | TCP 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 |
@@ -93,14 +97,14 @@ The TCP client pings the server on its own every `HeartbeatInterval` while conne
| Property | Type | Default | Description |
|----------|------|---------|-------------|
| `Enabled` | `bool` | `true` | Enable automatic reconnection when the connection drops |
-| `MaxRetries` | `int` | `0` | Maximum reconnection attempts (`0` = infinite) |
+| `MaxRetries` | `int` | `0` | Maximum failed passes through the known address roster (`0` = unlimited) |
| `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 |
-> **Warning:** Reconnection is **on by default** with **unlimited retries**. Only a failed dial is retried. A rejected certificate, bad credentials, or a missing leader is thrown right away. With the default `MaxRetries = 0` an unreachable server is retried forever, so a request that passes no `CancellationToken` waits for as long as the server stays down. Set `MaxRetries` or pass a token to bound the wait, and set `Enabled = false` to opt out of reconnection entirely. Reconnection only [...]
+> **Warning:** TCP reconnection is **on by default** with **unlimited retries**. A failed pass tries the known roster addresses before consuming one retry. Connection and TLS handshake failures can be retried; an invalid local CA file stops after the pass, and rejected login credentials are surfaced. Pass a `CancellationToken` or set `MaxRetries` to bound connection attempts, or set `Enabled = false` to disable reconnection. Request replay also has its own deadline and requires configure [...]
#### `AutoLoginSettings`
@@ -111,18 +115,20 @@ The TCP client pings the server on its own every `HeartbeatInterval` while conne
| `Password` | `string` | `""` | Password for auto-login |
| `PersonalAccessToken` | `string` | `""` | Personal access token to sign in with instead of a username and password; takes precedence over them when set |
-All properties are init-only: build the settings with an object initializer or the static factories `AutoLoginSettings.For(username, password)` and `AutoLoginSettings.ForPersonalAccessToken(token)`. Configuring auto-login is what enables transparent request replay after a reconnect. Without stored credentials a reconnect **cannot restore the session**, and requests fail until the client logs in again.
+All properties are init-only: build the settings with an object initializer or the static factories `AutoLoginSettings.For(username, password)` and `AutoLoginSettings.ForPersonalAccessToken(token)`. TCP reconnects use configured auto-login credentials first, then credentials remembered from a successful manual login. Without either, a reconnect cannot restore the authenticated session.
## 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:
+Set `IggyClientConfigurator.MessageEncryptor` to encrypt message payloads and user headers client-side. They are encrypted on send and decrypted on poll; routing information and message metadata remain visible to the server. 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 System.Security.Cryptography;
using Apache.Iggy.Encryption;
-using var encryptor = new AesMessageEncryptor(key); // 16, 24, or 32 bytes
+var key = RandomNumberGenerator.GetBytes(32);
+using var encryptor = new AesMessageEncryptor(key);
-var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
+using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp,
@@ -130,18 +136,18 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
});
```
-For a custom scheme, implement `IMessageEncryptor`.
+Keep the same key for clients that exchange encrypted messages and for later reads; the generated key above lasts only for this example. 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.
+- 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`. `IggyConsumer` separately rejects wire auto-commit with encryption; use `AfterReceive` or `Disabled` there.
- 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):
+TCP clients publish connection state changes, useful for reacting to reconnects. HTTP subscription methods are no-ops:
```csharp
Func<ConnectionStateChangedEventArgs, Task> handler = async args =>
@@ -171,7 +177,7 @@ Server-assigned ids are **0-based**: the first stream, topic, partition, or cons
### User login
-Begin with the root account (`iggy` / `iggy`):
+Use the credentials configured when the server was initialized. These examples use the `iggy` / `iggy` setup from the introduction:
```csharp
var response = await client.LoginUserAsync("iggy", "iggy");
@@ -225,7 +231,7 @@ var permissions = new Permissions
ReadUsers = false,
SendMessages = false
},
- Streams = new Dictionary<int, StreamPermissions>
+ Streams = new Dictionary<uint, StreamPermissions>
{
[0] = new StreamPermissions
{
@@ -235,7 +241,7 @@ var permissions = new Permissions
ReadTopics = true,
PollMessages = true,
SendMessages = true,
- Topics = new Dictionary<int, TopicPermissions>
+ Topics = new Dictionary<uint, TopicPermissions>
{
[0] = new TopicPermissions
{
@@ -261,6 +267,7 @@ var users = await client.GetUsersAsync();
// Update name and/or status (both optional)
await client.UpdateUserAsync(userId, userName: "renamed_user", status: UserStatus.Inactive);
+userId = Identifier.String("renamed_user");
// Replace permissions
await client.UpdatePermissionsAsync(userId, permissions);
@@ -317,7 +324,7 @@ await client.CreateTopicAsync(
partitionsCount: 3,
compressionAlgorithm: CompressionAlgorithm.None,
messageExpiry: TimeSpan.Zero, // null or TimeSpan.Zero = server default; TimeSpan.MaxValue = never expire
- maxTopicSize: 0 // 0 = unlimited
+ maxTopicSize: 1024 * 1024 * 1024 // 1 GiB
);
```
@@ -335,17 +342,18 @@ await client.CreateTopicAsync(
options: new TopicOptions
{
SegmentSize = 128 * 1024 * 1024, // segment size in bytes (multiple of 512)
- EnforceFsync = true, // fsync every write to this topic's partitions
- MessagesRequiredToSave = 1000, // flush the journal after this many messages
+ Durability = Durability.Persisted,
+ ConsumerOffsetDurability = Durability.Persisted,
+ MessagesRequiredToSave = 1000, // attempt a flush after this many messages
SizeOfMessagesRequiredToSave = 32 * 1024 * 1024, // or after this many bytes, whichever trips first
- PreallocateSegments = true // reserve SegmentSize bytes up front
+ PreallocateSegments = true // request filesystem preallocation
}.ToDictionary()
);
```
-A property left `null` emits no key and keeps the server default. These keys are **settable at creation only**: `UpdateTopicAsync` refuses them by name, the same way it refuses a key outside the server's catalog, so a mistyped key fails the call instead of being silently ignored. Use [`DescribeOptionsAsync`](#option-catalog) to enumerate the keys a server accepts.
+Nullable properties left `null` emit no key and keep the server default. `Durability` and `ConsumerOffsetDurability` are non-nullable, default independently to `Replicated`, and are always emitted by `ToDictionary()`. These keys are **settable at creation only**: `UpdateTopicAsync` refuses them by name, the same way it refuses a key outside the server's catalog, so a mistyped key fails the call instead of being silently ignored. Use [`DescribeOptionsAsync`](#option-catalog) to enumerate [...]
-There is **no on-demand flush command**. Durability is configured per topic through `EnforceFsync` and the flush thresholds above.
+There is **no on-demand flush command**. `Durability` controls message completion and `ConsumerOffsetDurability` controls explicit offset-store completion. `Persisted` waits for persistence on the required quorum; flush thresholds alone do not provide that guarantee. See [Durability](/docs/server/durability) and [Topic options](/docs/server/topic-options).
```csharp
var topicId = Identifier.String("my-topic");
@@ -355,13 +363,16 @@ var topic = await client.GetTopicByIdAsync(streamId, topicId);
var topics = await client.GetTopicsAsync(streamId);
// Update (name required; compression, expiry, size optional)
-await client.UpdateTopicAsync(streamId, topicId, "renamed-topic");
+await client.UpdateTopicAsync(streamId, topicId, "renamed-topic",
+ messageExpiry: TimeSpan.FromDays(7), maxTopicSize: 1024 * 1024 * 1024);
// Purge (delete all messages, keep the topic) / delete
await client.PurgeTopicAsync(streamId, Identifier.String("renamed-topic"));
await client.DeleteTopicAsync(streamId, Identifier.String("renamed-topic"));
```
+TCP omits zero expiry/size defaults from a topic update; HTTP sends them. Use explicit values when updating retention and review the [default-sentinel caveat](/docs/server/topic-options).
+
## Partitions
Add partitions to or remove them from an existing topic:
@@ -402,7 +413,7 @@ foreach (var confirmation in response.Confirmations)
}
```
-`SendMessagesAsync` returns a `SendMessagesResponse` whose `Confirmations` carry `StreamId`, `TopicId`, `PartitionId`, and `BaseOffset` (the offset of the first message of the batch) per committed batch. An empty list means the batch committed with no offsets to report. A single-message overload is also available: `SendMessagesAsync(streamId, topicId, partitioning, message)`.
+`SendMessagesAsync` returns a `SendMessagesResponse` whose `Confirmations` carry `StreamId`, `TopicId`, `PartitionId`, and `BaseOffset` (the offset of the first message of the batch) per committed batch. An empty list reports no assigned offsets. 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).
@@ -412,20 +423,20 @@ Control which partition receives each message:
```csharp
// Balanced - the client round-robins across partitions (default)
-Partitioning.None()
+Partitioning.None();
// Send to a specific partition (partition ids are 0-based)
-Partitioning.PartitionId(0)
+Partitioning.PartitionId(0);
// Key-based routing - messages with the same key land on the same partition
-Partitioning.EntityIdString("user-123")
-Partitioning.EntityIdInt(12345)
-Partitioning.EntityIdUlong(12345)
-Partitioning.EntityIdGuid(Guid.NewGuid())
-Partitioning.EntityIdBytes(new byte[] { 1, 2, 3 })
+Partitioning.EntityIdString("user-123");
+Partitioning.EntityIdInt(12345);
+Partitioning.EntityIdUlong(12345);
+Partitioning.EntityIdGuid(Guid.NewGuid());
+Partitioning.EntityIdBytes(new byte[] { 1, 2, 3 });
```
-Partition selection is resolved **client-side** on both transports. Balanced partitioning round-robins over the topic's partitions, and key-based routing hashes the key with xxHash32 modulo the partition count, consistent with the Rust SDK.
+Partition selection is resolved **client-side** on both transports. Balanced partitioning round-robins over the topic's partitions, and key-based routing hashes the key with xxHash32 modulo the partition count, consistent with the Rust SDK for the same encoded key bytes and partition count. Strings use UTF-8, integers use little-endian bytes, and GUID keys use `Guid.ToByteArray()`.
### User-defined headers
@@ -456,7 +467,7 @@ Available value factories: `FromString`, `FromBool`, `FromBytes`, `FromUInt8`, `
### Fetching messages
-Poll a batch of messages. The `partitionId` may be `null` to consume from any partition:
+Poll a batch of messages. A null `partitionId` selects partition 0 for an ordinary consumer. For a group consumer, null lets the TCP client rotate through its assigned partitions:
```csharp
var polledMessages = await client.PollMessagesAsync(
@@ -490,9 +501,11 @@ var polledMessages = await client.PollMessagesAsync(new MessageFetchRequest
});
```
+With `autoCommit: true`, the server advances/submits the offset before application processing; the poll reply does not wait for durable completion of that offset store. Use `autoCommit: false` and explicitly store offsets after processing when that distinction matters.
+
### 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.
+On TCP, `PollMessagesAsync` copies each payload into its own `byte[]`. `PollMessagesRentedAsync` instead returns a `PolledMessagesRental` whose payloads and raw headers share a pooled buffer. Message wrappers and lazily parsed headers still allocate. HTTP first materializes its JSON response and then adapts it to the rental API.
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:
@@ -523,7 +536,7 @@ A `MessageFetchRequest` overload (`PollMessagesRentedAsync(request)`) is also av
| Member | Type | Description |
|--------|------|-------------|
-| `PartitionId` | `int` | Partition the messages came from |
+| `PartitionId` | `uint` | Partition the messages came from |
| `CurrentOffset` | `ulong` | Current offset for the partition |
| `Messages` | `IReadOnlyList<RentedMessageResponse>` | The rented messages |
@@ -545,11 +558,11 @@ The high-level `IggyConsumer` exposes the same pooled path as an async stream vi
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
+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
@@ -641,7 +654,8 @@ foreach (var spec in specs)
`SendBinaryRequestAsync` is an escape hatch that sends a raw command code with a prebuilt payload and returns the raw response bytes, for commands the typed API does not cover:
```csharp
-var responseBytes = await client.SendBinaryRequestAsync(commandCode, payloadBytes);
+const uint PingCommand = 1;
+var responseBytes = await client.SendBinaryRequestAsync(PingCommand, Array.Empty<byte>());
```
Session-control codes (login, logout, and the like) are rejected with an invalid-command error, and the operation is TCP-only: HTTP clients throw `FeatureUnavailableException`.
@@ -675,7 +689,7 @@ A full snapshot can exceed the default 64 MiB reply-frame bound, in which case t
### Segment management
-Delete the last N segments from a partition:
+Delete up to N of the oldest sealed segments from a partition. The active segment is retained:
> **Note:** TCP-only - throws `FeatureUnavailableException` on HTTP.
diff --git a/content/docs/sdk/csharp/high-level-sdk.mdx b/content/docs/sdk/csharp/high-level-sdk.mdx
index ad8fd875..94328556 100644
--- a/content/docs/sdk/csharp/high-level-sdk.mdx
+++ b/content/docs/sdk/csharp/high-level-sdk.mdx
@@ -10,21 +10,25 @@ The per-call [`IIggyClient`](/docs/sdk/csharp/guide) API is explicit but verbose
- **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
+- Sending and receiving over **rented, pooled buffers** to avoid separate payload copies on TCP
-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`.
+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:
```csharp
+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;
-var publisher = IggyPublisherBuilder.Create(
+await using var publisher = IggyPublisherBuilder.Create(
client,
Identifier.String("my-stream"),
Identifier.String("my-topic")
@@ -46,19 +50,18 @@ 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). The builder-created client auto-logs in with these credentials and replays them across reconnects |
-| `WithConnection(protocol, address, personalAccessToken, receiveBufferSize = 4096, sendBufferSize = 4096, reconnectionSettings = null)` | Same, authenticating with a personal access token instead of a username and password |
+| `WithConnection(protocol, address, login, password, receiveBufferSize = null, sendBufferSize = null, reconnectionSettings = null)` | Connection settings - only used when the builder creates its own client (i.e. the `Create(streamId, topicId)` overload, without an existing client). A builder-created TCP client auto-logs in and reuses the credentials across reconnects. For HTTP, supply an explicitly authenticated client |
+| `WithConnection(protocol, address, personalAccessToken, receiveBufferSize = null, sendBufferSize = null, reconnectionSettings = null)` | Same, authenticating with a personal access token instead of a username and password |
| `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, 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 |
+| `WithRetry(enabled = true, maxAttempts = 3, initialDelay = 100ms, maxDelay = 10s, backoffMultiplier = 2.0)` | Retry background sends with exponential backoff; direct sends surface errors |
| `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) |
@@ -70,7 +73,7 @@ await publisher.DisposeAsync();
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `enabled` | `bool` | `true` | Whether retry is enabled |
-| `maxAttempts` | `int` | `3` | Maximum retry attempts |
+| `maxAttempts` | `int` | `3` | Total send attempts, including the first send |
| `initialDelay` | `TimeSpan?` | `100ms` | Delay before the first retry |
| `maxDelay` | `TimeSpan?` | `10s` | Maximum delay between retries |
| `backoffMultiplier` | `double` | `2.0` | Exponential backoff multiplier |
@@ -81,15 +84,15 @@ await publisher.DisposeAsync();
|-----------|------|---------|-------------|
| `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 |
+| `batchSize` | `int` | `100` | Flush threshold in messages; an individual send call is not split |
| `flushInterval` | `TimeSpan?` | `100ms` | Interval at which pending messages are flushed |
-| `disposalTimeout` | `TimeSpan?` | `5s` | How long `DisposeAsync` waits for the background processor to drain |
+| `disposalTimeout` | `TimeSpan?` | `5s` | How long `DisposeAsync` waits for the processor to stop; pending sends may be discarded |
-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.
+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 throug [...]
### 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`:
+`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`:
```csharp
using System.Text.Json;
@@ -109,6 +112,7 @@ foreach (var evt in events)
var batch = builder.Build();
await publisher.SendAsync(batch);
+await publisher.WaitUntilAllSendsAsync();
```
`RentedMessageBatchBuilder`:
@@ -135,13 +139,7 @@ Ownership rules:
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(
+await using var publisher = IggyPublisherBuilder<Order>.Create(
client,
Identifier.String("orders-stream"),
Identifier.String("orders-topic"),
@@ -149,7 +147,15 @@ var publisher = IggyPublisherBuilder<Order>.Create(
).Build();
await publisher.InitAsync();
-await publisher.SendAsync(new List<Order> { /* ... */ });
+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.
@@ -157,7 +163,7 @@ Besides the collection overload, `SendAsync` also accepts a single item (`SendAs
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(
+await using var publisher = IggyPublisherBuilder<Order>.Create(
client,
Identifier.String("orders-stream"),
Identifier.String("orders-topic"),
@@ -165,6 +171,8 @@ var publisher = IggyPublisherBuilder<Order>.Create(
).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](#typed-consumer)).
## IggyConsumer
@@ -177,7 +185,7 @@ using Apache.Iggy;
using Apache.Iggy.Consumers;
using Apache.Iggy.Kinds;
-var consumer = IggyConsumerBuilder.Create(
+await using var consumer = IggyConsumerBuilder.Create(
client,
Identifier.String("my-stream"),
Identifier.String("my-topic"),
@@ -201,9 +209,9 @@ await foreach (var message in consumer.ReceiveAsync())
| Method | Description |
|--------|-------------|
-| `WithConnection(protocol, address, login, password, receiveBufferSize = 4096, sendBufferSize = 4096, reconnectionSettings = null)` | Connection settings - only used when the builder creates its own client. The builder-created client auto-logs in with these credentials and replays them across reconnects |
-| `WithConnection(protocol, address, personalAccessToken, receiveBufferSize = 4096, sendBufferSize = 4096, reconnectionSettings = null)` | Same, authenticating with a personal access token instead of a username and password |
-| `WithPartitionId(partitionId)` | Consume from a specific partition |
+| `WithConnection(protocol, address, login, password, receiveBufferSize = null, sendBufferSize = null, reconnectionSettings = null)` | Connection settings - only used when the builder creates its own client. A builder-created TCP client auto-logs in and reuses the credentials across reconnects. For HTTP, supply an explicitly authenticated client |
+| `WithConnection(protocol, address, personalAccessToken, receiveBufferSize = null, sendBufferSize = null, reconnectionSettings = null)` | Same, authenticating with a personal access token instead of a username and password |
+| `WithPartitionId(partitionId)` | Ordinary consumer partition (null selects 0); ignored for consumer groups |
| `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) |
@@ -213,12 +221,14 @@ await foreach (var message in consumer.ReceiveAsync())
| `SubscribeOnPollingError(handler)` | Observe polling errors |
| `WithLogger(loggerFactory)` | Logger factory for diagnostics |
+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`):
| Mode | Description |
|------|-------------|
-| `Auto` | Commit the offset while polling |
-| `AfterReceive` | Commit after each message is received |
+| `Auto` | Advance/submit the offset during the server poll, before application processing |
+| `AfterReceive` | Store after the yielded message has been handled and enumeration resumes; breaking or throwing can leave that message uncommitted |
| `Disabled` | Commit manually |
### Manual offset control
@@ -233,14 +243,14 @@ await foreach (var message in consumer.ReceiveAsync())
}
```
-`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.
+`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:
```csharp
-var consumer = IggyConsumerBuilder.Create(
+await using var consumer = IggyConsumerBuilder.Create(
client,
Identifier.String("my-stream"),
Identifier.String("my-topic"),
@@ -259,12 +269,11 @@ await foreach (var message in consumer.ReceiveAsync())
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:
+On TCP, `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())
@@ -279,19 +288,13 @@ await foreach (var message in consumer.ReceiveRentedAsync())
`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.
+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:
```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"),
@@ -299,9 +302,10 @@ var builder = IggyConsumerBuilder<OrderEvent>.Create(
Consumer.Group("order-processors"),
new OrderDeserializer()
);
+builder.WithPollingStrategy(PollingStrategy.Next());
builder.WithAutoCommitMode(AutoCommitMode.AfterReceive);
-var consumer = builder.Build();
+await using var consumer = builder.Build();
await consumer.InitAsync();
await foreach (var message in consumer.ReceiveDeserializedAsync())
@@ -311,6 +315,14 @@ await foreach (var message in consumer.ReceiveDeserializedAsync())
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.
diff --git a/content/docs/sdk/csharp/intro.mdx b/content/docs/sdk/csharp/intro.mdx
index 7e8ae6e3..916b05c4 100644
--- a/content/docs/sdk/csharp/intro.mdx
+++ b/content/docs/sdk/csharp/intro.mdx
@@ -10,10 +10,23 @@ The SDK is built around the `IIggyClient` interface, which aggregates every feat
## Installation
```bash
-dotnet add package Apache.Iggy --prerelease
+dotnet add package Apache.Iggy --version 0.9.0-edge.9
```
-The SDK targets .NET 8 and .NET 10 (`net8.0` and `net10.0`). Stable versions on NuGet stop at `0.8.0`, which predates the current wire protocol and cannot talk to a current server. The `0.9.0` line ships as **prerelease versions** (`0.9.0-edge.x` at the time of writing), so `--prerelease` is required.
+The SDK targets .NET 8 and .NET 10 (`net8.0` and `net10.0`). Stable `0.8.0` uses TCP framing that predates the server 0.9.0 protocol. These pages prepare for server 0.9.0 and use the `0.9.0-edge.9` prerelease, which includes both topic durability options. When building from source, use the server and SDK from the same checkout.
+
+To use the source SDK with these release-preparation examples, run from the Iggy checkout root:
+
+```bash
+dotnet new console --name IggySample --framework net10.0
+dotnet add IggySample/IggySample.csproj reference foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
+```
+
+Put a sample in `IggySample/Program.cs` and run `dotnet run --project IggySample`. For console logging, add its package to that project:
+
+```bash
+dotnet add IggySample/IggySample.csproj package Microsoft.Extensions.Logging.Console --version 10.0.11
+```
## Supported protocols
@@ -26,57 +39,62 @@ Some operations are **TCP-only** and throw `FeatureUnavailableException` on HTTP
## Connection semantics
-Over TCP every request is wrapped in a 256-byte consensus header (Viewstamped Replication), the client registers a consensus session at login, and writes are replicated before they are acknowledged. The `IIggyClient` surface is unchanged. What changes is the behavior around connections, groups, and failures:
+Over TCP every request is wrapped in a 256-byte consensus header (Viewstamped Replication), the client registers a consensus session at login, and replicated writes complete after the required quorum acknowledges them. Topic `Durability.Persisted` additionally waits for local persistence on that quorum; message and explicit offset durability are configured independently. See [Durability](/docs/server/durability). Both transports implement `IIggyClient`, with these TCP connection rules:
- **Login binds a session.** `LoginUserAsync` / `LoginWithPersonalAccessTokenAsync` run the register handshake, and the session lives for as long as the connection. Logging out, being evicted, or losing the connection ends it, and the next login registers a fresh one.
- **Leader redirection is automatic.** The client reads the cluster roster, follows the current leader, and re-checks it when a request is refused because the node stopped being primary. `GetCurrentAddress()` reports the node the client currently talks to.
-- **The client picks partitions.** Balanced and message-key partitioning are resolved client-side (the message-key hash matches the Rust SDK byte for byte), and consumer-group polls round-robin over the partitions the coordinator assigned to this client.
+- **The client picks partitions.** Balanced and message-key partitioning are resolved client-side (identical encoded key bytes and partition counts produce the same xxHash32 routing result as the Rust SDK), and consumer-group polls round-robin over the partitions the coordinator assigned to this client.
- **Consumer groups are assignment-based.** `JoinConsumerGroupAsync` makes this client a member. The assignment is synced on demand and refreshed on every `PingAsync`. The TCP client pings on its own every `HeartbeatInterval` (5 seconds by default), so an idle session survives the server's heartbeat verification and assignments stay fresh.
- **Credentials are bounds-checked locally.** A username outside 3-50 bytes, a password outside 3-100 bytes, or a personal access token outside 1-255 bytes is rejected before the request is framed.
- **Consumer offsets need an explicit partition.** `StoreOffsetAsync` / `DeleteOffsetAsync` do not accept a `null` partition id under VSR. Passing one throws client-side.
-- **Polling a topic that does not exist returns an empty poll** rather than throwing. The server answers an unresolved topic with the empty-poll reply shape, so the client cannot tell it apart from a topic with no messages. Check the topic exists first if the distinction matters.
+- **Polling a missing topic throws `IggyInvalidStatusCodeException`.** An existing topic with no available messages returns an empty poll.
### Failed requests
-The SDK replays a request whenever the server says it never admitted it. Two exceptions surface to the caller:
+The SDK can replay a request the server reports as never admitted, within its request deadline. Replay-safe operations, including read-only polls, can also be retried after a lost connection. Two exceptions surface to the caller:
- `IggyInvalidStatusCodeException` carries the server status code, with `FromServer` telling apart a verdict the cluster reported from a failure the client raised itself.
-- `VsrRequestOutcomeUnknownException` means no server verdict arrived after the request was written (the connection was lost, the call was cancelled, or the server evicted the session mid-flight), so the cluster may or may not have committed it. The SDK will not replay it on a new session, because that would bypass server-side deduplication. Re-issuing it is **the caller's decision**.
+- `VsrRequestOutcomeUnknownException` means no definitive server verdict arrived for an operation that cannot safely be replayed after it was written (the connection was lost, the call was cancelled, or the server evicted the session mid-flight), so the cluster may or may not have committed it. The SDK will not replay it on a new session, because re-issuing a possibly committed operation can repeat its effects. Re-issuing it is **the caller's decision**.
The high-level abstractions handle the unknown-outcome case differently:
-- `IggyConsumer` rethrows it rather than swallowing it, because an auto-committing poll may already have advanced the offset. Rethrowing ends the polling loop, so catch it around the enumeration, decide whether the operation is safe to re-issue, and start consuming again:
+- `IggyConsumer` rethrows it rather than swallowing it, because an auto-committing poll may already have advanced the offset. Rethrowing ends the polling loop, so handle it around the enumeration. This fragment assumes an initialized `consumer` and a `CancellationToken token`, and stops consumption for an application-level recovery decision:
```csharp
- while (!token.IsCancellationRequested)
+ using Apache.Iggy.Exceptions;
+
+ try
{
- try
- {
- await foreach (var message in consumer.ReceiveAsync(token))
- {
- Process(message);
- }
- }
- catch (VsrRequestOutcomeUnknownException)
+ await foreach (var message in consumer.ReceiveAsync(token))
{
- // The last poll may or may not have committed. Decide whether
- // re-polling is safe for this workload, then resume the loop.
+ Console.WriteLine($"Offset {message.CurrentOffset}");
}
}
+ catch (VsrRequestOutcomeUnknownException exception)
+ {
+ Console.Error.WriteLine($"Poll outcome is unknown: {exception.Message}");
+ throw;
+ }
```
- `IggyPublisher` with background sending does not retry it: the batch is reported through the message-batch-failed event (`SubscribeOnMessageBatchFailed`), typed so a subscriber can tell "not sent" from "possibly sent twice". Direct (non-background) sends throw it to the caller.
## Creating a client
-Create a client with `IggyClientFactory.CreateClient`, then call `ConnectAsync`:
+The examples use `iggy` / `iggy`. For a new server data directory, start the matching source server from the Iggy repository root:
+
+```bash
+IGGY_ROOT_USERNAME=iggy IGGY_ROOT_PASSWORD=iggy cargo run --bin iggy-server
+```
+
+For an existing server, use its configured credentials. 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
+using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp
@@ -86,19 +104,22 @@ await client.ConnectAsync();
await client.LoginUserAsync("iggy", "iggy");
```
-Optionally, provide an `ILoggerFactory` for diagnostics (defaults to `NullLoggerFactory.Instance`):
+Optionally, provide an `ILoggerFactory` for diagnostics (defaults to `NullLoggerFactory.Instance`). The console logger requires the `Microsoft.Extensions.Logging.Console` package:
```csharp
+using Apache.Iggy.Configuration;
+using Apache.Iggy.Enums;
+using Apache.Iggy.Factory;
using Microsoft.Extensions.Logging;
-var loggerFactory = LoggerFactory.Create(builder =>
+using var loggerFactory = LoggerFactory.Create(builder =>
{
builder
.AddFilter("Apache.Iggy", LogLevel.Information)
.AddConsole();
});
-var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
+using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp,
@@ -106,7 +127,7 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
});
```
-`IggyClientConfigurator` also exposes buffer sizes, TLS, the heartbeat interval, the maximum response frame size, automatic reconnection with exponential backoff (on by default), 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.
+`IggyClientConfigurator` also exposes TCP socket buffers, TLS, heartbeat, response-frame limits, reconnection with exponential backoff (on by default), auto-login, and message encryption. HTTP clients require explicit login; the TCP connection settings are not applied to HTTP. See [Client configuration](/docs/sdk/csharp/guide#client-configuration) for the full reference.
## Quick start
@@ -123,7 +144,7 @@ using Apache.Iggy.Extensions;
using Apache.Iggy.Factory;
using Apache.Iggy.Messages;
-var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
+using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp
@@ -132,7 +153,7 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
await client.ConnectAsync();
await client.LoginUserAsync("iggy", "iggy");
-var publisher = client.CreatePublisherBuilder(
+await using var publisher = client.CreatePublisherBuilder(
Identifier.String("sample-stream"),
Identifier.String("sample-topic"))
.CreateStreamIfNotExists("sample-stream")
@@ -147,7 +168,6 @@ for (var i = 0; i < 10; i++)
await publisher.SendMessagesAsync(new List<Message> { new(Guid.NewGuid(), payload) });
}
-await publisher.DisposeAsync();
```
### Consumer
@@ -162,7 +182,7 @@ using Apache.Iggy.Extensions;
using Apache.Iggy.Factory;
using Apache.Iggy.Kinds;
-var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
+using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp
@@ -171,7 +191,7 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
await client.ConnectAsync();
await client.LoginUserAsync("iggy", "iggy");
-var consumer = client.CreateConsumerBuilder(
+await using var consumer = client.CreateConsumerBuilder(
Identifier.String("sample-stream"),
Identifier.String("sample-topic"),
Consumer.New(1))
@@ -193,5 +213,5 @@ await foreach (var message in consumer.ReceiveAsync())
## Next steps
- [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
+- [High-level SDK](/docs/sdk/csharp/high-level-sdk) - `IggyPublisher` / `IggyConsumer` with background sending, retries, auto-commit, typed (de)serialization, and pooled (rented) buffers to avoid separate payload copies on TCP
- [Examples](/docs/sdk/csharp/examples) - producer, consumer-group, and typed-message samples, plus links to runnable projects
diff --git a/content/docs/sdk/go/examples.mdx b/content/docs/sdk/go/examples.mdx
index b51e663e..b44a1be9 100644
--- a/content/docs/sdk/go/examples.mdx
+++ b/content/docs/sdk/go/examples.mdx
@@ -7,17 +7,17 @@ A runnable getting-started example lives in the [examples/go](https://github.com
## Starting the server
-The Go SDK speaks the VSR (Viewstamped Replication) wire protocol, so the examples run against the VSR server. The examples log in as `iggy`/`iggy`, and root credentials are applied **only on the very first startup**, when no data directory exists yet. Start the server with the default root credentials:
+The Go SDK speaks the VSR (Viewstamped Replication) wire protocol. Build the SDK and server from the same checkout for unreleased changes. The examples require Go 1.25 or newer and log in as `iggy`/`iggy`. From the repository root, start a disposable development server with no `IGGY_ROOT_USERNAME` or `IGGY_ROOT_PASSWORD` overrides:
```bash
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-If an example fails to log in, the server data directory (`local_data` by default) was created with different credentials: delete it and start the server again with the defaults. This setup is intended only for development and testing.
+`--fresh` **wipes this replica's local data directory** (`local_data` by default). Environment credentials take precedence over the flag, and bootstrap settings do not replace recovered credentials. A fresh cluster replica can recover credentials from peers. This setup is intended only for development and testing.
## Getting started
-The [producer](https://github.com/apache/iggy/blob/master/examples/go/getting-started/producer/main.go) creates `sample-stream` and `sample-topic` (if missing) and sends 5 batches of 10 messages to partition `0`. The [consumer](https://github.com/apache/iggy/blob/master/examples/go/getting-started/consumer/main.go) polls them from offset `0` and exits after 5 batches. Stream, topic, and partition IDs are **0-based**.
+The [producer](https://github.com/apache/iggy/blob/master/examples/go/getting-started/producer/main.go) creates `sample-stream` and `sample-topic` (if missing) and sends 5 batches of 10 messages to partition `0`. The [consumer](https://github.com/apache/iggy/blob/master/examples/go/getting-started/consumer/main.go) polls them from offset `0` and exits after 5 batches. Stream, topic, and partition IDs are **0-based**. These repository examples use numeric stream and topic IDs `0`, so crea [...]
Run both from the `examples/go` directory:
@@ -26,7 +26,7 @@ go run ./getting-started/producer/main.go
go run ./getting-started/consumer/main.go
```
-Run multiple producers and consumers at once to see how messages are distributed across clients.
+Each consumer reads partition `0` independently, starting from offset `0` or the earliest retained message. Running several consumers reads the same messages; these examples do not join a consumer group.
Both binaries accept the same flags:
@@ -37,16 +37,16 @@ Both binaries accept the same flags:
## TLS
-Start the server with TLS enabled, using the development certificates from the repository:
+From the repository root, start a disposable server with TLS enabled and the development certificates:
```bash
IGGY_TCP_TLS_ENABLED=true \
IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
-cargo run --bin iggy-server
+cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-Then run the examples with the TLS flags (paths relative to `examples/go`):
+The same data and credential prerequisites apply. These certificates are for development only. Run the producer before the consumer with the TLS flags (paths relative to `examples/go`):
```bash
go run ./getting-started/producer/main.go --tcp-server-address localhost:8090 --tls --tls-ca-file ../../core/certs/iggy_ca_cert.pem
diff --git a/content/docs/sdk/go/intro.mdx b/content/docs/sdk/go/intro.mdx
index e9306621..7a120bdf 100644
--- a/content/docs/sdk/go/intro.mdx
+++ b/content/docs/sdk/go/intro.mdx
@@ -7,24 +7,36 @@ The Iggy Go SDK is a client library for interacting with the Iggy server from Go
## Server compatibility
-The SDK and the server must speak the same wire protocol. Tagged releases up to `v0.8.0` predate VSR (and the current API) and pair only with servers of the same era. The module on `master` speaks VSR only and requires a current server. When working from source, build the SDK and the server from the same repository checkout. The code on this page tracks `master`.
+The SDK and the server must speak the same wire protocol. Tagged releases up to `v0.8.0` predate VSR (and the current API) and pair only with servers of the same era. The current SDK speaks VSR only. This page prepares for server `0.9.0`; use a compatible SDK release or edge version. When working from source, build the SDK and the server from the same repository checkout.
## Installation
+The current source requires Go 1.25 or newer. Run these commands from your application module:
+
```bash
go get github.com/apache/iggy/foreign/go
```
-`go get` resolves to the latest tagged release. Until a post-VSR release is tagged, track `master` to talk to a current server:
+`go get` without a version does not automatically select prereleases. For prerelease testing, pin a compatible VSR edge version, for example:
```bash
-go get github.com/apache/iggy/foreign/go@master
+go get github.com/apache/iggy/foreign/go@v0.9.0-edge.6
```
+For unreleased changes, use the local SDK replacement shown in the [repository examples](/docs/sdk/go/examples). A published edge tag need not contain every change in your checkout.
+
## Quick start
The snippets below follow the [getting-started example](https://github.com/apache/iggy/tree/master/examples/go/getting-started). Stream, topic, and partition IDs are **0-based**: the first partition of a topic is partition `0`.
+Run the server from the repository root with disposable development data and no root credential overrides:
+
+```bash
+cargo run --bin iggy-server -- --fresh --with-default-root-credentials
+```
+
+`--fresh` **wipes this replica's local data directory**. `IGGY_ROOT_USERNAME` and `IGGY_ROOT_PASSWORD` take precedence over the flag; bootstrap settings do not replace recovered credentials, including credentials recovered from cluster peers. The snippets use `iggy`/`iggy`. Run the producer before the consumer.
+
### Producer
```go
@@ -183,28 +195,30 @@ func main() {
continue
}
- offset += uint64(len(polled.Messages))
for _, message := range polled.Messages {
fmt.Printf("Offset: %d, Payload: %s\n",
message.Header.Offset, string(message.Payload))
+ offset = message.Header.Offset + 1
}
}
}
```
-Polled message payloads and user headers alias the reply buffer, so retaining one message pins the whole reply. Copy the bytes out when they outlive the poll.
+The consumer reads partition `0` independently and keeps its offset only in memory. Restarting reads from offset `0` again, or the earliest retained message if older messages were removed.
+
+Uncompressed payloads and user headers alias the reply buffer, so retaining their slices pins the whole reply. Copy those bytes to retain only the message you need. S2 decompression allocates a separate payload buffer.
## TLS
-Wrap the TCP options with `tcp.WithTLS`:
+Replace the quick-start client constructor with the following TCP options. Start the TLS server as shown on the [examples page](/docs/sdk/go/examples#tls); the development CA path below is relative to `examples/go`:
```go
cli, err := client.NewIggyClient(
client.WithTcp(
- tcp.WithServerAddress("iggy.example.com:8090"),
+ tcp.WithServerAddress("localhost:8090"),
tcp.WithTLS(
- tcp.WithTLSCAFile("/path/to/ca.pem"),
- tcp.WithTLSDomain("iggy.example.com"),
+ tcp.WithTLSCAFile("../../core/certs/iggy_ca_cert.pem"),
+ tcp.WithTLSDomain("localhost"),
),
),
)
@@ -214,7 +228,9 @@ cli, err := client.NewIggyClient(
## Delivery semantics
-`SendMessages` returns the placements the server committed. Delivery is at-least-once. A send whose reply is lost to a dropped connection returns `ErrDisconnected` without a replay: a reconnect registers a fresh client identity, so the server could not deduplicate the replay against a batch that may have already committed. Retrying such a send is the caller's decision and may write the batch twice. Consumers that need exactly-once handling deduplicate on the message id. A confirmation re [...]
+`SendMessages` returns any placements the server reports. A send whose reply is lost to a dropped connection returns `ErrDisconnected` without a replay: a reconnect registers a fresh client identity, so the server could not deduplicate the replay against a batch that may have already committed. Retrying such a send is the caller's decision and may write the batch twice. Consumers must handle duplicates through idempotent processing or application-level deduplication.
+
+Crash durability follows the topic's `durability` policy: `replicated` confirms replication, while `persisted` also waits for the required replicas to persist the message data. See [Durability](/docs/server/durability). An empty confirmation list is a valid success but does not by itself prove that new messages were appended.
## Consumer groups
diff --git a/content/docs/sdk/introduction.mdx b/content/docs/sdk/introduction.mdx
index 3ed4a387..ee2a8a71 100644
--- a/content/docs/sdk/introduction.mdx
+++ b/content/docs/sdk/introduction.mdx
@@ -14,7 +14,7 @@ Iggy provides official client SDKs in multiple languages. The **Rust SDK** is th
| Go | [iggy-go](https://pkg.go.dev/github.com/apache/iggy/foreign/go) | pkg.go.dev | TCP | No |
| C# | [Apache.Iggy](https://www.nuget.org/packages/Apache.Iggy/) | NuGet | TCP, HTTP | No |
| C++ | [iggy-cpp](https://github.com/apache/iggy/tree/master/foreign/cpp) | GitHub (WIP) | TCP, QUIC, HTTP, WebSocket | Yes (Rust FFI) |
-| PHP | [apache/iggy-php](https://github.com/apache/iggy/tree/master/foreign/php) | Composer (experimental) | TCP, QUIC, HTTP, WebSocket | Yes (Rust FFI) |
+| PHP | [apache/iggy-php](https://github.com/apache/iggy/tree/master/foreign/php) | GitHub (source) | TCP, QUIC, HTTP, WebSocket | Yes (Rust FFI) |
Python, C++, and PHP wrap the Rust SDK, so they inherit all transport protocols when using connection strings. Java and C# implement TCP and HTTP natively. Go and Node.js currently support TCP *only*.
@@ -22,13 +22,13 @@ The Rust SDK is always the first to receive new features.
## Prerequisites
-The samples across the SDK docs log in with the `iggy`/`iggy` root credentials. These **only exist** when the server is started with the default-credentials flag (or `IGGY_ROOT_USERNAME`/`IGGY_ROOT_PASSWORD` set) on its first boot. Otherwise root gets a generated password and the samples fail with `InvalidCredentials`:
+The examples below use the `iggy`/`iggy` root credentials. For a new local development instance, with no `IGGY_ROOT_USERNAME` or `IGGY_ROOT_PASSWORD` overrides, use:
```bash
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-`--fresh` **wipes the local data directory**, so an existing server with different stored credentials is re-initialized. This setup is for *development only*.
+`--fresh` **wipes this replica's local data directory**. In a cluster, it can recover stored credentials from another replica. The environment takes precedence over the default-credentials flag, and bootstrap settings do not replace recovered credentials. This setup is for *development only*; use a disposable data directory. See [Connection Strings](/docs/sdk/connection-strings) for the authentication details.
## Connection string
@@ -58,29 +58,34 @@ iggy://<personal-access-token>@host:port
**Examples:**
+The Rust snippets assume an async Tokio context. HTTP requires an explicit login after construction; connection-string credentials are applied automatically only by the binary transports.
+
```rust
+use iggy::prelude::*;
+
// Rust - TCP with default options
-let client = IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+let client = IggyClient::from_connection_string("iggy://iggy:iggy@127.0.0.1:8090")?;
// Rust - QUIC
-let client = IggyClient::from_connection_string("iggy+quic://iggy:iggy@localhost:8080")?;
+let client = IggyClient::from_connection_string("iggy+quic://iggy:iggy@127.0.0.1:8080")?;
// Rust - TCP with options
let client = IggyClient::from_connection_string(
- "iggy://iggy:iggy@localhost:8090?tls=true&reconnection_retries=unlimited&heartbeat_interval=5s"
+ "iggy://iggy:iggy@127.0.0.1:8090?nodelay=true&reconnection_retries=unlimited&heartbeat_interval=5s"
)?;
```
```python
-# Python
-client = IggyClient.from_connection_string("iggy://iggy:iggy@localhost:8090")
+from apache_iggy import IggyClient
+
+client = IggyClient.from_connection_string("iggy://iggy:iggy@127.0.0.1:8090")
```
The option keys differ per transport and unknown keys are rejected as **hard errors**. See [Connection Strings](/docs/sdk/connection-strings) for the full per-transport option tables, defaults, and default ports.
## Common operations
-All SDKs provide the same core operations through the unified client interface:
+The Rust SDK exposes these operations through traits on `IggyClient`. Availability and method names vary across foreign SDKs and transports:
- **SystemClient** - ping, stats, snapshot
- **StreamClient** - create, get, list, update, delete, purge streams
@@ -96,19 +101,19 @@ All SDKs provide the same core operations through the unified client interface:
## Polling strategies
-When polling messages, you can choose from several strategies:
+When polling messages, you can choose from several strategies (Rust method names are shown):
| Strategy | Description |
|----------|-------------|
| `offset(n)` | Start from a specific offset |
-| `timestamp(t)` | Start after a specific timestamp |
+| `timestamp(t)` | Start at or after the given broker append timestamp |
| `first()` | Start from the earliest available message |
-| `last()` | Start from the latest message |
-| `next()` | Continue from the last committed offset |
+| `last()` | Read the tail ending at the committed offset, up to the requested count |
+| `next()` | Continue after the stored consumer offset, or start at zero if none is stored |
## Partitioning strategies
-When sending messages, you can control partition routing:
+When sending messages, you can control partition routing (Rust method names are shown):
| Strategy | Description |
|----------|-------------|
@@ -116,4 +121,4 @@ When sending messages, you can control partition routing:
| `balanced()` | Round-robin across partitions |
| `messages_key(key)` | Hash-based routing by key |
-Strategies resolve in the client on binary transports, and on the server for the HTTP API and the Node SDK.
+The Rust SDK resolves these strategies before sending a binary request. The Rust HTTP client and Node SDK send the strategy to the server for resolution.
diff --git a/content/docs/sdk/java/examples.mdx b/content/docs/sdk/java/examples.mdx
index e56e6df7..441b8abc 100644
--- a/content/docs/sdk/java/examples.mdx
+++ b/content/docs/sdk/java/examples.mdx
@@ -3,21 +3,21 @@ title: Examples
description: "Runnable Java examples from the core repository, built with Gradle, and how to start a server for them."
---
-Runnable examples live in the [examples/java](https://github.com/apache/iggy/tree/master/examples/java) directory of the core repository as a standalone Gradle project. The project builds against the in-repo SDK (via an `includeBuild` substitution), so the examples always match the SDK source in the same checkout. Java 17 is recommended. The included `gradlew` wrapper downloads the pinned Gradle version on first run.
+Runnable examples live in the [examples/java](https://github.com/apache/iggy/tree/master/examples/java) directory of the core repository as a standalone Gradle project. The project builds against the in-repo SDK (via an `includeBuild` substitution), so the examples always match the SDK source in the same checkout. Java 17 or newer is required. The included `gradlew` wrapper downloads the pinned Gradle version on first run.
## Starting the server
-The Java SDK speaks the VSR (Viewstamped Replication) wire protocol, so the examples run against the VSR server. The examples log in as `iggy`/`iggy`, and root credentials are applied **only on the very first startup**, when no data directory exists yet. Start the server with the default root credentials:
+These examples prepare for server **0.9.0** and speak the VSR (Viewstamped Replication) wire protocol. Build the SDK and server from the same checkout for unreleased changes. The examples log in as `iggy`/`iggy`. From the repository root, start a disposable development server with no `IGGY_ROOT_USERNAME` or `IGGY_ROOT_PASSWORD` overrides:
```bash
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-Alternatively, set `IGGY_ROOT_USERNAME=iggy` and `IGGY_ROOT_PASSWORD=iggy` before the first start. If an example fails with `InvalidCredentials`, the server data directory (`local_data` by default) was created with different credentials: delete it and start the server again with the defaults. This setup is intended only for development and testing.
+`--fresh` **wipes this replica's local data directory** (`local_data` by default). Environment credentials take precedence over the flag, and bootstrap settings do not replace recovered credentials. A fresh cluster replica can recover credentials from peers. This setup is intended only for development and testing.
## Running the examples
-Run each example from the `examples/java` directory with its Gradle task:
+Run each example from the `examples/java` directory with its Gradle task. Run the producer before its consumer. The ordinary consumers keep their cursor in memory and start from offset `0` or the earliest retained message on each run; the multi-tenant consumer uses consumer groups and server-managed offsets.
- **[gettingstarted](https://github.com/apache/iggy/tree/master/examples/java/src/main/java/org/apache/iggy/examples/gettingstarted)** - basic blocking producer and consumer, the best starting point.
@@ -26,7 +26,7 @@ Run each example from the `examples/java` directory with its Gradle task:
./gradlew runGettingStartedConsumer
```
-- **[messageheaders](https://github.com/apache/iggy/tree/master/examples/java/src/main/java/org/apache/iggy/examples/messageheaders)** - message metadata via custom header keys and values, with header-based routing instead of payload-based typing.
+- **[messageheaders](https://github.com/apache/iggy/tree/master/examples/java/src/main/java/org/apache/iggy/examples/messageheaders)** - message metadata via custom header keys and values, with a `message_type` header selecting the application handler.
```bash
./gradlew runMessageHeadersProducer
@@ -47,13 +47,13 @@ Run each example from the `examples/java` directory with its Gradle task:
./gradlew runMultiTenantConsumer
```
-- **[sinkdataproducer](https://github.com/apache/iggy/tree/master/examples/java/src/main/java/org/apache/iggy/examples/sinkdataproducer)** - high-volume data generation (1000+ messages per batch) with realistic records, for testing and benchmarking.
+- **[sinkdataproducer](https://github.com/apache/iggy/tree/master/examples/java/src/main/java/org/apache/iggy/examples/sinkdataproducer)** - high-volume data generation (100 batches of 1000 to 1099 messages) with realistic records, for testing and benchmarking.
```bash
./gradlew runSinkDataProducer
```
-- **[streambuilder](https://github.com/apache/iggy/tree/master/examples/java/src/main/java/org/apache/iggy/examples/streambuilder)** - a combined producer and consumer in a single class.
+- **[streambuilder](https://github.com/apache/iggy/tree/master/examples/java/src/main/java/org/apache/iggy/examples/streambuilder)** - a combined producer and consumer in a single class. It deletes its `test_stream` stream after the run.
```bash
./gradlew runStreamBasic
@@ -66,15 +66,17 @@ Run each example from the `examples/java` directory with its Gradle task:
./gradlew runAsyncConsumer
```
-- **[tcptls](https://github.com/apache/iggy/tree/master/examples/java/src/main/java/org/apache/iggy/examples/tcptls)** - TLS-encrypted TCP connections with CA certificate verification. Requires a TLS-enabled server:
+- **[tcptls](https://github.com/apache/iggy/tree/master/examples/java/src/main/java/org/apache/iggy/examples/tcptls)** - TLS-encrypted TCP connections with CA certificate verification. Requires a TLS-enabled server. Run this server command from the repository root:
```bash
IGGY_TCP_TLS_ENABLED=true \
IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
- cargo run --bin iggy-server
+ cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
+ The same data and credential prerequisites apply. These certificates are for development only. Return to `examples/java` for the client commands; their CA path is relative to that directory.
+
```bash
./gradlew runTcpTlsProducer
./gradlew runTcpTlsConsumer
diff --git a/content/docs/sdk/java/intro.mdx b/content/docs/sdk/java/intro.mdx
index 651b1c0c..884aae57 100644
--- a/content/docs/sdk/java/intro.mdx
+++ b/content/docs/sdk/java/intro.mdx
@@ -7,24 +7,45 @@ The Iggy Java SDK is a client library for interacting with the Iggy server from
## Server compatibility
-The TCP transport speaks the VSR (Viewstamped Replication) wire protocol. The SDK and the server must speak the same protocol: the latest published artifact (`0.6.0`) predates VSR and only pairs with servers of the same era, while the SDK on `master` requires a current server. When working from source, build the SDK and the server from the same repository checkout. The code on this page tracks `master`.
+This documentation prepares for server **0.9.0**. The TCP transport speaks the VSR (Viewstamped Replication) wire protocol. The released Java artifact `0.8.0` uses the older protocol; use the `0.9.0-SNAPSHOT` build below for the pre-release SDK. When working from source, build the SDK and the server from the same repository checkout.
## Installation
+Use Java 17 or newer. These examples use the ASF snapshot repository until the `0.9.0` artifact is published to Maven Central.
+
### Maven
+Add the repository and dependency under the `project` element in `pom.xml`:
+
```xml
-<dependency>
- <groupId>org.apache.iggy</groupId>
- <artifactId>iggy</artifactId>
- <version>0.6.0</version>
-</dependency>
+<repositories>
+ <repository>
+ <id>apache-snapshots</id>
+ <url>https://repository.apache.org/content/repositories/snapshots/</url>
+ <snapshots><enabled>true</enabled></snapshots>
+ </repository>
+</repositories>
+
+<dependencies>
+ <dependency>
+ <groupId>org.apache.iggy</groupId>
+ <artifactId>iggy</artifactId>
+ <version>0.9.0-SNAPSHOT</version>
+ </dependency>
+</dependencies>
```
### Gradle
```groovy
-implementation 'org.apache.iggy:iggy:0.6.0'
+repositories {
+ mavenCentral()
+ maven { url = uri('https://repository.apache.org/content/repositories/snapshots/') }
+}
+
+dependencies {
+ implementation 'org.apache.iggy:iggy:0.9.0-SNAPSHOT'
+}
```
Check [Maven Central](https://central.sonatype.com/artifact/org.apache.iggy/iggy) for the latest published version. Snapshot builds are available from the [ASF snapshot repository](https://repository.apache.org/content/repositories/snapshots/).
@@ -33,6 +54,8 @@ Check [Maven Central](https://central.sonatype.com/artifact/org.apache.iggy/iggy
The snippets below follow the [getting-started example](https://github.com/apache/iggy/tree/master/examples/java/src/main/java/org/apache/iggy/examples/gettingstarted). Stream, topic, and partition IDs are **0-based**: the first partition of a topic is partition `0`.
+Start a development server using the [example setup](/docs/sdk/java/examples#starting-the-server), with `iggy`/`iggy` credentials, then run the producer before the consumer. Bootstrap credentials do not replace recovered credentials.
+
### Producer
```java
@@ -135,14 +158,16 @@ public class SampleConsumer {
for (Message message : polledMessages.messages()) {
String payload = new String(message.payload(), StandardCharsets.UTF_8);
System.out.printf("Offset: %d, Payload: %s%n", message.header().offset(), payload);
+ offset = message.header().offset().add(BigInteger.ONE);
}
- offset = offset.add(BigInteger.valueOf(polledMessages.messages().size()));
}
}
}
}
```
+The consumer reads partition `0` independently and stops on an empty poll. Its cursor starts at offset `0` or the earliest retained message each time it runs; `autoCommit` is disabled, so this sample does not store its progress on the server.
+
## Client types and configuration
`org.apache.iggy.Iggy` is the unified entrypoint for building clients. `Iggy.tcpClientBuilder().blocking()` returns the same builder as `IggyTcpClient.builder()` used above.
@@ -175,18 +200,21 @@ var httpClient = Iggy.httpClientBuilder()
.buildAndLogin();
```
-The TCP builder also accepts TLS and resilience options:
+Close clients when finished: `client.close()`, `asyncClient.close().join()`, and `httpClient.close()`. The HTTP close method declares `IOException`.
+
+The TCP builder also accepts TLS and resilience options. Start the [TLS development server](/docs/sdk/java/examples#running-the-examples) and run this fragment with `examples/java` as the working directory:
```java
import java.time.Duration;
+import org.apache.iggy.Iggy;
import org.apache.iggy.config.RetryPolicy;
var client = Iggy.tcpClientBuilder()
.blocking()
- .host("iggy.example.com")
+ .host("localhost")
.port(8090)
.enableTls()
- .tlsCertificate("/path/to/ca.pem") // optional custom CA
+ .tlsCertificate("../../core/certs/iggy_ca_cert.pem")
.connectionTimeout(Duration.ofSeconds(10))
.requestTimeout(Duration.ofSeconds(30))
.retryPolicy(RetryPolicy.exponentialBackoff())
@@ -194,7 +222,7 @@ var client = Iggy.tcpClientBuilder()
.buildAndLogin();
```
-Beyond streams, topics, and messages, the client exposes consumer groups, partitions, users, and personal access tokens through `client.consumerGroups()`, `client.partitions()`, `client.users()`, and `client.personalAccessTokens()`. All SDK exceptions inherit from `IggyException`.
+Beyond streams, topics, and messages, the client exposes consumer groups, partitions, users, and personal access tokens through `client.consumerGroups()`, `client.partitions()`, `client.users()`, and `client.personalAccessTokens()`. The SDK's custom exception types inherit from `IggyException`; failed futures can wrap them in `CompletionException` when joined.
## Examples
diff --git a/content/docs/sdk/node/examples.mdx b/content/docs/sdk/node/examples.mdx
index caae3e0c..a338baac 100644
--- a/content/docs/sdk/node/examples.mdx
+++ b/content/docs/sdk/node/examples.mdx
@@ -1,47 +1,52 @@
---
title: Examples
-description: "TypeScript examples for the Node.js SDK, exercised by CI so they match the current SDK."
+description: "Build the local Node.js SDK and run its TypeScript examples."
---
-Working examples are available in the [examples/node](https://github.com/apache/iggy/tree/master/examples/node) directory, written in TypeScript. CI exercises them, so they always match the current SDK:
+Examples are available in the [examples/node](https://github.com/apache/iggy/tree/master/examples/node) directory, written in TypeScript. They use the SDK from the same checkout:
- **getting-started** - basic producer and consumer
- **basic** - producer and consumer with utilities
- **message-envelope** - JSON message envelope pattern
-- **message-headers** - custom message headers
-- **multi-tenant** - multi-tenant streaming setup
+- **message-headers** - message-type dispatch using a JSON wrapper in the payload
+- **multi-tenant** - separate streams for tenants, using one client
- **tcp-tls** - TLS-encrypted TCP connections
- **stream-builder** - end-to-end walkthrough: create a stream and topic, produce, and consume
- **sink-data-producer** - bulk data generation for sink connectors
## Running the examples
-Start an Iggy server first:
+These examples target server **0.9.0**. Start the server from the same source checkout; see [Getting started](/docs/introduction/getting-started) for prerequisites:
```bash
-# Latest release
-docker run --rm \
- --cap-add=SYS_NICE --security-opt seccomp=unconfined --ulimit memlock=-1:-1 \
- -p 8090:8090 \
- -e IGGY_TCP_ADDRESS=0.0.0.0:8090 \
- -e IGGY_NODE_ADVERTISED_ADDRESS=localhost \
- -e IGGY_ROOT_USERNAME=iggy -e IGGY_ROOT_PASSWORD=iggy \
- apache/iggy:latest
-
-# Or from the repository source
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-The environment variables make the server reachable through the published port (it binds to `127.0.0.1` inside the container by default) and set the `iggy`/`iggy` root credentials the examples default to.
+The default-root flag sets the `iggy`/`iggy` credentials used by the examples. `--fresh` deletes existing local server data. Explicit credential environment variables override the default-root flag; bootstrap settings do not replace recovered credentials.
-Then, from `examples/node`, install the dependencies and run a producer/consumer pair:
+From the repository root, build the local SDK before installing the examples. The stream-builder example creates a stream and topic, sends and consumes three messages, and deletes its resources:
```bash
-npm ci
-npm run test:getting-started:producer
-npm run test:getting-started:consumer
+npm --prefix foreign/node ci
+npm --prefix foreign/node run build
+npm --prefix examples/node ci
+cd examples/node
+DEBUG=iggy:examples npm run test:stream-builder
```
-Each example set has a matching `test:<name>` script in `examples/node/package.json`.
+Each example has a script in `examples/node/package.json`. Set `DEBUG=iggy:examples*` to see its progress.
+
+The separate producer and consumer scripts do not consistently share resources: the envelope and headers consumers create new empty topics, and the tenant producer deletes its streams before exiting. The getting-started and basic producers spread batches over five partitions, while their consumers poll partition 0. Use the self-contained stream-builder command above for a complete send-and-consume walkthrough. The sink-data producer deletes its selected topic and stream when it finishes, [...]
+
+The TCP/TLS pair needs a TLS-enabled source server. Run this command from the repository root, then the two `test:tcp-tls:*` scripts from `examples/node`, where they load `../../core/certs/iggy_ca_cert.pem`:
+
+```bash
+IGGY_TCP_TLS_ENABLED=true \
+IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
+IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
+cargo run --bin iggy-server -- --fresh --with-default-root-credentials
+```
+
+These are repository development certificates.
For the quick-start code itself, see the [Node.js SDK intro](/docs/sdk/node/intro).
diff --git a/content/docs/sdk/node/intro.mdx b/content/docs/sdk/node/intro.mdx
index 62573825..717a947b 100644
--- a/content/docs/sdk/node/intro.mdx
+++ b/content/docs/sdk/node/intro.mdx
@@ -5,33 +5,35 @@ description: "The Node.js and TypeScript SDK, from installation to keeping clien
The Iggy Node.js SDK is a client library that allows you to interact with the Iggy API from your Node.js and TypeScript applications. It communicates with the Iggy server over TCP or TLS using the binary protocol. The package is available on [npm](https://www.npmjs.com/package/apache-iggy) and the source code can be found on [GitHub](https://github.com/apache/iggy/tree/master/foreign/node).
-The SDK speaks only the current (VSR) wire protocol and doesn't fall back to older formats. Keep the client and server versions in step: pair the newest package with the newest server.
+These docs target server **0.9.0**. Use the Node SDK's `edge` package with server 0.9.0 or `edge`; the stable Node package 0.8.0 uses the older protocol. The current SDK speaks only the VSR wire protocol and doesn't fall back to older formats.
## Installation
```bash
-npm install apache-iggy
+npm install apache-iggy@edge
```
## Quick start
-The samples below expect an Iggy server on `127.0.0.1:8090`:
+The samples below expect an Iggy server on `127.0.0.1:8090`. See [Getting started](/docs/introduction/getting-started) for the server prerequisites:
```bash
-# Latest release
+# Development image for the 0.9.0 release
docker run --rm \
--cap-add=SYS_NICE --security-opt seccomp=unconfined --ulimit memlock=-1:-1 \
-p 8090:8090 \
-e IGGY_TCP_ADDRESS=0.0.0.0:8090 \
-e IGGY_NODE_ADVERTISED_ADDRESS=localhost \
-e IGGY_ROOT_USERNAME=iggy -e IGGY_ROOT_PASSWORD=iggy \
- apache/iggy:latest
+ apache/iggy:edge
-# Or from the repository source
+# Or from the matching repository root
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-The environment variables make the server reachable through the published port (it binds to `127.0.0.1` inside the container by default) and set the root credentials the samples log in with. Without `--with-default-root-credentials` (or the `IGGY_ROOT_USERNAME` / `IGGY_ROOT_PASSWORD` variables), a first boot generates a random root password instead of `iggy`/`iggy`.
+The environment variables make the server reachable through the published port (it binds to `127.0.0.1` inside the container by default) and set the root credentials the samples log in with. Without `--with-default-root-credentials` (or the `IGGY_ROOT_USERNAME` / `IGGY_ROOT_PASSWORD` variables), a first boot generates a random root password instead of `iggy`/`iggy`. For source runs, `--fresh` deletes existing local server data. Explicit credential environment variables take precedence ov [...]
+
+Save the following snippets as `producer.mjs` and `consumer.mjs` in the project where you installed the SDK. Run `node producer.mjs` before `node consumer.mjs`.
### Producer
@@ -81,7 +83,7 @@ console.log(`Sent ${messages.length} message(s)`);
await client.destroy();
```
-The Node SDK sends the partitioning choice on the wire, and the server resolves balanced and key routing at admission (the Rust, C#, Java, and Go clients instead resolve them client-side).
+The Node SDK sends the partitioning choice on the wire, and the server resolves balanced and key routing at admission (the Rust, C#, Java, and Go binary clients instead resolve them client-side).
### Consumer
@@ -98,8 +100,8 @@ const client = new Client({
credentials: { username: 'iggy', password: 'iggy' },
});
-// Next with autocommit continues from this consumer's last committed offset,
-// so each run picks up where the previous one finished.
+// Next starts after consumer 0's stored offset.
+// Autocommit requests offset storage before application processing.
const polledMessages = await client.message.poll({
streamId: STREAM_NAME,
topicId: TOPIC_NAME,
@@ -118,19 +120,21 @@ for (const message of polledMessages.messages) {
await client.destroy();
```
+The poll response does not wait for the auto-commit to finish. The sample processes messages after polling, so auto-commit does not confirm that the application handled each message.
+
## Configuration
The client constructor accepts a few options:
- **reconnect**: automatic reconnection with `{ enabled, interval, maxRetries }`.
-- **heartbeatInterval**: the client pings the server every 5 seconds by default. `0` disables heartbeats. The server evicts connections that stay silent too long, so keep it well below the server's heartbeat timeout.
+- **heartbeatInterval**: the client pings the server every 5 seconds by default. `0` disables heartbeats. When server heartbeat eviction is enabled, silent consumer-group members can be evicted, so keep it well below the server's heartbeat timeout.
- **maxResponseFrameSize**: response frames larger than this limit (default 64 MiB) are rejected and close the connection. Raise it when polling very large batches.
- **poolSize**: the client currently supports exactly one pooled connection. `min`/`max` above 1 throw.
- **TLS**: set `transport: 'TLS'` and pass Node `tls.ConnectionOptions` in `options`. See the [tcp-tls example](https://github.com/apache/iggy/tree/master/examples/node/src/tcp-tls).
## Beyond the basics
-- **Consumer groups**: `client.group` manages groups (including ensure-and-join in one call), and polling with `Consumer.Group` distributes partitions across group members.
+- **Consumer groups**: `client.group` manages groups (including ensure-and-join in one call), and polling with `Consumer.Group(groupId)` and `partitionId: null` follows the partitions assigned to that member.
- **Consumer streams**: `singleConsumerStream` and `groupConsumerStream` expose polling as Node streams.
- **Administration**: users, personal access tokens, offsets, partitions, segments, and cluster metadata are all available on the client.
- **Errors**: the package exports `ResponseError`, `DeserializeError`, `ProtocolFrameError`, and `VsrEvictionError` for precise error handling.
@@ -139,4 +143,4 @@ See the [foreign/node README](https://github.com/apache/iggy/tree/master/foreign
## Examples
-Working examples are available in the [examples/node](https://github.com/apache/iggy/tree/master/examples/node) directory, written in TypeScript. See [Examples](/docs/sdk/node/examples) for the list and how to run them.
+Examples are available in the [examples/node](https://github.com/apache/iggy/tree/master/examples/node) directory, written in TypeScript. See [Examples](/docs/sdk/node/examples) for the list and how to run them.
diff --git a/content/docs/sdk/php/intro.mdx b/content/docs/sdk/php/intro.mdx
index d29c1087..37857efe 100644
--- a/content/docs/sdk/php/intro.mdx
+++ b/content/docs/sdk/php/intro.mdx
@@ -34,25 +34,25 @@ On macOS the library is `libiggy_php.dylib`.
## Quick start
-The sample below expects an Iggy server on `127.0.0.1:8090`:
+The sample below targets Iggy 0.9.0 on `127.0.0.1:8090`. When testing unreleased SDK changes, build the server from the same source checkout:
```bash
-# Latest release
+# Server 0.9.0
docker run --rm \
--cap-add=SYS_NICE --security-opt seccomp=unconfined --ulimit memlock=-1:-1 \
-p 8090:8090 \
-e IGGY_TCP_ADDRESS=0.0.0.0:8090 \
-e IGGY_NODE_ADVERTISED_ADDRESS=localhost \
-e IGGY_ROOT_USERNAME=iggy -e IGGY_ROOT_PASSWORD=iggy \
- apache/iggy:latest
+ apache/iggy:0.9.0
-# Or from the repository source
+# Or from the repository root
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-The environment variables make the server reachable through the published port (it binds to `127.0.0.1` inside the container by default) and set the `iggy`/`iggy` root credentials the sample's connection string uses.
+The environment variables make the server reachable through the published port (it binds to `127.0.0.1` inside the container by default) and bootstrap the `iggy`/`iggy` root credentials for a new data directory. Existing stored credentials are not replaced. For the source command, ensure `IGGY_ROOT_USERNAME` and `IGGY_ROOT_PASSWORD` are unset or match the sample; environment values take precedence over the flag. `--fresh` removes the local replica state, so use it only for disposable dev [...]
-Credentials in the connection string take effect on `connect()`. Alternatively, construct the client with a plain address (`new \Iggy\Client('127.0.0.1:8090')`) and call `loginUser('iggy', 'iggy')` explicitly.
+For binary transports, credentials in the connection string take effect on `connect()`. HTTP requires `loginUser()` explicitly. Alternatively, construct the client with a plain address (`new \Iggy\Client('127.0.0.1:8090')`) and call `loginUser('iggy', 'iggy')` explicitly.
```php
<?php
@@ -90,7 +90,7 @@ foreach ($polled as $message) {
## Consumer groups
-`consumerGroup()` returns an `Iggy\Consumer` that creates and joins the group by default. Consume with a bounded callback loop or a PHP iterator:
+`consumerGroup()` returns an `Iggy\Consumer` that creates and joins the group by default. The partition argument is ignored for consumer groups; the server assigns partitions. After the quickstart has sent ten messages, these two loops consume five each. Both wait for messages until their limit is reached:
```php
<?php
@@ -105,19 +105,22 @@ $consumer = $client->consumerGroup(
\Iggy\AutoCommit::disabled(),
);
-// Callback style: the message limit is required and must be finite.
+$limit = 5;
$consumer->consumeMessages(
function (\Iggy\ReceiveMessage $message) use ($consumer): void {
- process($message->payload());
+ echo $message->payload(), PHP_EOL;
$consumer->storeOffset($message->offset(), $message->partitionId());
},
- 100,
+ $limit,
);
-// Iterator style: break out when done.
+$consumed = 0;
foreach ($consumer->iterMessages() as $message) {
- process($message->payload());
+ echo $message->payload(), PHP_EOL;
$consumer->storeOffset($message->offset(), $message->partitionId());
+ if (++$consumed === $limit) {
+ break;
+ }
}
```
@@ -129,7 +132,7 @@ foreach ($consumer->iterMessages() as $message) {
- Partition ids are 0-based: for a topic with one partition, use `0`.
- PHP strings are passed as named identifiers, including strings that contain only digits. PHP integers are passed as numeric identifiers.
- `PollingStrategy::timestamp()` and `timestampMicros()` expect microseconds since the Unix epoch. Use `PollingStrategy::timestampSeconds()` for PHP `time()` values.
-- Large unsigned values that can overflow PHP integers, such as message checksums, are returned as decimal strings.
+- Message IDs and checksums are returned as decimal strings. Offset and timestamp getters return PHP integers and are limited to `PHP_INT_MAX`.
- Errors are thrown as exceptions under `Iggy\Exception`: `AuthenticationException`, `ConnectionException`, `NotFoundException`, `TransientException`, and the `IggyException` base.
- TLS uses the Rust SDK connection-string format, for example `iggy+tcp://user:pass@host:port?tls=true&tls_domain=localhost&tls_ca_file=/path/to/ca.pem`.
- The extension owns a lazy global Tokio runtime. Don't call `pcntl_fork()` after the first Iggy call: the child inherits file descriptors but not the runtime's worker threads.
@@ -140,9 +143,10 @@ foreach ($consumer->iterMessages() as $message) {
## Examples
-Working examples are available in the [examples/php](https://github.com/apache/iggy/tree/master/examples/php) directory. They read `IGGY_CONNECTION_STRING`, or build one from `IGGY_HOST`, `IGGY_PORT`, `IGGY_USERNAME`, and `IGGY_PASSWORD`:
+Working examples are available in the [examples/php](https://github.com/apache/iggy/tree/master/examples/php) directory. They read `IGGY_CONNECTION_STRING` for a binary transport with credentials. Otherwise they connect over TCP using `IGGY_HOST` and `IGGY_PORT`, then log in with the literal `IGGY_USERNAME` and `IGGY_PASSWORD` values:
```bash
+# From the repository root
cd examples/php
(cd ../../foreign/php && cargo build)
export PHP_IGGY_EXTENSION="$(pwd)/../../foreign/php/target/debug/libiggy_php.so"
diff --git a/content/docs/sdk/python/examples.mdx b/content/docs/sdk/python/examples.mdx
index 9bf7c115..7852bb26 100644
--- a/content/docs/sdk/python/examples.mdx
+++ b/content/docs/sdk/python/examples.mdx
@@ -3,7 +3,7 @@ title: Examples
description: "Working Python examples from the core repository, covering connection strings, message headers and TLS."
---
-Working examples are available in the [examples/python](https://github.com/apache/iggy/tree/master/examples/python) directory. CI exercises them, so they always match the current SDK:
+Working examples are available in the [examples/python](https://github.com/apache/iggy/tree/master/examples/python) directory. CI runs the source examples against a source-built server, including a separate TLS pass:
- **getting-started** - producer and consumer with optional TLS via `TcpConfig`
- **basic** - producer and consumer using connection strings
@@ -11,25 +11,25 @@ Working examples are available in the [examples/python](https://github.com/apach
## Running the examples
-Start an Iggy server first:
+These examples target Iggy 0.9.0. For unreleased changes, use the server and Python SDK from the same source checkout:
```bash
-# Latest release
+# Server 0.9.0
docker run --rm \
--cap-add=SYS_NICE --security-opt seccomp=unconfined --ulimit memlock=-1:-1 \
-p 8090:8090 \
-e IGGY_TCP_ADDRESS=0.0.0.0:8090 \
-e IGGY_NODE_ADVERTISED_ADDRESS=localhost \
-e IGGY_ROOT_USERNAME=iggy -e IGGY_ROOT_PASSWORD=iggy \
- apache/iggy:latest
+ apache/iggy:0.9.0
-# Or from the repository source
+# Or from the repository root
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-The environment variables make the server reachable through the published port (it binds to `127.0.0.1` inside the container by default) and set the `iggy`/`iggy` root credentials the examples default to.
+The environment variables make the server reachable through the published port (it binds to `127.0.0.1` inside the container by default) and bootstrap the `iggy`/`iggy` root credentials for new data. Stored credentials are not replaced, and environment credentials override the source command's default-credentials flag. Use `--fresh` only with disposable local replica data.
-Then, from `examples/python`, install the dependencies and run a producer/consumer pair:
+With Python 3.10 or newer and Rust/Cargo available, run from `examples/python`. `uv` uses the local SDK path in `pyproject.toml`; the pip command below names that path explicitly:
```bash
# Using uv
@@ -40,7 +40,7 @@ uv run getting-started/consumer.py
# Without uv
python -m venv .venv
source .venv/bin/activate
-pip install .
+pip install ../../foreign/python .
python getting-started/producer.py
python getting-started/consumer.py
```
diff --git a/content/docs/sdk/python/intro.mdx b/content/docs/sdk/python/intro.mdx
index d114ef2e..9c19f800 100644
--- a/content/docs/sdk/python/intro.mdx
+++ b/content/docs/sdk/python/intro.mdx
@@ -5,33 +5,43 @@ description: "The Python SDK, a PyO3 wrapper around the Rust SDK, and how to ins
The Iggy Python SDK is a client library that allows you to interact with the Iggy API from your Python application. It is built as a PyO3 wrapper around the Rust SDK, which means it supports TCP, QUIC, HTTP, and WebSocket transports via connection strings. The package is available on [PyPI](https://pypi.org/project/apache-iggy/) and the source code can be found on [GitHub](https://github.com/apache/iggy/tree/master/foreign/python).
-Because the wheel bundles the Rust SDK, it speaks only the current Iggy wire protocol and doesn't fall back to older formats. Keep the client and server versions in step: pair the newest wheel with the newest server.
+The wheel bundles the Rust SDK, whose binary transports speak the current Iggy wire protocol without falling back to older formats. Use an SDK release compatible with your server. For unreleased changes, build the SDK and server from the same source checkout; the published wheel can lag source API changes.
## Installation
+Python 3.10 or newer is required. Install the published package in a virtual environment:
+
```bash
+python -m venv .venv
+source .venv/bin/activate
pip install apache-iggy
```
+For the source version described here, install from the repository root with Rust and Cargo available:
+
+```bash
+pip install ./foreign/python
+```
+
## Quick start
-The samples below expect an Iggy server on `127.0.0.1:8090`:
+The samples below target Iggy 0.9.0 on `127.0.0.1:8090`:
```bash
-# Latest release
+# Server 0.9.0
docker run --rm \
--cap-add=SYS_NICE --security-opt seccomp=unconfined --ulimit memlock=-1:-1 \
-p 8090:8090 \
-e IGGY_TCP_ADDRESS=0.0.0.0:8090 \
-e IGGY_NODE_ADVERTISED_ADDRESS=localhost \
-e IGGY_ROOT_USERNAME=iggy -e IGGY_ROOT_PASSWORD=iggy \
- apache/iggy:latest
+ apache/iggy:0.9.0
-# Or from the repository source
+# Or from the same repository root as the SDK
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-The environment variables make the server reachable through the published port (it binds to `127.0.0.1` inside the container by default) and set the `iggy`/`iggy` root credentials the samples' connection string uses. Without `--with-default-root-credentials` (or the `IGGY_ROOT_USERNAME` / `IGGY_ROOT_PASSWORD` variables), a first boot generates a random root password instead.
+The environment variables make the server reachable through the published port (it binds to `127.0.0.1` inside the container by default) and bootstrap the `iggy`/`iggy` root credentials for new data. Stored credentials are not replaced. Environment credentials override the default-credentials flag, so unset them or set them to match the sample. `--fresh` deletes local replica state; use disposable development data. A new standalone server without explicit credentials generates a root pas [...]
### Producer
@@ -82,7 +92,7 @@ asyncio.run(main())
```python
import asyncio
-from apache_iggy import IggyClient, PollingStrategy
+from apache_iggy import Consumer, IggyClient, PollingStrategy
STREAM_NAME = "sample-stream"
TOPIC_NAME = "sample-topic"
@@ -95,11 +105,12 @@ async def main():
)
await client.connect()
- # Next() with auto_commit=True continues from this consumer's last committed
- # offset, so each run picks up where the previous one finished.
+ # Next() resumes this named consumer after its stored offset. Auto-commit
+ # can store the offset before application processing finishes.
polled_messages = await client.poll_messages(
stream=STREAM_NAME,
topic=TOPIC_NAME,
+ consumer=Consumer.Single("sample-consumer"),
partition_id=PARTITION_ID,
polling_strategy=PollingStrategy.Next(),
count=10,
@@ -116,8 +127,8 @@ asyncio.run(main())
## Beyond the basics
-- **TLS**: construct the client from a `TcpConfig` instead of a connection string and set `tls_enabled` plus `tls_ca_file`. The [getting-started example](https://github.com/apache/iggy/tree/master/examples/python/getting-started) shows the full setup.
-- **Consumer groups**: `client.consumer_group(...)` returns an `IggyConsumer` that creates and joins the group by default and commits offsets according to the configured `AutoCommit` mode.
+- **TLS**: use `TcpConfig` with `tls_enabled` and `tls_ca_file`, or the TLS options in a [connection string](/docs/sdk/connection-strings). The certificate must match `tls_domain`, which defaults to the dialed host. The [getting-started example](https://github.com/apache/iggy/tree/master/examples/python/getting-started) shows the full setup.
+- **Consumer groups**: `await client.consumer_group(...)` returns an `IggyConsumer` that creates and joins the group by default and commits offsets according to the configured `AutoCommit` mode. Groups require a binary transport; HTTP supports ordinary consumers only.
- **User headers**: `SendMessage(data, user_headers=...)` accepts a plain `dict` with `str`, `bytes`, `bool`, `int`, or `float` values. `HeaderKey`/`HeaderValue` give explicit control over the wire type. See the [message-headers examples](https://github.com/apache/iggy/tree/master/examples/python/message-headers).
- **Topic options**: `create_topic(..., options=...)` accepts extra option keys as a `dict[str, str]`, validated against the server's catalog. `client.describe_options("topic")` lists the keys a server accepts.
- **Administration**: user and permission management. Personal access tokens can be used for login via `AutoLogin.personal_access_token(...)` in the connection config. PAT management isn't exposed yet.
diff --git a/content/docs/sdk/rust/examples.mdx b/content/docs/sdk/rust/examples.mdx
index b3d89278..0e0617bf 100644
--- a/content/docs/sdk/rust/examples.mdx
+++ b/content/docs/sdk/rust/examples.mdx
@@ -5,13 +5,15 @@ description: "The Rust SDK examples in the core repository, and how to start a s
In the core repository, you can find the following [examples](https://github.com/apache/iggy/tree/master/examples/rust/src) using the Rust SDK.
-Before running any of them, start the server with the default root credentials. Without the flag (or `IGGY_ROOT_USERNAME`/`IGGY_ROOT_PASSWORD`) root gets a **generated password** and every example fails with `InvalidCredentials`:
+Build the SDK and server from the same checkout for unreleased changes. Start the server from the repository root in a separate terminal. The examples use `iggy`/`iggy`; for a new local instance with no `IGGY_ROOT_USERNAME` or `IGGY_ROOT_PASSWORD` overrides:
```bash
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-The examples are cargo example targets of the `iggy_examples` crate. Run them from the `examples/rust` directory with `cargo run --example <name>`. Partition IDs are **0-based**: the examples send to and poll from partition `0`.
+Environment credentials take precedence over the flag, and bootstrap settings do not replace recovered credentials. `--fresh` deletes this replica's local data; use disposable development data. A new standalone server without explicit credentials generates a root password; a new cluster requires explicit credentials.
+
+The examples are cargo example targets of the `iggy_examples` crate. Run them from the **repository root** with `cargo run --example <name>` so relative certificate paths resolve. Run each producer before its consumer. Some examples create fixed names and require fresh data when repeated; the getting-started consumer also expects stream and topic IDs `0`, so run that pair first on a fresh server. Partition IDs are **0-based**, but balanced producers and consumer groups can use multiple p [...]
- **[Getting started](https://github.com/apache/iggy/tree/master/examples/rust/src/getting-started)** - the basic example which is discussed in the [getting started](/docs/introduction/getting-started) guide.
@@ -69,7 +71,16 @@ The examples are cargo example targets of the `iggy_examples` crate. Run them fr
cargo run --example multi-tenant-consumer
```
-- **[TCP TLS](https://github.com/apache/iggy/tree/master/examples/rust/src/tcp-tls)** - TLS-encrypted TCP connections with custom CA certificates.
+- **[TCP TLS](https://github.com/apache/iggy/tree/master/examples/rust/src/tcp-tls)** - TLS-encrypted TCP connections with custom CA certificates. Stop the plain server and start a TLS-enabled server from the repository root:
+
+ ```bash
+ IGGY_TCP_TLS_ENABLED=true \
+ IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
+ IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
+ cargo run --bin iggy-server -- --fresh --with-default-root-credentials
+ ```
+
+ These certificates are for local testing. Then run:
```bash
cargo run --example tcp-tls-producer
diff --git a/content/docs/sdk/rust/high-level-sdk.mdx b/content/docs/sdk/rust/high-level-sdk.mdx
index 9ed00c03..f4b90347 100644
--- a/content/docs/sdk/rust/high-level-sdk.mdx
+++ b/content/docs/sdk/rust/high-level-sdk.mdx
@@ -6,41 +6,50 @@ description: "IggyProducer and IggyConsumer: batching, consumer groups and offse
If you've read through the [getting started](/docs/introduction/getting-started) guide, you might have noticed that it's quite verbose and requires a lot of boilerplate code to get started. This is where the High-level SDK comes in, as it does provide a more user-friendly interface to interact with the Iggy API for both, producer and consumer. Let's consider the following features:
- **Automatically creating & joining** the consumer groups
-- **Committing the offset** depending on the particular mode (e.g. in the background based on some interval, after polling N messages etc.)
+- **Committing the offset** depending on the particular mode (e.g. in the background based on some interval, when an offset matches the configured trigger etc.)
- **Batching the messages**, whether it's about producing or consuming
- **Processing the messages** as if the stream was an async iterator
- **Reusing the same client** for both, producing and consuming on the same topic without repeating the configuration
- And more...
+The Rust snippets assume an async function returning `Result<(), Box<dyn std::error::Error>>`, with `iggy`, Tokio and `futures-util` dependencies. Use the [Rust SDK setup](/docs/sdk/rust/intro) and a server with the sample credentials.
+
## Connection string
Instead of providing the configuration for the client, you can use the connection string. It's a string that contains all the necessary information to connect to the Iggy API, and it works with all four transports:
-```bash
-iggy://iggy:iggy@localhost:8090 (TCP, default)
-iggy+tcp://iggy:iggy@localhost:8090 (TCP, explicit)
-iggy+quic://iggy:iggy@localhost:8080 (QUIC)
-iggy+http://iggy:iggy@localhost:3000 (HTTP)
-iggy+ws://iggy:iggy@localhost:8092 (WebSocket)
+```
+iggy://iggy:iggy@127.0.0.1:8090 (TCP, default)
+iggy+tcp://iggy:iggy@127.0.0.1:8090 (TCP, explicit)
+iggy+quic://iggy:iggy@127.0.0.1:8080 (QUIC)
+iggy+http://iggy:iggy@127.0.0.1:3000 (HTTP)
+iggy+ws://iggy:iggy@127.0.0.1:8092 (WebSocket)
```
Which can be used to create the client like this:
```rust
-let client = IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+use futures_util::StreamExt;
+use iggy::prelude::*;
+use std::str::FromStr;
+
+let client = IggyClient::from_connection_string("iggy://iggy:iggy@127.0.0.1:8090")?;
+client.connect().await?;
```
+HTTP requires an explicit login after construction. Consumer groups require a binary transport; use an ordinary consumer with an explicit partition over HTTP.
+
Options are appended as query parameters. For TCP:
-```bash
-iggy://iggy:iggy@localhost:8090?tls=true&tls_domain=test.com&reconnection_retries=5&reconnection_interval=5s&reestablish_after=10s&heartbeat_interval=3s&nodelay=true
+```
+iggy://iggy:iggy@127.0.0.1:8090?reconnection_retries=5&reconnection_interval=5s&reestablish_after=10s&heartbeat_interval=3s&nodelay=true
```
The option keys differ per transport (for example QUIC uses `reconnection_max_retries` and `reconnection_reestablish_after` where TCP uses `reconnection_retries` and `reestablish_after`). Unknown keys are rejected as hard errors. See [Connection Strings](/docs/sdk/connection-strings) for the full per-transport option tables and defaults.
-Unless you need to provide a specific implementation of the client-side `Encryptor`, `Partitioner` or adjust some other settings, you should be good to go with the connection string. On the other hand, you can always make use of `IggyClientBuilder::from_connection_string()` to extend the options on top of the provided connection string.
+Unless you need client-side encryption, a custom `Partitioner`, or other settings outside the connection-string options, you should be good to go with the connection string. On the other hand, you can always make use of `IggyClientBuilder::from_connection_string()` to extend the options on top of the provided connection string.
-The `IggyClient` internally wraps its transport in an `IggyRwLock`, so a single instance is **cheap to clone** and can be shared across multiple tasks. Typically, it's a good idea to create a separate connection for producing and consuming, but it's not a requirement.
+The `IggyClient` internally wraps its transport in an `IggyRwLock`, but the client itself does not implement `Clone`. Share an `Arc<IggyClient>` across tasks; cloning the `Arc` keeps the same client and connection. Typically, it's a good idea to create a separate connection for producing and consuming, but it's not a requirement.
## Producer
@@ -60,7 +69,7 @@ let producer = client
producer.init().await?;
```
-The code above will result in creating the producer that will try to send the messages in batches of 1000 every 5 milliseconds. You can choose between the `direct` (an instant producer) or the `background` (which will send the messages in the background by buffering them). The partitioning is set to `balanced` which means that the producer will try to distribute the messages evenly across all the partitions. The `init()` method is used to ensure that the producer is ready to send the mes [...]
+The direct producer splits each `send()` call into requests of at most 1000 messages. For sequential calls, the 5 ms linger setting delays a call until at least 5 ms after the previous successful send; it does not buffer messages between calls or delay individual chunks within a call. Choose `background` to buffer sends on worker tasks. `balanced` routes requests round-robin across partitions. By default, `init()` creates a missing stream and topic, requiring the corresponding permissions.
Finally, you can use the `send()` method to send the messages to the topic. The producer **never needs to be a mutable binding**: both `init()` and `send()` take `&self`, so a plain `let producer` is enough for its whole life. Here's how you can send the messages:
@@ -80,22 +89,22 @@ The `background` mode buffers messages and sends them from shard workers, tradin
| Field | Default | Description |
|-------|---------|-------------|
| `num_shards` | `1` | Number of shard workers running in parallel |
-| `sharding` | `OrderedSharding` | Maps a message to a shard. `OrderedSharding` routes the same stream/topic to the same shard, preserving order. `BalancedSharding` round-robins across shards for maximum throughput, order not preserved |
+| `sharding` | `OrderedSharding` | Routes each send to a worker. `OrderedSharding` keeps a stream/topic pair on one sequential worker; `BalancedSharding` round-robins across workers for maximum throughput and can reorder sends |
| `linger_time` | `1ms` | How long a shard may wait before flushing an incomplete batch |
-| `batch_size` | 1 MiB | Maximum total **size in bytes** of a batch; `0` disables size-based batching |
-| `batch_length` | `1000` | Maximum **number of messages** per batch; `0` disables length-based batching |
-| `failure_mode` | `Block` | `BackpressureMode` applied when limits are reached: `Block`, `BlockWithTimeout(duration)`, or `FailImmediately` |
-| `max_buffer_size` | 32 MiB | Upper bound for bytes held in memory across all shards; `0` means unlimited |
-| `max_in_flight` | `1` | Maximum batches being sent concurrently. More than 1 can reorder messages when retries occur; `0` means unlimited |
+| `batch_size` | 1 MiB | Per-worker flush threshold in bytes; `0` disables this trigger |
+| `batch_length` | `1000` | Per-worker flush threshold in **queued sends**, not individual messages; `0` disables this trigger |
+| `failure_mode` | `Block` | Behavior when the byte budget is exhausted: `Block`, `BlockWithTimeout(duration)`, or `FailImmediately` |
+| `max_buffer_size` | 32 MiB | Budget for message bytes buffered or in flight across all workers; `0` disables this budget |
+| `max_in_flight` | `1` | Maximum concurrent writes across workers; each worker remains sequential. `0` uses the semaphore's maximum capacity |
| `error_callback` | `LogErrorCallback` | Async callback invoked on errors the producer cannot recover from |
-`batch_size` limits bytes while `batch_length` limits the message count. Whichever limit fires first (including `linger_time`) triggers the flush.
+`batch_size` and `batch_length` are flush triggers checked after adding a send. The final send can take the buffer past the byte threshold. The linger deadline starts when an empty buffer receives a send; the first trigger reached causes a flush. Each worker also has a bounded queue, which can make dispatch wait independently of `failure_mode`. A send larger than the entire byte budget fails under every backpressure mode. Nonzero `max_buffer_size` and `max_in_flight` values must not exce [...]
-Call `producer.shutdown().await` to flush the background buffer before exiting. Dropping the producer without it **silently discards unflushed messages**.
+Stop all senders, then call `producer.shutdown().await` to drain background queues and wait for writes and error callbacks. Dropping the producer without shutdown can lose buffered messages. Shutdown does not turn failed writes into successful ones; the default error callback logs and drops failed sends.
### Send confirmations
-Every send method returns a `SendMessagesResponse` carrying the commit confirmations of the chunks the send was split into. The list is **empty** whenever the server sends no confirmation payload, and always for a `background` producer, which hands the messages to a dispatcher and returns before the send happens. Branch on `confirmations.is_empty()` instead of indexing. Delivery is *at-least-once*: a retried chunk may already have been committed by an earlier attempt, so an offset in a c [...]
+Every send method returns a `SendMessagesResponse` carrying the commit confirmations of the chunks the send was split into. The list is **empty** whenever the server sends no confirmation payload, and always for a `background` producer, which hands the messages to a dispatcher and returns before the send happens. Branch on `confirmations.is_empty()` instead of indexing. A retried chunk may already have been committed by an earlier attempt, so a confirmation offset does not imply uniquene [...]
## Consumer
@@ -105,7 +114,7 @@ The consumer is a high-level abstraction that allows you to receive the messages
let mut consumer = client
.consumer_group("my-consumer-group", "my-stream", "my-topic")?
.auto_commit(AutoCommit::IntervalOrWhen(
- IggyDuration::from_str("1s")?,
+ NonZeroIggyDuration::ONE_SECOND,
AutoCommitWhen::ConsumingAllMessages,
))
.create_consumer_group_if_not_exists()
@@ -116,7 +125,7 @@ let mut consumer = client
.build();
```
-The code above will result in creating the consumer that will try to consume the messages in batches of 1000 every 1 millisecond. The auto-commit is set to commit the offset every second or when all the messages are consumed (fetched). The polling strategy is set to `next` which means that the consumer will try to consume the next available message from the partition currently assigned to the consumer group (you can also invoke a regular `consumer()` builder if you do not plan to use the [...]
+The consumer polls at most 1000 messages per request with a minimum 1 ms gap between polls. It requests offset commits every second or when the current message buffer becomes empty; these triggers do not observe application processing. The polling strategy is set to `next` which means that the consumer will try to consume the next available message from the partition currently assigned to the consumer group (you can also invoke a regular `consumer()` builder if you do not plan to use the [...]
Finally, you can use the `next()` method to receive the messages from the topic. Unlike the producer, the consumer **must stay a mutable binding** for its whole life: `init()` takes `&mut self`, and `next()` comes from the futures `Stream` implementation which also requires mutable access. The `init()` is used to ensure that the consumer is ready to receive the messages by validating the existence of the stream, topic, consumer group etc. Here's how you can consume the messages:
@@ -127,7 +136,7 @@ consumer.init().await?;
while let Some(message) = consumer.next().await {
match message {
Ok(received) => {
- // received.message is the IggyMessage
+ println!("{}", String::from_utf8_lossy(&received.message.payload));
}
Err(error) => eprintln!("Error while receiving message: {error}"),
}
@@ -152,14 +161,15 @@ Defaults and lesser-known knobs on `IggyConsumerBuilder`:
| `init_retries(retries, interval)` | disabled | Retry `init()` when the stream or topic does not exist yet (e.g. created dynamically by a producer) |
| `allow_replay()` | disabled | Allow re-consuming messages at or below the stored offset |
| `offset_drain_timeout(d)` | `5s` | How long `shutdown()` waits for background auto-commit tasks to drain before leaving the group |
-| `commit_failed_messages()` | not set | Sets `AutoCommit::Disabled`, so offsets of failed messages are never skipped; store offsets manually after successful processing |
-| `encryptor(...)` | none | Client-side payload decryption |
+| `encryptor(...)` | inherited from the client; none by default | Client-side payload and user-header decryption |
+
+With an encryptor, `init()` rejects the `PollingMessages` auto-commit trigger because a batch might fail decryption after its offset was committed. Use a compatible mode, such as `AutoCommit::Disabled`, and store offsets explicitly after successful processing.
### Auto-commit matrix
-`AutoCommit` decides when the consumer offset is stored on the server:
+`AutoCommit` decides when the consumer requests offset storage:
-| Variant | Offset is stored |
+| Variant | Commit trigger |
|---------|------------------|
| `Disabled` | Never automatically; store it manually |
| `Interval(d)` | Every `d` in the background |
@@ -168,13 +178,15 @@ Defaults and lesser-known knobs on `IggyConsumerBuilder`:
| `When(when)` | At the `when` trigger |
| `After(after)` | At the `after` trigger |
-`AutoCommitWhen` triggers fire while receiving: `PollingMessages`, `ConsumingAllMessages`, `ConsumingEachMessage`, `ConsumingEveryNthMessage(n)`. `AutoCommitAfter` triggers fire after processing: `ConsumingAllMessages`, `ConsumingEachMessage`, `ConsumingEveryNthMessage(n)`.
+`AutoCommitWhen` triggers fire while receiving: `PollingMessages` accompanies the poll request, `ConsumingEachMessage` runs before yielding a message, and `ConsumingAllMessages` runs when the current buffer is empty. `ConsumingEveryNthMessage(n)` checks whether the message offset is divisible by `n`, rather than counting processed messages; `n = 0` disables that trigger.
+
+`AutoCommitAfter` triggers run after the handler returns, including when it returns an error. `ConsumingEachMessage` and `ConsumingEveryNthMessage(n)` use the handled message's offset. `ConsumingAllMessages` fires when that offset reaches the partition head reported by the poll, so a lagging consumer may process several batches before it fires. For commits only after successful processing, disable auto-commit and store offsets explicitly.
-The `After` and `IntervalOrAfter` variants *only* work with `consume_messages()` from the `IggyConsumerMessageExt` trait (see [Stream Builder](/docs/sdk/rust/stream-builder)). The plain `next()` loop cannot observe when your processing finished.
+The `After` triggers require `consume_messages()` from the `IggyConsumerMessageExt` trait (see [Stream Builder](/docs/sdk/rust/stream-builder)). The plain `next()` loop cannot observe when processing finished, though `IntervalOrAfter` still commits on its interval. Background commits can fail; use an explicit offset-store result when completion must be confirmed.
## Raw requests
For custom commands the `IggyClient` exposes two escape hatches below the typed API:
-- `send_binary_request(code, payload)` sends a raw binary command and returns the raw response. Binary transports only. On HTTP it returns `FeatureUnavailable`. Login and logout codes are rejected with `InvalidCommand`, use `login_user`/`logout_user` so the SDK session state stays correct. Custom codes are forwarded to the server, which decides whether it implements them.
+- `send_binary_request(code, payload)` sends a raw binary command and returns the raw response. Binary transports only. On HTTP it returns `FeatureUnavailable`. Login, registration and logout codes are rejected with `InvalidCommand`, use `login_user`/`logout_user` so the SDK session state stays correct. Custom codes are forwarded to the server, which decides whether it implements them.
- `send_http_request(method, path, body)` invokes an arbitrary HTTP endpoint and returns the raw response body. HTTP transport only. Binary transports return `FeatureUnavailable`.
diff --git a/content/docs/sdk/rust/intro.mdx b/content/docs/sdk/rust/intro.mdx
index 20e46f7a..7bb98cd9 100644
--- a/content/docs/sdk/rust/intro.mdx
+++ b/content/docs/sdk/rust/intro.mdx
@@ -5,23 +5,27 @@ description: "The Rust SDK, the primary and most complete Iggy client, and where
The Rust SDK is the primary and most feature-complete client for Iggy. It is available on [crates.io](https://crates.io/crates/iggy) and the source code is part of the [core repository](https://github.com/apache/iggy/tree/master/core/sdk).
+Run from your application crate to add the published SDK:
+
```bash
cargo add iggy
```
+Use an SDK release compatible with your server. For unreleased changes, build both from the same checkout; the [Getting Started](/docs/introduction/getting-started) guide shows the source dependency setup.
+
## Prerequisites
-The examples below log in with the `iggy`/`iggy` root credentials, which **only exist** when the server is started with the default-credentials flag (or `IGGY_ROOT_USERNAME`/`IGGY_ROOT_PASSWORD` set) on its first boot. Otherwise root gets a generated password and the examples fail with `InvalidCredentials`:
+The examples below log in with `iggy`/`iggy`. For a new local development instance, with no `IGGY_ROOT_USERNAME` or `IGGY_ROOT_PASSWORD` overrides, use:
```bash
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-`--fresh` **wipes the local data directory**. This setup is for *development only*.
+`--fresh` **wipes this replica's local data directory**. Environment credentials take precedence over the flag, and bootstrap settings do not replace recovered credentials. In a cluster, a fresh replica can recover credentials from peers. Use disposable development data. A new standalone server without explicit credentials generates a root password; a new cluster requires explicit credentials.
## High-level vs low-level API
-The SDK provides two layers:
+The SDK provides two layers. The snippets below assume an async Tokio context with a `Result` return type.
**High-level API (recommended)** - the easiest way to get started. It handles connection management, auto-batching, consumer group lifecycle, offset commits, retry logic, and reconnection out of the box. Use `IggyClient`, `IggyProducer`, and `IggyConsumer` for the best developer experience.
@@ -29,13 +33,13 @@ The SDK provides two layers:
use iggy::prelude::*;
// Connect with a connection string
-let client = IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+let client = IggyClient::from_connection_string("iggy://iggy:iggy@127.0.0.1:8090")?;
client.connect().await?;
// Or use the builder
let client = IggyClientBuilder::new()
.with_tcp()
- .with_server_address("localhost:8090".to_string())
+ .with_server_address("127.0.0.1:8090".to_string())
.build()?;
client.connect().await?;
client.login_user("iggy", "iggy").await?;
@@ -57,17 +61,19 @@ iggy+http://user:pass@host:port (HTTP)
iggy+ws://user:pass@host:port (WebSocket)
```
+TCP, QUIC, and WebSocket apply connection-string credentials on `connect()`. HTTP requires an explicit login and supports ordinary consumers only.
+
Options can be appended as query parameters:
```
-iggy://iggy:iggy@my-server:8090?tls=true&tls_ca_file=/path/to/ca.crt
-iggy://iggy:iggy@localhost:8090?reconnection_retries=unlimited&heartbeat_interval=5s
+iggy://iggy:iggy@127.0.0.1:8090?nodelay=true&heartbeat_interval=5s
+iggy://iggy:iggy@127.0.0.1:8090?reconnection_retries=unlimited&heartbeat_interval=5s
```
Personal Access Tokens are also supported. Any credential without a colon is treated as a PAT. Paste the token exactly as the server returned it, with **no prefix**:
```
-iggy://<personal-access-token>@localhost:8090
+iggy://<personal-access-token>@127.0.0.1:8090
```
See [Connection Strings](/docs/sdk/connection-strings) for the full per-transport option tables and defaults.
@@ -138,6 +144,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
}
```
-Partition IDs are **0-based**: the first partition of a topic is partition `0`.
+Partition IDs are **0-based**: the first partition of a topic is partition `0`. The consumer uses the default consumer identity and requests auto-commit; its offset can advance before application processing finishes. Use explicit offset storage after successful processing when that distinction matters.
For the full getting started tutorial, see [Getting Started](/docs/introduction/getting-started). For the high-level producer/consumer builders, see [High-level SDK](/docs/sdk/rust/high-level-sdk) and [Stream Builder](/docs/sdk/rust/stream-builder).
diff --git a/content/docs/sdk/rust/stream-builder.mdx b/content/docs/sdk/rust/stream-builder.mdx
index cfcf905f..697b45eb 100644
--- a/content/docs/sdk/rust/stream-builder.mdx
+++ b/content/docs/sdk/rust/stream-builder.mdx
@@ -17,13 +17,15 @@ The stream builder provides a convenient way to create the iggy client, producer
All source code examples are located in the [**examples folder**](https://github.com/apache/iggy/tree/master/examples/rust/src/stream-builder) of the iggy repository. Also,
if you encounter a problem with any of the examples below, please ask in the [**community discord**](https://discord.gg/apache-iggy).
-The examples below connect with the `iggy`/`iggy` root credentials, which **only exist** when the server was started with:
+The examples below use `iggy`/`iggy`. For a new local development instance, with no `IGGY_ROOT_USERNAME` or `IGGY_ROOT_PASSWORD` overrides, use:
```bash
cargo run --bin iggy-server -- --fresh --with-default-root-credentials
```
-Setting `IGGY_ROOT_USERNAME`/`IGGY_ROOT_PASSWORD` works too. Otherwise root gets a generated password and the examples fail with `InvalidCredentials`.
+Environment credentials take precedence over the flag, and bootstrap settings do not replace recovered credentials. `--fresh` deletes this replica's local data; use disposable development data. A new standalone server without explicit credentials generates a root password; a new cluster requires explicit credentials.
+
+Use the SDK and server from the same checkout for unreleased changes. The examples import `PrintEventConsumer` from the local `iggy_examples` crate and can run as the targets in `examples/rust`. In your own application, use the implementation shown below instead of that import.
## IggyStream Builder
@@ -35,7 +37,7 @@ use iggy_examples::shared::stream::PrintEventConsumer;
use std::str::FromStr;
use tokio::sync::oneshot;
-const IGGY_URL: &str = "iggy://iggy:iggy@localhost:8090";
+const IGGY_URL: &str = "iggy://iggy:iggy@127.0.0.1:8090";
#[tokio::main]
async fn main() -> Result<(), IggyError> {
@@ -110,7 +112,7 @@ When you implement the producer side, you can use the `IggyStreamProducer` to ge
use iggy::prelude::*;
use std::str::FromStr;
-const IGGY_URL: &str = "iggy://iggy:iggy@localhost:8090";
+const IGGY_URL: &str = "iggy://iggy:iggy@127.0.0.1:8090";
#[tokio::main]
async fn main() -> Result<(), IggyError> {
@@ -153,7 +155,7 @@ repository.
The IggyProducerConfig gives you a way to configure the producer in sufficient detail. Please note, if you have
questions about any of those settings, please ask in the community discord. For basic customization, the `from_stream_topic` constructor
-lets you set a custom stream and topic name as well as the maximum batch length and linger time between sends.
+lets you set a custom stream and topic name as well as the maximum batch length and linger time between sends. The producer resolves destinations by name; keep the corresponding identifier fields consistent with those names.
```rust
use iggy::prelude::*;
@@ -209,8 +211,8 @@ async fn main() -> Result<(), IggyError> {
// - `PartitionId` - the partition ID is provided by the client.
// - `MessagesKey` - the client hashes the provided messages key
// to a partition ID.
- // The partition count is cached per topic and not refreshed
- // while the client stays connected.
+ // Binary clients cache the partition count for their lifetime;
+ // reconnecting does not refresh it.
.partitioning(Partitioning::balanced())
// Sets the retry policy (maximum number of retries and interval)
// in case of messages sending failure.
@@ -218,12 +220,12 @@ async fn main() -> Result<(), IggyError> {
// or to the server rejecting the messages.
// Default is 3 retries with 1 second interval between them.
.send_retries_count(3)
- .send_retries_interval(IggyDuration::new_from_secs(1))
+ .send_retries_interval(NonZeroIggyDuration::ONE_SECOND)
// Optionally, set a custom client side encryptor for encrypting
// the messages' payloads. Currently only Aes256Gcm is supported.
// Note, this is independent of server side encryption.
// You can add client encryption, server encryption, or both.
- // .encryptor(Arc::new(EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(&[1; 32])?)))
+ // .encryptor(std::sync::Arc::new(EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(&[1; 32])?)))
.build();
Ok(())
}
@@ -245,7 +247,7 @@ use iggy::prelude::*;
use iggy_examples::shared::stream::PrintEventConsumer;
use tokio::sync::oneshot;
-const IGGY_URL: &str = "iggy://iggy:iggy@localhost:8090";
+const IGGY_URL: &str = "iggy://iggy:iggy@127.0.0.1:8090";
#[tokio::main]
async fn main() -> Result<(), IggyError> {
@@ -322,27 +324,27 @@ async fn main() -> Result<(), IggyError> {
.create_stream_if_not_exists(true)
// Create the topic if it doesn't exist.
.create_topic_if_not_exists(true)
- // The name of the consumer. Must be unique.
+ // Members of the same consumer group use the same name.
.consumer_name("test_consumer".to_string())
// The type of consumer. It can be either `Consumer` or `ConsumerGroup`.
// ConsumerGroup is default.
.consumer_kind(ConsumerKind::ConsumerGroup)
- // Sets the number of partitions for ConsumerKind `Consumer`.
- // Does not apply to `ConsumerGroup`.
+ // Partition count when creating a topic. Also used as the partition
+ // ID for an ordinary consumer; group assignment ignores this value.
.partitions_count(1)
// The polling interval for messages.
.polling_interval(IggyDuration::from_str("5ms").unwrap())
// `PollingStrategy` specifies from where to start polling messages.
// It has the following kinds:
// - `Offset` - start polling from the specified offset.
- // - `Timestamp` - start polling from the specified timestamp.
+ // - `Timestamp` - start at or after the broker append timestamp.
// - `First` - start polling from the first message in the partition.
- // - `Last` - start polling from the last message in the partition.
+ // - `Last` - read up to batch_length messages ending at the committed offset.
// - `Next` - start polling from the next message after the
// last polled message based on the stored consumer offset.
.polling_strategy(PollingStrategy::last())
// Sets the polling retry interval in case of server disconnection.
- .polling_retry_interval(IggyDuration::new_from_secs(1))
+ .polling_retry_interval(NonZeroIggyDuration::ONE_SECOND)
// Sets the number of retries and the interval when initializing
// the consumer if the stream or topic is not found.
// Useful when the stream or topic is created dynamically
@@ -350,20 +352,22 @@ async fn main() -> Result<(), IggyError> {
// The default is 5 retries with a 3 second interval,
// so init retry is enabled out of the box.
.init_retries(5)
- .init_interval(IggyDuration::new_from_secs(1))
- // Optionally, set a custom client side encryptor for encrypting
- // the messages' payloads. Currently only Aes256Gcm is supported.
+ .init_interval(NonZeroIggyDuration::ONE_SECOND)
+ // Optionally, configure Aes256Gcm payload and user-header decryption.
+ // Replace PollingMessages auto-commit with a compatible mode first.
// Key must be identical to the one used by the producer;
// thus ensure secure key exchange.
// Note, this is independent of server side encryption.
// you can add client encryption, server encryption, or both.
- // .encryptor(Arc::new(EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(&[1; 32])?)))
+ // .encryptor(std::sync::Arc::new(EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(&[1; 32])?)))
.build();
Ok(())
}
```
+`partitions_count` also selects the partition ID for an ordinary consumer. When it creates a topic with `n` partitions, the valid IDs are `0..n-1`, so using the same `n` as an ordinary consumer's ID fails. The sample uses a consumer group. For an ordinary consumer, create the topic separately and use `client.consumer(name, stream, topic, partition_id)` to select a valid partition.
+
## Add consumers dynamically at runtime.
When you create consumers on demand at application runtime, the consumer has to survive the window where the stream
@@ -376,13 +380,11 @@ you have to set `create_stream_if_not_exists` and `create_topic_if_not_exists` t
the `IggyStreamConsumer` constructors as before i.e.:
```rust
- let config = get_my_custom_iggy_consumer_config();
let (client, mut consumer) =
IggyStreamConsumer::with_client_from_url(IGGY_URL, &config).await?;
```
-Where `get_my_custom_iggy_consumer_config` refers to a function that returns an `IggyConsumerConfig`
-that specifies the stream and topic to consume as well the init retry or whether to create the stream and topic.
+Here `config` is the `IggyConsumerConfig` built above, which selects the stream and topic, initialization retries, and automatic creation behavior.
## Add producers dynamically at runtime.