You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/08/17 18:37:00 UTC
(iggy) branch master updated: feat(csharp): add heartbeat mechanism (#3894)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 1890c35f3 feat(csharp): add heartbeat mechanism (#3894)
1890c35f3 is described below
commit 1890c35f3a0cdd7766d68eef0fc297d33941f505
Author: Łukasz Zborek <lu...@gmail.com>
AuthorDate: Mon Aug 17 20:36:55 2026 +0200
feat(csharp): add heartbeat mechanism (#3894)
- TCP client heartbeat: PeriodicTimer loop pings server every
HeartbeatInterval (default 5s, new option on IggyClientConfigurator).
Ping goes through normal request path, so reconnect + auto login repair
dead session. Skips ping when connecting or when reconnect/auto-login
disabled and connection dead. Started under connect gate, no duplicate
loops.
- AutoLoginSettings: init-only, PersonalAccessToken support, public
For(user, pwd) / ForPersonalAccessToken(token) factories.
- Consumer/Publisher builders: heartbeat + auto-login wiring, warning
when reconnect enabled without auto login.
- Enable reconnection by default
---
core/server/config.toml | 5 +-
.../Fixtures/IggyServerFixture.cs | 18 +-
.../Iggy_SDK.Tests.Integration/HeartbeatTests.cs | 163 +++++++++++++++++
.../Iggy_SDK/Configuration/AutoLoginSettings.cs | 40 +++-
.../Configuration/IggyClientConfigurator.cs | 9 +
.../Iggy_SDK/Configuration/ReconnectionSettings.cs | 8 +-
.../Iggy_SDK/Consumers/IggyConsumerBuilder.cs | 56 +++++-
.../Iggy_SDK/Consumers/IggyConsumerBuilderOfT.cs | 3 +-
.../Iggy_SDK/Consumers/IggyConsumerConfig.cs | 15 +-
.../csharp/Iggy_SDK/Factory/IggyClientFactory.cs | 9 +
.../IggyClient/Implementations/TcpMessageStream.cs | 203 ++++++++++++++++++---
.../Iggy_SDK/Publishers/IggyPublisherBuilder.cs | 56 +++++-
.../Iggy_SDK/Publishers/IggyPublisherBuilderOfT.cs | 3 +-
.../Iggy_SDK/Publishers/IggyPublisherConfig.cs | 15 +-
.../ClientTests/IggyClientFactoryTests.cs | 46 +++++
.../ConsumerTests/IggyConsumerBuilderTests.cs | 22 +++
foreign/csharp/README.md | 62 ++++---
17 files changed, 644 insertions(+), 89 deletions(-)
diff --git a/core/server/config.toml b/core/server/config.toml
index 9168ccf26..5f37e36c3 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -313,10 +313,7 @@ interval = "1 m"
# rebalance off it. Only a connection that still holds a group
# membership is evicted; one that holds none is left alone and is
# reaped when its socket closes.
-# TODO(hubcio): flip the default to true once every SDK heartbeats
-# automatically; today the blocking Java and C# SDKs never ping, so an
-# enabled default would evict their idle consumer-group members.
-enabled = false
+enabled = true
# Interval for expected client heartbeats. The Rust, Go, Python, Node and
# async Java SDKs ping automatically every 5 s, well inside the resulting 36 s
# staleness window, but only from a connected high-level client: the Rust and
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs
index 1670cbcdf..4d3138eab 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs
@@ -56,11 +56,24 @@ public class IggyServerFixture : IAsyncInitializer, IAsyncDisposable
/// </summary>
protected bool EnabledServerTraceLogs => true;
+ /// <summary>
+ /// Server-side heartbeat interval. Short so a client that stops pinging is evicted within a few seconds
+ /// instead of the default five. Clients from <see cref="CreateClient" /> ping at half this interval;
+ /// any client built by hand must set <see cref="IggyClientConfigurator.HeartbeatInterval" /> below it or
+ /// be evicted once idle past 1.2 intervals.
+ /// </summary>
+ public static readonly TimeSpan ServerHeartbeatInterval = TimeSpan.FromSeconds(2);
+
/// <summary>
/// Extra environment variables for every node, layered over the cluster's base configuration.
- /// Override in subclasses to customize.
+ /// Heartbeat verification is on for every fixture so the eviction tests share the general server
+ /// instead of starting their own. Override in subclasses to customize.
/// </summary>
- protected virtual Dictionary<string, string> EnvironmentVariables => [];
+ protected virtual Dictionary<string, string> EnvironmentVariables => new()
+ {
+ { "IGGY_HEARTBEAT_ENABLED", "true" },
+ { "IGGY_HEARTBEAT_INTERVAL", $"{ServerHeartbeatInterval.TotalSeconds}s" }
+ };
/// <summary>
/// Resource mappings (certificates, etc.) mounted into every node. Override in subclasses.
@@ -131,6 +144,7 @@ public class IggyServerFixture : IAsyncInitializer, IAsyncDisposable
BaseAddress = address ?? await GetIggyAddressAsync(protocol),
Protocol = protocol,
ReconnectionSettings = new ReconnectionSettings { Enabled = true },
+ HeartbeatInterval = ServerHeartbeatInterval / 2,
AutoLoginSettings = new AutoLoginSettings
{
Enabled = autoLogin,
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/HeartbeatTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/HeartbeatTests.cs
new file mode 100644
index 000000000..7a9de7fb6
--- /dev/null
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/HeartbeatTests.cs
@@ -0,0 +1,163 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+using Apache.Iggy.Configuration;
+using Apache.Iggy.Enums;
+using Apache.Iggy.Exceptions;
+using Apache.Iggy.Factory;
+using Apache.Iggy.IggyClient;
+using Apache.Iggy.Tests.Integrations.Fixtures;
+using Shouldly;
+
+namespace Apache.Iggy.Tests.Integrations;
+
+/// <summary>
+/// Runs against a server that evicts consumer-group members it has not heard from. Only a group member is
+/// evicted, so every test joins a group before going idle.
+/// </summary>
+public class HeartbeatTests
+{
+ private const string TopicName = "heartbeat-topic";
+ private const string GroupName = "heartbeat-group";
+
+ // Comfortably past the server's stale threshold of 1.2 intervals, with room for the verifier's own tick.
+ private static readonly TimeSpan IdleFor = IggyServerFixture.ServerHeartbeatInterval * 3;
+
+ [ClassDataSource<IggyServerFixture>(Shared = SharedType.PerAssembly)]
+ public required IggyServerFixture Fixture { get; init; }
+
+ [Test]
+ public async Task IdleGroupMember_WithHeartbeat_Should_StayMember()
+ {
+ using var client = await CreateClient(TimeSpan.FromSeconds(1));
+ var (streamName, groupId) = await JoinFreshGroup(client);
+
+ await Task.Delay(IdleFor);
+
+ using var observer = await Fixture.CreateAuthenticatedClient(Protocol.Tcp);
+ var group = await observer.GetConsumerGroupByIdAsync(Identifier.String(streamName),
+ Identifier.String(TopicName), Identifier.Numeric(groupId));
+ group.ShouldNotBeNull();
+ group.MembersCount.ShouldBe(1u);
+
+ var me = await client.GetMeAsync();
+ me.ShouldNotBeNull();
+ me.ConsumerGroupsCount.ShouldBe(1);
+ }
+
+ [Test]
+ public async Task IdleGroupMember_WithSlowHeartbeat_Should_BeEvicted_And_Reconnect()
+ {
+ // Heartbeat cannot be turned off, so a ping interval far past the server's threshold stands in for one.
+ using var client = await CreateClient(TimeSpan.FromHours(1));
+ var (streamName, groupId) = await JoinFreshGroup(client);
+
+ await Task.Delay(IdleFor);
+
+ using var observer = await Fixture.CreateAuthenticatedClient(Protocol.Tcp);
+ var group = await observer.GetConsumerGroupByIdAsync(Identifier.String(streamName),
+ Identifier.String(TopicName), Identifier.Numeric(groupId));
+ group.ShouldNotBeNull();
+ group.MembersCount.ShouldBe(0u);
+
+ // The eviction reaches the client with its next request, which the default reconnection replays over a
+ // fresh, auto-logged-in session.
+ var me = await client.GetMeAsync();
+ me.ShouldNotBeNull();
+ me.ConsumerGroupsCount.ShouldBe(0);
+ }
+
+ [Test]
+ public async Task EvictedClient_WithPersonalAccessTokenAutoLogin_Should_Reconnect()
+ {
+ using var issuer = await Fixture.CreateAuthenticatedClient(Protocol.Tcp);
+ var pat = await issuer.CreatePersonalAccessTokenAsync($"heartbeat-{Guid.NewGuid():N}",
+ TimeSpan.FromHours(1));
+
+ using var client = await CreateClient(TimeSpan.FromHours(1),
+ AutoLoginSettings.ForPersonalAccessToken(pat!.Token));
+ var (streamName, groupId) = await JoinFreshGroup(client);
+
+ await Task.Delay(IdleFor);
+
+ var group = await issuer.GetConsumerGroupByIdAsync(Identifier.String(streamName),
+ Identifier.String(TopicName), Identifier.Numeric(groupId));
+ group.ShouldNotBeNull();
+ group.MembersCount.ShouldBe(0u);
+
+ var me = await client.GetMeAsync();
+ me.ShouldNotBeNull();
+ me.ConsumerGroupsCount.ShouldBe(0);
+ }
+
+ [Test]
+ public async Task EvictedClient_WithoutAutoLogin_Should_FailFast_And_NotReconnect()
+ {
+ using var client = await CreateClient(TimeSpan.FromHours(1), false);
+ await client.LoginUserAsync("iggy", "iggy");
+ var (streamName, _) = await JoinFreshGroup(client);
+
+ var reconnected = false;
+ client.SubscribeConnectionEvents(args =>
+ {
+ reconnected |= args.CurrentState == ConnectionState.Connecting;
+ return Task.CompletedTask;
+ });
+
+ await Task.Delay(IdleFor);
+
+ // A reconnect could not bring the session back, so the request surfaces the loss instead of coming back
+ // over an unauthenticated connection.
+ await Should.ThrowAsync<Exception>(() => client.GetMeAsync());
+ reconnected.ShouldBeFalse();
+ await Should.ThrowAsync<NotConnectedException>(() =>
+ client.GetStreamByIdAsync(Identifier.String(streamName)));
+ }
+
+ private Task<IIggyClient> CreateClient(TimeSpan heartbeatInterval, bool autoLogin = true)
+ {
+ return CreateClient(heartbeatInterval,
+ autoLogin ? AutoLoginSettings.For("iggy", "iggy") : new AutoLoginSettings());
+ }
+
+ private async Task<IIggyClient> CreateClient(TimeSpan heartbeatInterval, AutoLoginSettings autoLogin)
+ {
+ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
+ {
+ BaseAddress = await Fixture.GetIggyAddressAsync(Protocol.Tcp),
+ Protocol = Protocol.Tcp,
+ HeartbeatInterval = heartbeatInterval,
+ AutoLoginSettings = autoLogin
+ });
+ await client.ConnectAsync();
+
+ return client;
+ }
+
+ private static async Task<(string streamName, uint groupId)> JoinFreshGroup(IIggyClient client)
+ {
+ var streamName = $"heartbeat-stream-{Guid.NewGuid():N}";
+ await client.CreateStreamAsync(streamName);
+ await client.CreateTopicAsync(Identifier.String(streamName), TopicName, 2);
+ var group = await client.CreateConsumerGroupAsync(Identifier.String(streamName),
+ Identifier.String(TopicName), GroupName);
+ await client.JoinConsumerGroupAsync(Identifier.String(streamName), Identifier.String(TopicName),
+ Identifier.Numeric(group!.Id));
+
+ return (streamName, group.Id);
+ }
+}
diff --git a/foreign/csharp/Iggy_SDK/Configuration/AutoLoginSettings.cs b/foreign/csharp/Iggy_SDK/Configuration/AutoLoginSettings.cs
index 3c88446ec..cdf020bf8 100644
--- a/foreign/csharp/Iggy_SDK/Configuration/AutoLoginSettings.cs
+++ b/foreign/csharp/Iggy_SDK/Configuration/AutoLoginSettings.cs
@@ -23,32 +23,54 @@ namespace Apache.Iggy.Configuration;
public class AutoLoginSettings
{
/// <summary>
- /// Enable automatic login on connection establishment
+ /// Enable automatic login on connection establishment.
/// </summary>
- public bool Enabled { get; set; }
+ public bool Enabled { get; init; }
/// <summary>
/// Specifies the username for auto-login configuration.
/// </summary>
- public string Username { get; set; } = string.Empty;
+ public string Username { get; init; } = string.Empty;
/// <summary>
/// Specifies the password for auto-login authentication
/// </summary>
- public string Password { get; set; } = string.Empty;
+ public string Password { get; init; } = string.Empty;
/// <summary>
- /// Settings for a builder-owned client that signs in with the given credentials. The credentials must
- /// reach the client and not only the explicit login the wrapper performs at startup: a reconnect or a
- /// leader redirect drops the session, and without them the client comes back unauthenticated.
+ /// Personal access token to sign in with instead of a username and password. Takes precedence over them
+ /// when set.
/// </summary>
- internal static AutoLoginSettings For(string username, string password)
+ public string PersonalAccessToken { get; init; } = string.Empty;
+
+ /// <summary>
+ /// Auto login with a username and password.
+ /// </summary>
+ /// <exception cref="ArgumentException">Thrown when the username is empty.</exception>
+ public static AutoLoginSettings For(string username, string password)
{
+ ArgumentException.ThrowIfNullOrEmpty(username);
+
return new AutoLoginSettings
{
- Enabled = !string.IsNullOrEmpty(username),
+ Enabled = true,
Username = username,
Password = password
};
}
+
+ /// <summary>
+ /// Auto login with a personal access token.
+ /// </summary>
+ /// <exception cref="ArgumentException">Thrown when the token is empty.</exception>
+ public static AutoLoginSettings ForPersonalAccessToken(string token)
+ {
+ ArgumentException.ThrowIfNullOrEmpty(token);
+
+ return new AutoLoginSettings
+ {
+ Enabled = true,
+ PersonalAccessToken = token
+ };
+ }
}
diff --git a/foreign/csharp/Iggy_SDK/Configuration/IggyClientConfigurator.cs b/foreign/csharp/Iggy_SDK/Configuration/IggyClientConfigurator.cs
index fad8604a6..48e50d3b7 100644
--- a/foreign/csharp/Iggy_SDK/Configuration/IggyClientConfigurator.cs
+++ b/foreign/csharp/Iggy_SDK/Configuration/IggyClientConfigurator.cs
@@ -53,6 +53,15 @@ public sealed class IggyClientConfigurator
/// </summary>
public int SendBufferSize { get; set; } = 4096;
+ /// <summary>
+ /// Interval between the pings the <see cref="Enums.Protocol.Tcp" /> client sends on its own while
+ /// connected, so an idle session survives the server's heartbeat verification and consumer-group
+ /// assignments stay fresh. Default is 5 seconds; must be between 1 millisecond and about 49 days.
+ /// Init-only: the client validates the interval once and hands it to a timer that would fault on an
+ /// out-of-range value.
+ /// </summary>
+ public TimeSpan HeartbeatInterval { get; init; } = TimeSpan.FromSeconds(5);
+
/// <summary>
/// TLS settings
/// </summary>
diff --git a/foreign/csharp/Iggy_SDK/Configuration/ReconnectionSettings.cs b/foreign/csharp/Iggy_SDK/Configuration/ReconnectionSettings.cs
index 01986bd3c..f8cf495c4 100644
--- a/foreign/csharp/Iggy_SDK/Configuration/ReconnectionSettings.cs
+++ b/foreign/csharp/Iggy_SDK/Configuration/ReconnectionSettings.cs
@@ -23,14 +23,14 @@ namespace Apache.Iggy.Configuration;
public sealed class ReconnectionSettings
{
/// <summary>
- /// Enable automatic reconnection when connection is lost
+ /// Enable automatic reconnection when connection is lost. Enabled by default.
/// </summary>
- public bool Enabled { get; set; } = false;
+ public bool Enabled { get; set; } = true;
/// <summary>
- /// Maximum number of reconnection attempts (0 = infinite)
+ /// Maximum number of reconnection attempts (0 = infinite). Default is 0.
/// </summary>
- public int MaxRetries { get; set; } = 3;
+ public int MaxRetries { get; set; }
/// <summary>
/// Initial delay before first reconnection attempt
diff --git a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerBuilder.cs b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerBuilder.cs
index fc768662b..400cc7234 100644
--- a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerBuilder.cs
+++ b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerBuilder.cs
@@ -108,6 +108,29 @@ public class IggyConsumerBuilder
return this;
}
+ /// <summary>
+ /// Configures the connection settings using a personal access token instead of a username and password.
+ /// </summary>
+ /// <param name="protocol">The protocol to use for the connection (e.g., TCP, UDP).</param>
+ /// <param name="address">The address of the server to connect to.</param>
+ /// <param name="personalAccessToken">The personal access token to authenticate with.</param>
+ /// <param name="receiveBufferSize">The size of the receive buffer.</param>
+ /// <param name="sendBufferSize">The size of the send buffer.</param>
+ /// <param name="reconnectionSettings">Reconnection settings for the client.</param>
+ /// <returns>The current instance of <see cref="IggyConsumerBuilder" /> to allow method chaining.</returns>
+ public IggyConsumerBuilder WithConnection(Protocol protocol, string address, string personalAccessToken,
+ int receiveBufferSize = 4096, int sendBufferSize = 4096, ReconnectionSettings? reconnectionSettings = null)
+ {
+ Config.Protocol = protocol;
+ Config.Address = address;
+ Config.PersonalAccessToken = personalAccessToken;
+ Config.ReceiveBufferSize = receiveBufferSize;
+ Config.SendBufferSize = sendBufferSize;
+ Config.ReconnectionSettings = reconnectionSettings;
+
+ return this;
+ }
+
/// <summary>
/// Configures message encryption on the client this builder creates. Encryption is a client-level concern:
/// the encryptor decrypts on poll (and encrypts on send) for the whole connection. Only valid when the
@@ -245,7 +268,8 @@ public class IggyConsumerBuilder
ReceiveBufferSize = Config.ReceiveBufferSize,
SendBufferSize = Config.SendBufferSize,
ReconnectionSettings = Config.ReconnectionSettings ?? new ReconnectionSettings(),
- AutoLoginSettings = AutoLoginSettings.For(Config.Login, Config.Password),
+ HeartbeatInterval = Config.HeartbeatInterval,
+ AutoLoginSettings = AutoLogin(),
LoggerFactory = Config.LoggerFactory ?? NullLoggerFactory.Instance,
MessageEncryptor = _encryptor
});
@@ -262,6 +286,18 @@ public class IggyConsumerBuilder
return consumer;
}
+ /// <summary>
+ /// The credentials given to WithConnection must reach the client and not only the explicit login
+ /// performed at startup: a reconnect or a leader redirect drops the session, and without them the
+ /// client would come back unauthenticated.
+ /// </summary>
+ private protected AutoLoginSettings AutoLogin()
+ {
+ return string.IsNullOrEmpty(Config.PersonalAccessToken)
+ ? AutoLoginSettings.For(Config.Login, Config.Password)
+ : AutoLoginSettings.ForPersonalAccessToken(Config.PersonalAccessToken);
+ }
+
/// <summary>
/// Validates the consumer configuration and throws if invalid.
/// </summary>
@@ -275,14 +311,18 @@ public class IggyConsumerBuilder
throw new InvalidOperationException("Address must be provided when CreateIggyClient is true.");
}
- if (string.IsNullOrWhiteSpace(Config.Login))
- {
- throw new InvalidOperationException("Login must be provided when CreateIggyClient is true.");
- }
-
- if (string.IsNullOrWhiteSpace(Config.Password))
+ if (string.IsNullOrWhiteSpace(Config.PersonalAccessToken))
{
- throw new InvalidOperationException("Password must be provided when CreateIggyClient is true.");
+ if (string.IsNullOrWhiteSpace(Config.Login))
+ {
+ throw new InvalidOperationException(
+ "Login or PersonalAccessToken must be provided when CreateIggyClient is true.");
+ }
+
+ if (string.IsNullOrWhiteSpace(Config.Password))
+ {
+ throw new InvalidOperationException("Password must be provided when CreateIggyClient is true.");
+ }
}
}
else
diff --git a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerBuilderOfT.cs b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerBuilderOfT.cs
index ad640ebef..3800f72f6 100644
--- a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerBuilderOfT.cs
+++ b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerBuilderOfT.cs
@@ -96,7 +96,8 @@ public class IggyConsumerBuilder<T> : IggyConsumerBuilder
ReceiveBufferSize = Config.ReceiveBufferSize,
SendBufferSize = Config.SendBufferSize,
ReconnectionSettings = Config.ReconnectionSettings ?? new ReconnectionSettings(),
- AutoLoginSettings = AutoLoginSettings.For(Config.Login, Config.Password),
+ HeartbeatInterval = Config.HeartbeatInterval,
+ AutoLoginSettings = AutoLogin(),
LoggerFactory = Config.LoggerFactory ?? NullLoggerFactory.Instance,
MessageEncryptor = _encryptor
});
diff --git a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerConfig.cs b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerConfig.cs
index 1418282d3..d81430110 100644
--- a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerConfig.cs
+++ b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerConfig.cs
@@ -69,6 +69,12 @@ public class IggyConsumerConfig
/// </summary>
public string Password { get; set; } = string.Empty;
+ /// <summary>
+ /// Personal access token used instead of <see cref="Login" /> and <see cref="Password" /> when creating
+ /// the client. Takes precedence over them when set.
+ /// </summary>
+ public string PersonalAccessToken { get; set; } = string.Empty;
+
/// <summary>
/// The size of the receive buffer in bytes. Default is 4096.
/// </summary>
@@ -148,7 +154,14 @@ public class IggyConsumerConfig
/// <summary>
/// Gets or sets the reconnection settings to control the behavior of the iggy client
/// in case of a disconnect or network failure.
- /// This property is optional and can be null. If null, reconnection will be disabled.
+ /// This property is optional and can be null. If null, the default <see cref="Configuration.ReconnectionSettings" />
+ /// apply.
/// </summary>
public ReconnectionSettings? ReconnectionSettings { get; set; }
+
+ /// <summary>
+ /// Interval between the pings the created client sends to keep an idle session alive.
+ /// See <see cref="IggyClientConfigurator.HeartbeatInterval" />.
+ /// </summary>
+ public TimeSpan HeartbeatInterval { get; set; } = TimeSpan.FromSeconds(5);
}
diff --git a/foreign/csharp/Iggy_SDK/Factory/IggyClientFactory.cs b/foreign/csharp/Iggy_SDK/Factory/IggyClientFactory.cs
index ef965f34b..515a06dd8 100644
--- a/foreign/csharp/Iggy_SDK/Factory/IggyClientFactory.cs
+++ b/foreign/csharp/Iggy_SDK/Factory/IggyClientFactory.cs
@@ -67,6 +67,15 @@ public static class IggyClientFactory
throw new ArgumentOutOfRangeException(nameof(options), options.MaxResponseFrameSize,
$"MaxResponseFrameSize must be at least {VsrHeader.HEADER_SIZE} bytes.");
}
+
+ // The bounds PeriodicTimer accepts; anything outside them would fault the heartbeat task at start
+ // instead of failing the caller here.
+ if (options.HeartbeatInterval < TimeSpan.FromMilliseconds(1) ||
+ options.HeartbeatInterval > TimeSpan.FromMilliseconds(uint.MaxValue - 1))
+ {
+ throw new ArgumentOutOfRangeException(nameof(options), options.HeartbeatInterval,
+ "HeartbeatInterval must be between 1 millisecond and about 49 days.");
+ }
}
private static IIggyClient CreateIggyTcpClient(IggyClientConfigurator options)
diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
index 75189e84c..32ea6e03e 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
@@ -50,6 +50,9 @@ public sealed partial class TcpMessageStream : IIggyClient
{
private const int InvalidCommandStatus = 3;
+ // Ping carries no body, so the frame is constant and shared by the heartbeat and PingAsync.
+ private static readonly byte[] PingPayload = CreatePingPayload();
+
private static readonly HashSet<uint> SessionControlCodes =
[
CommandCodes.LOGIN_USER_CODE,
@@ -63,6 +66,7 @@ public sealed partial class TcpMessageStream : IIggyClient
private readonly SemaphoreSlim _connectGate = new(1, 1);
private readonly EventAggregator<ConnectionStateChangedEventArgs> _connectionEvents;
private readonly SemaphoreSlim _connectionSemaphore;
+ private readonly CancellationTokenSource _heartbeatCancellation = new();
private readonly ILogger<TcpMessageStream> _logger;
private readonly SemaphoreSlim _sendingSemaphore;
private string _currentAddress = string.Empty;
@@ -74,6 +78,7 @@ public sealed partial class TcpMessageStream : IIggyClient
private string _currentRemoteAddress = string.Empty;
private X509Certificate2Collection _customCaStore = [];
private volatile bool _disposed;
+ private Task? _heartbeatTask;
private int _isConnecting;
private DateTimeOffset _lastConnectionTime;
private int _leaderRedirectCount;
@@ -99,18 +104,26 @@ public sealed partial class TcpMessageStream : IIggyClient
}
/// <summary>
- /// Closes the underlying stream and releases the connection's semaphores.
+ /// Closes the underlying stream and stops the heartbeat. The semaphores are left alone on purpose: a
+ /// heartbeat or request still in flight releases them from its finally block, and a disposed
+ /// <see cref="SemaphoreSlim" /> turns that release into an <see cref="ObjectDisposedException" /> that
+ /// replaces the real error and skips the connection drop. They own no unmanaged handle, so there is
+ /// nothing to leak. The heartbeat token source is kept undisposed for the same reason: the loop and
+ /// <see cref="StartHeartbeat" /> read its token after this method may already have run.
/// </summary>
public void Dispose()
{
+ if (_disposed)
+ {
+ return;
+ }
+
_disposed = true;
+ _heartbeatCancellation.Cancel();
_stream?.Close();
_stream?.Dispose();
SetConnectionStateAsync(ConnectionState.Disconnected);
- _sendingSemaphore.Dispose();
- _connectionSemaphore.Dispose();
- _connectGate.Dispose();
_connectionEvents.Clear();
}
@@ -618,12 +631,7 @@ public sealed partial class TcpMessageStream : IIggyClient
/// <inheritdoc />
public async Task PingAsync(CancellationToken token = default)
{
- var message = Array.Empty<byte>();
- var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length];
- TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.PING_CODE);
-
- await SendAckAsync(payload, token);
-
+ await SendAckAsync(PingPayload, token);
await RefreshGroupAssignmentsAsync(token);
}
@@ -660,6 +668,12 @@ public sealed partial class TcpMessageStream : IIggyClient
/// <inheritdoc />
public Task ConnectAsync(CancellationToken token = default)
{
+ if (_configuration.ReconnectionSettings.Enabled && !_configuration.AutoLoginSettings.Enabled)
+ {
+ _logger.LogWarning(
+ "Reconnection is enabled without auto login: a lost session cannot be restored, requests will fail until the client logs in again");
+ }
+
return ConnectAsync(true, token);
}
@@ -898,6 +912,8 @@ public sealed partial class TcpMessageStream : IIggyClient
Interlocked.Exchange(ref _isConnecting, 1);
try
{
+ ObjectDisposedException.ThrowIf(_disposed, this);
+
if (_state is ConnectionState.Connected
or ConnectionState.Authenticating
or ConnectionState.Authenticated)
@@ -912,6 +928,18 @@ public sealed partial class TcpMessageStream : IIggyClient
SetConnectionStateAsync(ConnectionState.Connecting);
await TryEstablishConnectionAsync(autoLogin, token);
+
+ // Dispose is synchronous and cannot take the sending semaphore, so a Dispose that ran while this
+ // connect was dialing already read a stream that did not exist yet. Reading the flag after the
+ // stream is installed leaves no window: either Dispose sees the new stream and closes it, or it is
+ // seen here and the stream is dropped.
+ if (_disposed)
+ {
+ await DropStreamAsync();
+ throw new ObjectDisposedException(nameof(TcpMessageStream));
+ }
+
+ StartHeartbeat();
}
finally
{
@@ -920,6 +948,67 @@ public sealed partial class TcpMessageStream : IIggyClient
}
}
+ /// <summary>
+ /// Starts the idle ping loop. Called under <see cref="_connectGate" />, so a reconnect finds the loop
+ /// already running and does not start a second one. A loop that ended anyway - an interval the caller
+ /// mutated out of range faults <see cref="PeriodicTimer" />'s constructor - is started again rather than
+ /// leaving the client without a heartbeat for the rest of its life.
+ /// </summary>
+ private void StartHeartbeat()
+ {
+ if (_disposed)
+ {
+ return;
+ }
+
+ if (_heartbeatTask is null or { IsCompleted: true })
+ {
+ _heartbeatTask = RunHeartbeatAsync(_configuration.HeartbeatInterval, _heartbeatCancellation.Token);
+ }
+ }
+
+ /// <summary>
+ /// Pings on a timer so the server keeps hearing from an otherwise idle connection. The ping goes through
+ /// the regular request path, so a lost session is repaired by the same reconnect and auto login every
+ /// other request would trigger; the request deadline bounds a stalled server. Failures are logged and
+ /// the loop keeps going until the client is disposed.
+ /// </summary>
+ private async Task RunHeartbeatAsync(TimeSpan interval, CancellationToken token)
+ {
+ try
+ {
+ using var timer = new PeriodicTimer(interval);
+ while (await timer.WaitForNextTickAsync(token))
+ {
+ // Without reconnection and auto login a ping on a dead connection can only fail; with them,
+ // the ping is what brings an idle client back.
+ var unrecoverable = _state is ConnectionState.Disconnected or ConnectionState.Connecting
+ && !(_configuration.ReconnectionSettings.Enabled
+ && _configuration.AutoLoginSettings.Enabled);
+ if (IsConnecting || unrecoverable)
+ {
+ continue;
+ }
+
+ try
+ {
+ await PingAsync(token);
+ }
+ catch (Exception e) when (!token.IsCancellationRequested && !_disposed)
+ {
+ _logger.LogWarning(e, "Heartbeat failed");
+ }
+ }
+ }
+ catch (Exception) when (token.IsCancellationRequested || _disposed)
+ {
+ }
+ catch (Exception e)
+ {
+ _logger.LogError(e, "Heartbeat stopped");
+ }
+ }
+
private async Task<PolledMessagesRental> PollPartitionMessagesRentedAsync(Identifier streamId, Identifier topicId,
uint? partitionId, Consumer consumer, PollingStrategy pollingStrategy, uint count, bool autoCommit,
CancellationToken token)
@@ -1043,19 +1132,7 @@ public sealed partial class TcpMessageStream : IIggyClient
var delay = _configuration.ReconnectionSettings.InitialDelay;
do
{
- // The sending semaphore owns every write to _stream, so an in-flight request never observes the
- // field changing between its write and its reply.
- await _sendingSemaphore.WaitAsync(token);
- try
- {
- _stream?.Dispose();
-
- ResetConsensusSession();
- }
- finally
- {
- _sendingSemaphore.Release();
- }
+ await DropStreamAsync();
if (string.IsNullOrEmpty(_currentAddress))
{
@@ -1068,6 +1145,7 @@ public sealed partial class TcpMessageStream : IIggyClient
}
Socket? socket = null;
+ var dialed = false;
try
{
socket = new Socket(ServerAddress.AddressFamilyOf(host), SocketType.Stream, ProtocolType.Tcp);
@@ -1080,6 +1158,7 @@ public sealed partial class TcpMessageStream : IIggyClient
socket.NoDelay = true;
await socket.ConnectAsync(host, port, token);
+ dialed = true;
_currentRemoteAddress = socket.RemoteEndPoint is IPEndPoint remote
? ServerAddress.HostPort(remote.Address.ToString(), (ushort)remote.Port)
@@ -1114,10 +1193,7 @@ public sealed partial class TcpMessageStream : IIggyClient
// forwards the register to the primary.
if (autoLogin && _configuration.AutoLoginSettings.Enabled && !ConsumeSkipAutoLogin())
{
- _logger.LogInformation("Auto login enabled. Trying to login with credentials: {Username}",
- _configuration.AutoLoginSettings.Username);
- await LoginUserAsync(_configuration.AutoLoginSettings.Username,
- _configuration.AutoLoginSettings.Password, token);
+ await AutoLoginAsync(token);
if (await RedirectAsync(token))
{
@@ -1128,6 +1204,21 @@ public sealed partial class TcpMessageStream : IIggyClient
break;
}
+ // Only a failed dial is worth another attempt. Everything past it - a rejected certificate, bad
+ // credentials, a leader that cannot be found - fails the same way every time, and with unlimited
+ // retries a caller would otherwise never get the error back.
+ catch (Exception e) when (dialed || e is OperationCanceledException || _disposed)
+ {
+ socket?.Dispose();
+
+ // The stream is already installed by the time auto login, a redirect or the leader lookup can
+ // fail, and this path leaves the loop for good, so nothing else would ever close it.
+ await DropStreamAsync();
+
+ _logger.LogError(e, "Failed to establish connection");
+ SetConnectionStateAsync(ConnectionState.Disconnected);
+ throw;
+ }
catch (Exception e)
{
socket?.Dispose();
@@ -1180,6 +1271,43 @@ public sealed partial class TcpMessageStream : IIggyClient
}
}
+ /// <summary>
+ /// Closes the current connection and forgets the consensus session bound to it. Takes the sending
+ /// semaphore, which owns every write to <see cref="_stream" />, so an in-flight request never observes
+ /// the field changing between its write and its reply. Never cancellable: a caller giving up is exactly
+ /// when the stream has to be released.
+ /// </summary>
+ private async Task DropStreamAsync()
+ {
+ await _sendingSemaphore.WaitAsync(CancellationToken.None);
+ try
+ {
+ // Left pointing at the disposed stream rather than nulled: a request that raced this drop gets an
+ // ObjectDisposedException, which the reconnect path already understands, instead of an NRE.
+ _stream?.Dispose();
+
+ ResetConsensusSession();
+ }
+ finally
+ {
+ _sendingSemaphore.Release();
+ }
+ }
+
+ private async Task AutoLoginAsync(CancellationToken token)
+ {
+ var settings = _configuration.AutoLoginSettings;
+ if (!string.IsNullOrEmpty(settings.PersonalAccessToken))
+ {
+ _logger.LogInformation("Auto login enabled. Trying to login with a personal access token");
+ await LoginWithPersonalAccessTokenAsync(settings.PersonalAccessToken, token);
+ return;
+ }
+
+ _logger.LogInformation("Auto login enabled. Trying to login with credentials: {Username}", settings.Username);
+ await LoginUserAsync(settings.Username, settings.Password, token);
+ }
+
/// <summary>
/// Whether this connect was triggered by a login or register request that will re-authenticate itself,
/// so the auto-login must sit this one out. Consumes the flag.
@@ -1232,6 +1360,15 @@ public sealed partial class TcpMessageStream : IIggyClient
throw;
}
+ // Without auto login a reconnect cannot re-establish the session, so the request would only come
+ // back unauthenticated. Login and register are the exception: they re-authenticate themselves.
+ if (!_configuration.AutoLoginSettings.Enabled && Volatile.Read(ref _skipAutoLoginOnce) == 0)
+ {
+ _logger.LogWarning("Auto login is disabled, the session cannot be re-established");
+ SetConnectionStateAsync(ConnectionState.Disconnected);
+ throw;
+ }
+
return await HandleReconnectionAsync(payload, token);
}
}
@@ -1279,13 +1416,23 @@ public sealed partial class TcpMessageStream : IIggyClient
return SendRawVsrAsync(payload, token);
}
+ // A stale-client eviction is the server telling this connection its session is gone: the transport already
+ // dropped the stream, so the request is replayed over a fresh connection like any other lost one.
private static bool IsConnectionException(Exception ex)
{
return ex is IggyZeroBytesException or
NotConnectedException or
SocketException or
IOException or
- ObjectDisposedException;
+ ObjectDisposedException or
+ IggyInvalidStatusCodeException { StatusCode: VsrError.STALE_CLIENT, FromServer: true };
+ }
+
+ private static byte[] CreatePingPayload()
+ {
+ var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH];
+ TcpMessageStreamHelpers.CreatePayload(payload, Array.Empty<byte>(), CommandCodes.PING_CODE);
+ return payload;
}
private static int CalculatePayloadBufferSize(int messageBufferSize)
diff --git a/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherBuilder.cs b/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherBuilder.cs
index 270bba930..81ec6c297 100644
--- a/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherBuilder.cs
+++ b/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherBuilder.cs
@@ -127,6 +127,29 @@ public class IggyPublisherBuilder
return this;
}
+ /// <summary>
+ /// Configures the connection settings using a personal access token instead of a username and password.
+ /// </summary>
+ /// <param name="protocol">The protocol to use for the connection (e.g., TCP, UDP).</param>
+ /// <param name="address">The address of the server to connect to.</param>
+ /// <param name="personalAccessToken">The personal access token to authenticate with.</param>
+ /// <param name="receiveBufferSize">The size of the receive buffer.</param>
+ /// <param name="sendBufferSize">The size of the send buffer.</param>
+ /// <param name="reconnectionSettings">Reconnection settings for the client.</param>
+ /// <returns>The current instance of <see cref="IggyPublisherBuilder" /> to allow method chaining.</returns>
+ public IggyPublisherBuilder WithConnection(Protocol protocol, string address, string personalAccessToken,
+ int receiveBufferSize = 4096, int sendBufferSize = 4096, ReconnectionSettings? reconnectionSettings = null)
+ {
+ Config.Protocol = protocol;
+ Config.Address = address;
+ Config.PersonalAccessToken = personalAccessToken;
+ Config.ReceiveBufferSize = receiveBufferSize;
+ Config.SendBufferSize = sendBufferSize;
+ Config.ReconnectionSettings = reconnectionSettings;
+
+ return this;
+ }
+
/// <summary>
/// Configures the partitioning strategy for messages sent by the publisher.
/// Determines how messages are distributed across topic partitions.
@@ -303,7 +326,8 @@ public class IggyPublisherBuilder
ReceiveBufferSize = Config.ReceiveBufferSize,
SendBufferSize = Config.SendBufferSize,
ReconnectionSettings = Config.ReconnectionSettings ?? new ReconnectionSettings(),
- AutoLoginSettings = AutoLoginSettings.For(Config.Login, Config.Password),
+ HeartbeatInterval = Config.HeartbeatInterval,
+ AutoLoginSettings = AutoLogin(),
LoggerFactory = Config.LoggerFactory ?? NullLoggerFactory.Instance,
MessageEncryptor = _encryptor
});
@@ -326,6 +350,18 @@ public class IggyPublisherBuilder
return publisher;
}
+ /// <summary>
+ /// The credentials given to WithConnection must reach the client and not only the explicit login
+ /// performed at startup: a reconnect or a leader redirect drops the session, and without them the
+ /// client would come back unauthenticated.
+ /// </summary>
+ private protected AutoLoginSettings AutoLogin()
+ {
+ return string.IsNullOrEmpty(Config.PersonalAccessToken)
+ ? AutoLoginSettings.For(Config.Login, Config.Password)
+ : AutoLoginSettings.ForPersonalAccessToken(Config.PersonalAccessToken);
+ }
+
/// <summary>
/// Validates the publisher configuration and throws if invalid.
/// </summary>
@@ -339,14 +375,18 @@ public class IggyPublisherBuilder
throw new InvalidOperationException("Address must be provided when CreateIggyClient is true.");
}
- if (string.IsNullOrWhiteSpace(Config.Login))
- {
- throw new InvalidOperationException("Login must be provided when CreateIggyClient is true.");
- }
-
- if (string.IsNullOrWhiteSpace(Config.Password))
+ if (string.IsNullOrWhiteSpace(Config.PersonalAccessToken))
{
- throw new InvalidOperationException("Password must be provided when CreateIggyClient is true.");
+ if (string.IsNullOrWhiteSpace(Config.Login))
+ {
+ throw new InvalidOperationException(
+ "Login or PersonalAccessToken must be provided when CreateIggyClient is true.");
+ }
+
+ if (string.IsNullOrWhiteSpace(Config.Password))
+ {
+ throw new InvalidOperationException("Password must be provided when CreateIggyClient is true.");
+ }
}
}
else
diff --git a/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherBuilderOfT.cs b/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherBuilderOfT.cs
index 6198c7643..97db2487a 100644
--- a/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherBuilderOfT.cs
+++ b/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherBuilderOfT.cs
@@ -92,7 +92,8 @@ public class IggyPublisherBuilder<T> : IggyPublisherBuilder
ReceiveBufferSize = Config.ReceiveBufferSize,
SendBufferSize = Config.SendBufferSize,
ReconnectionSettings = Config.ReconnectionSettings ?? new ReconnectionSettings(),
- AutoLoginSettings = AutoLoginSettings.For(Config.Login, Config.Password),
+ HeartbeatInterval = Config.HeartbeatInterval,
+ AutoLoginSettings = AutoLogin(),
LoggerFactory = Config.LoggerFactory ?? NullLoggerFactory.Instance,
MessageEncryptor = _encryptor
});
diff --git a/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherConfig.cs b/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherConfig.cs
index a4ef3571f..36604663b 100644
--- a/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherConfig.cs
+++ b/foreign/csharp/Iggy_SDK/Publishers/IggyPublisherConfig.cs
@@ -74,6 +74,12 @@ public class IggyPublisherConfig
/// </summary>
public string Password { get; set; } = string.Empty;
+ /// <summary>
+ /// Personal access token used instead of <see cref="Login" /> and <see cref="Password" /> when creating
+ /// the client. Takes precedence over them when set.
+ /// </summary>
+ public string PersonalAccessToken { get; set; } = string.Empty;
+
/// <summary>
/// Gets or sets the identifier of the target stream.
/// Can be either a numeric ID or a string name.
@@ -252,7 +258,14 @@ public class IggyPublisherConfig
/// <summary>
/// Gets or sets the reconnection settings to control the behavior of the iggy client
/// in case of a disconnect or network failure.
- /// This property is optional and can be null. If null, reconnection will be disabled.
+ /// This property is optional and can be null. If null, the default <see cref="Configuration.ReconnectionSettings" />
+ /// apply.
/// </summary>
public ReconnectionSettings? ReconnectionSettings { get; set; }
+
+ /// <summary>
+ /// Interval between the pings the created client sends to keep an idle session alive.
+ /// See <see cref="IggyClientConfigurator.HeartbeatInterval" />.
+ /// </summary>
+ public TimeSpan HeartbeatInterval { get; set; } = TimeSpan.FromSeconds(5);
}
diff --git a/foreign/csharp/Iggy_SDK_Tests/ClientTests/IggyClientFactoryTests.cs b/foreign/csharp/Iggy_SDK_Tests/ClientTests/IggyClientFactoryTests.cs
index de665603d..68f32d264 100644
--- a/foreign/csharp/Iggy_SDK_Tests/ClientTests/IggyClientFactoryTests.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/ClientTests/IggyClientFactoryTests.cs
@@ -51,6 +51,52 @@ public sealed class IggyClientFactoryTests
Assert.Throws<ArgumentOutOfRangeException>(() => IggyClientFactory.CreateClient(options));
}
+ [Theory]
+ [InlineData(0)]
+ [InlineData(-1000)]
+ [InlineData(0.5)]
+ [InlineData(uint.MaxValue)]
+ public void CreateClient_RejectsHeartbeatIntervalOutsideTimerBounds(double milliseconds)
+ {
+ var options = new IggyClientConfigurator
+ {
+ BaseAddress = "127.0.0.1:8090",
+ Protocol = Protocol.Tcp,
+ HeartbeatInterval = TimeSpan.FromMilliseconds(milliseconds)
+ };
+
+ Assert.Throws<ArgumentOutOfRangeException>(() => IggyClientFactory.CreateClient(options));
+ }
+
+ [Fact]
+ public void CreateClient_TcpClientDisposesTwice()
+ {
+ var options = new IggyClientConfigurator
+ {
+ BaseAddress = "127.0.0.1:8090",
+ Protocol = Protocol.Tcp
+ };
+
+ var client = (IDisposable)IggyClientFactory.CreateClient(options);
+
+ client.Dispose();
+ client.Dispose();
+ }
+
+ [Fact]
+ public void Defaults_KeepIdleSessionsAlive()
+ {
+ var options = new IggyClientConfigurator
+ {
+ BaseAddress = "127.0.0.1:8090",
+ Protocol = Protocol.Tcp
+ };
+
+ Assert.Equal(TimeSpan.FromSeconds(5), options.HeartbeatInterval);
+ Assert.True(options.ReconnectionSettings.Enabled);
+ Assert.Equal(0, options.ReconnectionSettings.MaxRetries);
+ }
+
[Fact]
public void CreateClient_AcceptsMaxResponseFrameSizeUnderHttp()
{
diff --git a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/IggyConsumerBuilderTests.cs b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/IggyConsumerBuilderTests.cs
index 817848470..f7d01c11f 100644
--- a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/IggyConsumerBuilderTests.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/IggyConsumerBuilderTests.cs
@@ -71,6 +71,28 @@ public class IggyConsumerBuilderTests
Assert.NotNull(consumer);
}
+ [Fact]
+ public void Build_WithPersonalAccessToken_CreatesTheClient()
+ {
+ var consumer = IggyConsumerBuilder
+ .Create(StreamId, TopicId, Consumer.New(1))
+ .WithConnection(Protocol.Tcp, "127.0.0.1:8090", "token")
+ .Build();
+
+ Assert.NotNull(consumer);
+ }
+
+ [Fact]
+ public void Build_WithoutCredentials_Throws()
+ {
+ var builder = IggyConsumerBuilder
+ .Create(StreamId, TopicId, Consumer.New(1))
+ .WithConnection(Protocol.Tcp, "127.0.0.1:8090", string.Empty);
+
+ var ex = Assert.Throws<InvalidOperationException>(() => builder.Build());
+ Assert.Contains("PersonalAccessToken", ex.Message);
+ }
+
[Fact]
public void TypedBuild_OverTcp_CreatesTheClient()
{
diff --git a/foreign/csharp/README.md b/foreign/csharp/README.md
index 6f60264b4..ce15956ab 100644
--- a/foreign/csharp/README.md
+++ b/foreign/csharp/README.md
@@ -14,11 +14,14 @@ The Apache Iggy C# SDK provides a comprehensive client library for interacting w
offers a modern, async-first API with support for multiple transport protocols and comprehensive message streaming
capabilities.
-> Apache Iggy (Incubating) is an effort undergoing incubation at the Apache Software Foundation (ASF), sponsored by the Apache Incubator PMC.
+> Apache Iggy (Incubating) is an effort undergoing incubation at the Apache Software Foundation (ASF), sponsored by the
+> Apache Incubator PMC.
>
-> Incubation is required of all newly accepted projects until a further review indicates that the infrastructure, communications, and decision making process have stabilized in a manner consistent with other successful ASF projects.
+> Incubation is required of all newly accepted projects until a further review indicates that the infrastructure,
+> communications, and decision making process have stabilized in a manner consistent with other successful ASF projects.
>
-> While incubation status is not necessarily a reflection of the completeness or stability of the code, it does indicate that the project has yet to be fully endorsed by the ASF.
+> While incubation status is not necessarily a reflection of the completeness or stability of the code, it does indicate
+> that the project has yet to be fully endorsed by the ASF.
## Getting Started
@@ -55,7 +58,8 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
await client.ConnectAsync();
```
-Optionally, you can provide an `ILoggerFactory` for diagnostics and debugging (defaults to `NullLoggerFactory.Instance`):
+Optionally, you can provide an `ILoggerFactory` for diagnostics and debugging (defaults to
+`NullLoggerFactory.Instance`):
```c#
var loggerFactory = LoggerFactory.Create(builder =>
@@ -97,11 +101,15 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
CertificatePath = "/path/to/cert"
},
- // Automatic reconnection with exponential backoff
+ // Idle ping keeping the session alive under server-side heartbeat verification (TCP only).
+ // Default 5 seconds, must be between 1 millisecond and about 49 days. Init-only.
+ HeartbeatInterval = TimeSpan.FromSeconds(5),
+
+ // Automatic reconnection with exponential backoff (enabled by default, infinite retries)
ReconnectionSettings = new ReconnectionSettings
{
Enabled = true,
- MaxRetries = 3, // 0 = infinite retries
+ MaxRetries = 0, // 0 = infinite retries
InitialDelay = TimeSpan.FromSeconds(5),
MaxDelay = TimeSpan.FromSeconds(30),
WaitAfterReconnect = TimeSpan.FromSeconds(1),
@@ -109,13 +117,10 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
BackoffMultiplier = 2.0
},
- // Auto-login after connection
- AutoLoginSettings = new AutoLoginSettings
- {
- Enabled = true,
- Username = "your_username",
- Password = "your_password"
- },
+ // Auto-login after connection. Reconnection needs it: without credentials to replay a reconnect cannot
+ // restore the session, so a lost connection fails the request instead
+ AutoLoginSettings = AutoLoginSettings.For("your_username", "your_password"),
+ // or AutoLoginSettings.ForPersonalAccessToken("your_token")
// Optional: logging
LoggerFactory = loggerFactory
@@ -139,12 +144,7 @@ var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
// Upper bound on a reply frame the server announces, 64 MiB by default.
MaxResponseFrameSize = 64 * 1024 * 1024,
- AutoLoginSettings = new AutoLoginSettings
- {
- Enabled = true,
- Username = "iggy",
- Password = "iggy"
- }
+ AutoLoginSettings = AutoLoginSettings.For("iggy", "iggy")
});
await client.ConnectAsync();
@@ -166,8 +166,10 @@ await client.ConnectAsync();
- **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 register body is framed.
- **`PingAsync` costs more than a ping.** Besides the ping it re-syncs the assignment of every consumer group
- this client has joined, so it makes one extra round trip per joined group. The SDK runs no background
- heartbeat: an application that wants assignments refreshed calls `PingAsync` on its own cadence.
+ this client has joined, so it makes one extra round trip per joined group. The TCP client pings on its own
+ every `HeartbeatInterval` (5 seconds by default) while connected, so an idle session survives the server's
+ heartbeat verification and assignments stay fresh; a lost session is repaired by the regular reconnect and
+ auto login.
### Retries and failed requests
@@ -202,6 +204,21 @@ The SDK replays a request whenever the server says it never admitted it. Two cas
- Clients built with `IggyConsumerBuilder` / `IggyPublisherBuilder` now auto-login with the credentials passed
to `WithConnection`. Before, a builder-created client came back from a
reconnect unauthenticated; now the credentials are held for the lifetime of the connection and replayed.
+- The TCP client now pings the server every `HeartbeatInterval` (5 seconds, always on) on its own, and
+ reconnection is on by default (it was off before) with unlimited retries, like the Rust client. Only a
+ failed dial is retried; a rejected certificate, bad credentials or a missing leader is thrown right away.
+ A dropped connection fails every in-flight request at once; they share a single reconnect and are replayed
+ on the connection it establishes. 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 it. Reconnection only replays a request when `AutoLoginSettings` can
+ restore the session; a client that logged in by hand fails fast on a lost connection. Set
+ `ReconnectionSettings.Enabled = false` to opt out of reconnection.
+- `AutoLoginSettings` properties are now `init`-only, as is `IggyClientConfigurator.HeartbeatInterval`. Build
+ them with an object initializer or the `AutoLoginSettings.For` / `AutoLoginSettings.ForPersonalAccessToken`
+ factories instead of assigning after construction.
+- `IggyConsumerBuilder` / `IggyPublisherBuilder` accept a personal access token through the
+ `WithConnection(protocol, address, personalAccessToken, ...)` overload, as an alternative to a username and
+ password.
- The SDK now ships a dependency on `System.IO.Hashing`, used for the client-side message-key partitioner.
- TCP sockets are opened with `NoDelay`. The protocol is request/reply, so a write is
always the last one before the client waits for the answer and Nagle has nothing to coalesce it with - it
@@ -559,7 +576,8 @@ var fullSnapshot = await client.GetSnapshotAsync(
Available compression methods: `Stored`, `Deflated`, `Bzip2`, `Zstd`, `Lzma`, `Xz`.
-Available snapshot types: `FilesystemOverview`, `ProcessList`, `ResourceUsage`, `Test`, `ServerLogs`, `ServerConfig`, `All`.
+Available snapshot types: `FilesystemOverview`, `ProcessList`, `ResourceUsage`, `Test`, `ServerLogs`, `ServerConfig`,
+`All`.
### Segment Management