From 2df6ae6480fd9cf46e8e3b4b248f00af0825cbeb Mon Sep 17 00:00:00 2001 From: Anders Johan Holmefjord Date: Wed, 26 Aug 2026 13:13:32 +0200 Subject: [PATCH 1/8] Add intial sdk code from stream --- .../HeimdallStreamClient.cs | 183 ++++++++++++++++++ .../IHeimdallStreamClient.cs | 16 ++ 2 files changed, 199 insertions(+) create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallStreamClient.cs create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/IHeimdallStreamClient.cs diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallStreamClient.cs new file mode 100644 index 0000000..c5a74ab --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallStreamClient.cs @@ -0,0 +1,183 @@ +using System.Net.ServerSentEvents; +using System.Reflection; +using System.Runtime.CompilerServices; +using System.Text.Json; +using System.Text.Json.Serialization; + +namespace HeimdallPower.Api.Client; + +public class HeimdallStreamClient(HttpClient httpClient) : IHeimdallStreamClient +{ + private static readonly TimeSpan InitialRetryDelay = TimeSpan.FromSeconds(1); + private static readonly TimeSpan MaxRetryDelay = TimeSpan.FromSeconds(30); + + private static readonly string AssemblyVersion = Assembly.GetExecutingAssembly().GetName().Version?.ToString() ?? "0.0.0"; + private const string ClientName = "dotnet-sdk"; + + + /// + /// Streams events, transparently reconnecting with exponential backoff when the + /// connection drops or fails. The consumer only ever sees a continuous sequence of events. + /// + public async IAsyncEnumerable ReceiveAsync( + Guid? gridOwnerId, + Action infoLogger, + [EnumeratorCancellation] CancellationToken token) + { + var failedAttempts = 0; + + // Do as the api sdk does + httpClient.DefaultRequestHeaders.TryAddWithoutValidation("x-client-name", ClientName); + httpClient.DefaultRequestHeaders.TryAddWithoutValidation("x-client-version", AssemblyVersion); + + while (!token.IsCancellationRequested) + { + // The inner iterator is driven manually so that try/catch can wrap MoveNextAsync + // without ever wrapping a yield return (which the compiler forbids). + await using var enumerator = ConnectAndReadAsync(gridOwnerId, httpClient, infoLogger, token) + .GetAsyncEnumerator(token); + + while (true) + { + HeimdallEventEnvelope envelope; + + try + { + if (!await enumerator.MoveNextAsync()) + break; // Stream closed by server - reconnect. + + envelope = enumerator.Current; + failedAttempts = 0; // A successful read resets the backoff. + } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + yield break; + } + catch (Exception ex) + { + failedAttempts++; + infoLogger($"Stream error: {ex.Message}. Reconnecting... (attempt #{failedAttempts})"); + break; + } + + yield return envelope; + } + + if (token.IsCancellationRequested) + yield break; + + try + { + // Should we break the loop after a number of failed attempts? + await Task.Delay(GetRetryDelay(failedAttempts), token); + } + catch (OperationCanceledException) + { + yield break; + } + } + } + + private static async IAsyncEnumerable ConnectAndReadAsync( + Guid? gridOwnerId, + HttpClient httpClient, + Action infoLogger, + [EnumeratorCancellation] CancellationToken token) + { + using var request = new HttpRequestMessage( + HttpMethod.Get, + $"/v1/stream?gridownerid={gridOwnerId}"); + + using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, token); + response.EnsureSuccessStatusCode(); + + await using var stream = await response.Content.ReadAsStreamAsync(token); + + await foreach (SseItem item in SseParser.Create(stream).EnumerateAsync(token)) + { + if (item.EventType == "heartbeat") + { + infoLogger("Heartbeat received..");// This is debug information, should check if the user wants to log it or not + continue; + } + + if (string.IsNullOrWhiteSpace(item.Data)) + continue; + + if (item.EventType == HeimdallDlr.MetricName) + { + HeimdallEventEnvelope? envelope = JsonSerializer.Deserialize(item.Data, HeimdallJsonSerializerOptions.Default); + + if (envelope is null) + continue; + + yield return envelope; + } + } + } + + private static TimeSpan GetRetryDelay(int failedAttempts) + { + if (failedAttempts <= 0) + return InitialRetryDelay; + + var exponential = InitialRetryDelay * Math.Pow(2, Math.Min(failedAttempts, 10)); + var capped = exponential < MaxRetryDelay ? exponential : MaxRetryDelay; + + // Jitter avoids a thundering herd of clients reconnecting simultaneously. + return capped * (0.8 + (Random.Shared.NextDouble() * 0.4)); + } +} + + + +public record HeimdallEventEnvelope( + string SchemaVersion, + string Metric, + string Unit, + JsonElement Data) +{ + private HeimdallDlr? _heimdallDlr; + + public HeimdallDlr? HeimdallDlr + { + get + { + return _heimdallDlr ??= Data.Deserialize(HeimdallJsonSerializerOptions.Default); + } + } +} + +public record HeimdallDlr( + Guid AtLineId, + Guid AtSpanId, + DateTimeOffset Timestamp, + double Value, + bool IsFallback) +{ + public const string MetricName = "Heimdall DLR"; +} + +public static class HeimdallJsonSerializerOptions +{ + private static JsonSerializerOptions? _jsonOptions; + + private static JsonSerializerOptions CreateJsonOptions() + { + var options = new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true, + PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower + }; + options.Converters.Add(new JsonStringEnumConverter()); + return options; + } + + public static JsonSerializerOptions Default + { + get + { + return _jsonOptions ??= CreateJsonOptions(); + } + } +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/IHeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/IHeimdallStreamClient.cs new file mode 100644 index 0000000..20d27e3 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/IHeimdallStreamClient.cs @@ -0,0 +1,16 @@ +namespace HeimdallPower.Api.Client; + +/// +/// Interface for a Heimdall stream client that receives events from the Heimdall Stream API. +/// +public interface IHeimdallStreamClient +{ + /// + /// Receives Heimdall events from the Heimdall Stream API as an asynchronous stream. + /// + /// The ID of the grid owner. + /// A logger action for informational messages. + /// A cancellation token. + /// An asynchronous stream of Heimdall event envelopes. + IAsyncEnumerable ReceiveAsync(Guid? gridOwnerId, Action infoLogger, CancellationToken token); +} From c5f27d6721bd1252698635ec129efa625ea1453e Mon Sep 17 00:00:00 2001 From: Anders Johan Holmefjord Date: Wed, 26 Aug 2026 15:33:47 +0200 Subject: [PATCH 2/8] Make stream client more in line with existing client - and add tests --- dotnet/HeimdallPower.Api.Client.slnx | 1 + .../HeimdallStreamClientExtensions.cs | 52 +++++++ .../HeimdallStreamClientOptions.cs | 28 ++++ .../AccessTokenHeaderRefresher.cs | 57 ++++++++ .../HeimdallPower.Api.Client/ClientHeaders.cs | 31 +++++ .../HeimdallApiClient.cs | 10 +- .../HeimdallApiEndpoints.cs | 14 ++ .../HeimdallApiHttpClient.cs | 93 ++----------- .../Stream/HeimdallDlrEvent.cs | 11 ++ .../Stream/HeimdallEventEnvelope.cs | 20 +++ .../{ => Stream}/HeimdallStreamClient.cs | 127 ++++++------------ .../HeimdallStreamJsonSerializerOptions.cs | 28 ++++ .../{ => Stream}/IHeimdallStreamClient.cs | 11 +- .../Stream/StreamConnectionRetryPolicy.cs | 22 +++ dotnet/README.md | 34 +++++ .../Api.Client.StreamExample.csproj | 14 ++ .../Api.Client.StreamExample/Program.cs | 40 ++++++ .../WhenComputingRetryDelay.cs | 59 ++++++++ .../Fakes/CountingAccessTokenProvider.cs | 25 ++++ .../Fakes/NeverEndingSseStream.cs | 41 ++++++ .../WhenStreaming/Fakes/SseTestData.cs | 43 ++++++ .../WhenStreaming/WhenAuthenticating.cs | 107 +++++++++++++++ .../WhenStreaming/WhenParsingEvents.cs | 124 +++++++++++++++++ .../WhenStreaming/WhenReconnecting.cs | 104 ++++++++++++++ 24 files changed, 921 insertions(+), 175 deletions(-) create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientOptions.cs create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/AccessTokenHeaderRefresher.cs create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/ClientHeaders.cs create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiEndpoints.cs create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallDlrEvent.cs create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs rename dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/{ => Stream}/HeimdallStreamClient.cs (53%) create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamJsonSerializerOptions.cs rename dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/{ => Stream}/IHeimdallStreamClient.cs (58%) create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs create mode 100644 dotnet/examples/Api.Client.StreamExample/Api.Client.StreamExample.csproj create mode 100644 dotnet/examples/Api.Client.StreamExample/Program.cs create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenComputingRetryDelay/WhenComputingRetryDelay.cs create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/CountingAccessTokenProvider.cs create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/NeverEndingSseStream.cs create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/SseTestData.cs create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs diff --git a/dotnet/HeimdallPower.Api.Client.slnx b/dotnet/HeimdallPower.Api.Client.slnx index f321152..42f3630 100644 --- a/dotnet/HeimdallPower.Api.Client.slnx +++ b/dotnet/HeimdallPower.Api.Client.slnx @@ -6,6 +6,7 @@ + diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs new file mode 100644 index 0000000..9db568b --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs @@ -0,0 +1,52 @@ +using HeimdallPower.Api.Client.Stream; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace HeimdallPower.Api.Client.Extensions; + +/// +/// Extension methods for adding the Heimdall Power stream client to the service collection. +/// +public static class HeimdallStreamClientExtensions +{ + /// + /// Adds the Heimdall Power stream client to the service collection. + /// + /// + /// Unlike , no standard resilience + /// handler is applied: its default total-request timeout would tear down the long-lived streaming + /// connection. Reconnection is instead handled internally by . + /// + public static IServiceCollection AddHeimdallPowerStreamClient(this IServiceCollection services, Action configureOptions) + { + const string clientName = "HeimdallPowerStream"; + + services.Configure(configureOptions); + + services.AddHttpClient(clientName) + .ConfigureHttpClient((_, client) => + { + client.BaseAddress = new Uri("https://external-api.heimdallcloud.com"); + client.DefaultRequestHeaders.Add("Accept", "text/event-stream"); + }) + .ConfigurePrimaryHttpMessageHandler(sp => + { + var options = sp.GetRequiredService>().Value; + return ProxyHandlerFactory.CreateHandler(options.Proxy) ?? new HttpClientHandler(); + }); + + services.AddSingleton(sp => + { + var options = sp.GetRequiredService>().Value; + var httpClientFactory = sp.GetRequiredService(); + var httpClient = httpClientFactory.CreateClient(clientName); + var proxyHandler = ProxyHandlerFactory.CreateHandler(options.Proxy); + + return new HeimdallStreamClient(options.ClientId, options.ClientSecret, httpClient, options.ClientMetadata, proxyHandler); + }); + + services.AddSingleton(sp => sp.GetRequiredService()); + + return services; + } +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientOptions.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientOptions.cs new file mode 100644 index 0000000..eefa2d6 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientOptions.cs @@ -0,0 +1,28 @@ +namespace HeimdallPower.Api.Client.Extensions; + +/// +/// Options for configuring the Heimdall Power stream client. +/// +public class HeimdallStreamClientOptions +{ + /// + /// The client ID for the Heimdall Power API. + /// + public required string ClientId { get; set; } + + /// + /// The client secret for the Heimdall Power API. + /// + public required string ClientSecret { get; set; } + + /// + /// Additional metadata to include in the request headers. + /// + public Dictionary? ClientMetadata { get; set; } + + /// + /// Optional proxy configuration. When set, all HTTP requests (stream connection and token acquisition) + /// are routed through the specified proxy. + /// + public ProxyOptions? Proxy { get; set; } +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/AccessTokenHeaderRefresher.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/AccessTokenHeaderRefresher.cs new file mode 100644 index 0000000..3b6ce03 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/AccessTokenHeaderRefresher.cs @@ -0,0 +1,57 @@ +namespace HeimdallPower.Api.Client; + +/// +/// Keeps an 's auth and client headers fresh, coalescing concurrent +/// refresh attempts behind a single in-flight token request. +/// +internal sealed class AccessTokenHeaderRefresher( + IAccessTokenProvider accessTokenProvider, + HttpClient httpClient, + Dictionary? clientMetadata = null) +{ + private static readonly TimeSpan TokenExpirationBuffer = TimeSpan.FromMinutes(2); + + private readonly SemaphoreSlim _tokenLock = new(1, 1); + private DateTimeOffset _tokenExpiresOn; + + private bool IsTokenFresh => _tokenExpiresOn != default && DateTimeOffset.UtcNow.Add(TokenExpirationBuffer) <= _tokenExpiresOn; + + /// Refreshes the token only if it is missing or within the expiration buffer. + public Task EnsureFreshTokenAsync(CancellationToken cancellationToken) => + IsTokenFresh ? Task.CompletedTask : RefreshAsync(force: false, cancellationToken); + + /// Refreshes the token unconditionally, e.g. after the server rejects it as unauthorized. + public Task ForceRefreshAsync(CancellationToken cancellationToken) => RefreshAsync(force: true, cancellationToken); + + private async Task RefreshAsync(bool force, CancellationToken cancellationToken) + { + await _tokenLock.WaitAsync(TimeSpan.FromSeconds(30), cancellationToken); + try + { + if (!force && IsTokenFresh) + return; // Another caller already refreshed while we were waiting. + + await accessTokenProvider.AcquireTokenAsync(cancellationToken); + _tokenExpiresOn = accessTokenProvider.GetTokenExpiry(); + + foreach (var header in accessTokenProvider.GetAccessHeaders()) + { + httpClient.DefaultRequestHeaders.Remove(header.Key); + httpClient.DefaultRequestHeaders.TryAddWithoutValidation(header.Key, header.Value); + } + + foreach (var header in ClientHeaders.Build(clientMetadata)) + { + if (header.Key.Equals("x-region", StringComparison.OrdinalIgnoreCase)) + continue; // x-region comes from the token, not client metadata + + httpClient.DefaultRequestHeaders.Remove(header.Key); + httpClient.DefaultRequestHeaders.TryAddWithoutValidation(header.Key, header.Value); + } + } + finally + { + _tokenLock.Release(); + } + } +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/ClientHeaders.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/ClientHeaders.cs new file mode 100644 index 0000000..ba602dd --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/ClientHeaders.cs @@ -0,0 +1,31 @@ +using System.Reflection; + +namespace HeimdallPower.Api.Client; + +/// +/// Builds the x-client-name/x-client-version (plus any caller-supplied metadata) headers sent by every SDK client. +/// +internal static class ClientHeaders +{ + private const string ClientName = "dotnet-sdk"; + private static readonly string AssemblyVersion = Assembly.GetExecutingAssembly().GetName().Version?.ToString() ?? "0.0.0"; + + public static Dictionary Build(Dictionary? clientMetadata) + { + var headers = new Dictionary + { + { "x-client-name", ClientName }, + { "x-client-version", AssemblyVersion }, + }; + + if (clientMetadata != null) + { + foreach (var kvp in clientMetadata) + { + headers[kvp.Key] = kvp.Value; // Overwrite defaults if present + } + } + + return headers; + } +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiClient.cs index 60f189c..b1ca652 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiClient.cs @@ -13,12 +13,6 @@ namespace HeimdallPower.Api.Client; /// public class HeimdallApiClient : IHeimdallApiClient { - private const string ApiUrl = "https://external-api.heimdallcloud.com"; - private const string Policy = "B2C_1A_CLIENTCREDENTIALSFLOW"; - private const string Instance = "https://hpadb2cprod.b2clogin.com"; - private const string Domain = "hpadb2cprod.onmicrosoft.com"; - private const string Scope = $"https://{Domain}/dc5758ae-4eea-416e-9e61-812914d9a49a/.default"; - private const string Authority = $"{Instance}/tfp/{Domain}/{Policy}"; private readonly HeimdallApiHttpClient _heimdallApiClient; /// @@ -27,8 +21,8 @@ public class HeimdallApiClient : IHeimdallApiClient /// public HeimdallApiClient(string clientId, string clientSecret, HttpClient? httpClient = null, Dictionary? clientMetadata = null, HttpMessageHandler? proxyHandler = null) { - var accessTokenProvider = new AccessTokenProvider(clientId, clientSecret, Authority, Scope, proxyHandler); - _heimdallApiClient = new HeimdallApiHttpClient(accessTokenProvider, httpClient ?? new HttpClient { BaseAddress = new Uri(ApiUrl) }, clientMetadata); + var accessTokenProvider = new AccessTokenProvider(clientId, clientSecret, HeimdallApiEndpoints.Authority, HeimdallApiEndpoints.Scope, proxyHandler); + _heimdallApiClient = new HeimdallApiHttpClient(accessTokenProvider, httpClient ?? new HttpClient { BaseAddress = new Uri(HeimdallApiEndpoints.ApiUrl) }, clientMetadata); } /// diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiEndpoints.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiEndpoints.cs new file mode 100644 index 0000000..7a0c61b --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiEndpoints.cs @@ -0,0 +1,14 @@ +namespace HeimdallPower.Api.Client; + +/// +/// Endpoint/authority constants shared by every client (REST, streaming) in this SDK. +/// +internal static class HeimdallApiEndpoints +{ + public const string ApiUrl = "https://external-api.heimdallcloud.com"; + private const string Policy = "B2C_1A_CLIENTCREDENTIALSFLOW"; + private const string Instance = "https://hpadb2cprod.b2clogin.com"; + private const string Domain = "hpadb2cprod.onmicrosoft.com"; + public const string Scope = $"https://{Domain}/dc5758ae-4eea-416e-9e61-812914d9a49a/.default"; + public const string Authority = $"{Instance}/tfp/{Domain}/{Policy}"; +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiHttpClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiHttpClient.cs index 7e23d2a..b98f650 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiHttpClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiHttpClient.cs @@ -1,18 +1,13 @@ using System.Net; -using System.Reflection; using System.Text.Json; using System.Text.RegularExpressions; namespace HeimdallPower.Api.Client; -internal class HeimdallApiHttpClient( - IAccessTokenProvider accessTokenProvider, - HttpClient httpClient, - Dictionary? clientMetadata = null) +internal class HeimdallApiHttpClient { - private HttpClient HttpClient { get; } = httpClient; - - private readonly SemaphoreSlim _tokenLock = new(1, 1); + private HttpClient HttpClient { get; } + private readonly AccessTokenHeaderRefresher _tokenRefresher; private readonly JsonSerializerOptions _jsonSerializerOptions = new() { @@ -21,14 +16,17 @@ internal class HeimdallApiHttpClient( WriteIndented = true }; - private DateTimeOffset _tokenExpiresOn; - private static readonly TimeSpan TokenExpirationBuffer = TimeSpan.FromMinutes(2); - private static readonly JsonSerializerOptions ProblemDetailsOptions = new() { PropertyNameCaseInsensitive = true }; + public HeimdallApiHttpClient(IAccessTokenProvider accessTokenProvider, HttpClient httpClient, Dictionary? clientMetadata = null) + { + HttpClient = httpClient; + _tokenRefresher = new AccessTokenHeaderRefresher(accessTokenProvider, httpClient, clientMetadata); + } + public async Task GetAsync(string url, CancellationToken cancellationToken = default) { return await ExecuteWithAuthRetry(async () => @@ -92,82 +90,13 @@ private async Task ExecuteWithAuthRetry(Func> operationFunc, Cance { try { - await UpdateAccessTokenIfExpired(cancellationToken); + await _tokenRefresher.EnsureFreshTokenAsync(cancellationToken); return await operationFunc(); } catch (UnauthorizedAccessException) { - await RefreshAccessToken(cancellationToken); + await _tokenRefresher.ForceRefreshAsync(cancellationToken); return await operationFunc(); } } - - private async Task UpdateAccessTokenIfExpired(CancellationToken cancellationToken) - { - if (_tokenExpiresOn == default || DateTimeOffset.UtcNow.Add(TokenExpirationBuffer) > _tokenExpiresOn) - { - await RefreshAccessToken(cancellationToken); - } - } - - private static readonly string AssemblyVersion = Assembly.GetExecutingAssembly().GetName().Version?.ToString() ?? "0.0.0"; - private const string ClientName = "dotnet-sdk"; - - /// - /// Builds the client headers to be sent with each request. - /// Includes client name, version, and any additional metadata provided. - /// - /// - private Dictionary BuildClientHeaders() - { - var headers = new Dictionary - { - { "x-client-name", ClientName }, - { "x-client-version", AssemblyVersion }, - }; - - if (clientMetadata != null) - { - foreach (var kvp in clientMetadata) - { - headers[kvp.Key] = kvp.Value; // Overwrite defaults if present - } - } - - return headers; - } - - private async Task RefreshAccessToken(CancellationToken cancellationToken) - { - await _tokenLock.WaitAsync(TimeSpan.FromSeconds(30), cancellationToken); - try - { - // Check if another thread already refreshed while we were waiting - if (_tokenExpiresOn != default && DateTimeOffset.UtcNow.Add(TokenExpirationBuffer) <= _tokenExpiresOn) - return; - - await accessTokenProvider.AcquireTokenAsync(cancellationToken); - _tokenExpiresOn = accessTokenProvider.GetTokenExpiry(); - - foreach (var header in accessTokenProvider.GetAccessHeaders()) - { - HttpClient.DefaultRequestHeaders.Remove(header.Key); - HttpClient.DefaultRequestHeaders.TryAddWithoutValidation(header.Key, header.Value); - } - foreach (var header in BuildClientHeaders()) - { - if (header.Key.Equals("x-region", StringComparison.OrdinalIgnoreCase)) - { - continue; // Skip adding x-region as this should be set from the token - } - - HttpClient.DefaultRequestHeaders.Remove(header.Key); - HttpClient.DefaultRequestHeaders.TryAddWithoutValidation(header.Key, header.Value); - } - } - finally - { - _tokenLock.Release(); - } - } } diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallDlrEvent.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallDlrEvent.cs new file mode 100644 index 0000000..b04c402 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallDlrEvent.cs @@ -0,0 +1,11 @@ +namespace HeimdallPower.Api.Client.Stream; + +public record HeimdallDlrEvent( + Guid AtLineId, + Guid AtSpanId, + DateTimeOffset Timestamp, + double Value, + bool IsFallback) +{ + public const string MetricName = "Heimdall DLR"; +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs new file mode 100644 index 0000000..b0e63b2 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs @@ -0,0 +1,20 @@ +using System.Text.Json; + +namespace HeimdallPower.Api.Client.Stream; + +public record HeimdallEventEnvelope( + string SchemaVersion, + string Metric, + string Unit, + JsonElement Data) +{ + private HeimdallDlrEvent? _heimdallDlr; + + public HeimdallDlrEvent? HeimdallDlr + { + get + { + return _heimdallDlr ??= Data.Deserialize(HeimdallStreamJsonSerializerOptions.Default); + } + } +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs similarity index 53% rename from dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallStreamClient.cs rename to dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs index c5a74ab..0489ad0 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallStreamClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs @@ -1,19 +1,35 @@ +using System.Net; using System.Net.ServerSentEvents; -using System.Reflection; using System.Runtime.CompilerServices; using System.Text.Json; -using System.Text.Json.Serialization; -namespace HeimdallPower.Api.Client; +namespace HeimdallPower.Api.Client.Stream; -public class HeimdallStreamClient(HttpClient httpClient) : IHeimdallStreamClient +public class HeimdallStreamClient : IHeimdallStreamClient { - private static readonly TimeSpan InitialRetryDelay = TimeSpan.FromSeconds(1); - private static readonly TimeSpan MaxRetryDelay = TimeSpan.FromSeconds(30); + private readonly HttpClient _httpClient; + private readonly AccessTokenHeaderRefresher _tokenRefresher; + private readonly StreamConnectionRetryPolicy _retryPolicy; - private static readonly string AssemblyVersion = Assembly.GetExecutingAssembly().GetName().Version?.ToString() ?? "0.0.0"; - private const string ClientName = "dotnet-sdk"; + /// + /// A client that lets you consume the Heimdall Stream API. + /// Throws on non-transient errors. + /// + public HeimdallStreamClient(string clientId, string clientSecret, HttpClient? httpClient = null, Dictionary? clientMetadata = null, HttpMessageHandler? proxyHandler = null) + : this( + new AccessTokenProvider(clientId, clientSecret, HeimdallApiEndpoints.Authority, HeimdallApiEndpoints.Scope, proxyHandler), + httpClient ?? new HttpClient { BaseAddress = new Uri(HeimdallApiEndpoints.ApiUrl) }, + clientMetadata) + { + } + // Seam for unit tests to inject a stub IAccessTokenProvider and a zero-delay retry policy instead of hitting real MSAL/AAD and real backoff delays. + internal HeimdallStreamClient(IAccessTokenProvider accessTokenProvider, HttpClient httpClient, Dictionary? clientMetadata = null, StreamConnectionRetryPolicy? retryPolicy = null) + { + _httpClient = httpClient; + _tokenRefresher = new AccessTokenHeaderRefresher(accessTokenProvider, httpClient, clientMetadata); + _retryPolicy = retryPolicy ?? new StreamConnectionRetryPolicy(); + } /// /// Streams events, transparently reconnecting with exponential backoff when the @@ -26,15 +42,11 @@ public async IAsyncEnumerable ReceiveAsync( { var failedAttempts = 0; - // Do as the api sdk does - httpClient.DefaultRequestHeaders.TryAddWithoutValidation("x-client-name", ClientName); - httpClient.DefaultRequestHeaders.TryAddWithoutValidation("x-client-version", AssemblyVersion); - while (!token.IsCancellationRequested) { // The inner iterator is driven manually so that try/catch can wrap MoveNextAsync // without ever wrapping a yield return (which the compiler forbids). - await using var enumerator = ConnectAndReadAsync(gridOwnerId, httpClient, infoLogger, token) + await using var enumerator = ConnectAndReadAsync(gridOwnerId, infoLogger, token) .GetAsyncEnumerator(token); while (true) @@ -53,6 +65,13 @@ public async IAsyncEnumerable ReceiveAsync( { yield break; } + catch (UnauthorizedAccessException) + { + failedAttempts++; + infoLogger($"Stream error: unauthorized. Refreshing token and reconnecting... (attempt #{failedAttempts})"); + await _tokenRefresher.ForceRefreshAsync(token); + break; + } catch (Exception ex) { failedAttempts++; @@ -69,7 +88,7 @@ public async IAsyncEnumerable ReceiveAsync( try { // Should we break the loop after a number of failed attempts? - await Task.Delay(GetRetryDelay(failedAttempts), token); + await Task.Delay(_retryPolicy.GetDelay(failedAttempts), token); } catch (OperationCanceledException) { @@ -78,17 +97,22 @@ public async IAsyncEnumerable ReceiveAsync( } } - private static async IAsyncEnumerable ConnectAndReadAsync( + private async IAsyncEnumerable ConnectAndReadAsync( Guid? gridOwnerId, - HttpClient httpClient, Action infoLogger, [EnumeratorCancellation] CancellationToken token) { + await _tokenRefresher.EnsureFreshTokenAsync(token); + using var request = new HttpRequestMessage( HttpMethod.Get, $"/v1/stream?gridownerid={gridOwnerId}"); - using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, token); + using var response = await _httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, token); + + if (response.StatusCode == HttpStatusCode.Unauthorized) + throw new UnauthorizedAccessException("Unauthorized access. Please check your credentials."); + response.EnsureSuccessStatusCode(); await using var stream = await response.Content.ReadAsStreamAsync(token); @@ -104,9 +128,9 @@ private static async IAsyncEnumerable ConnectAndReadAsync if (string.IsNullOrWhiteSpace(item.Data)) continue; - if (item.EventType == HeimdallDlr.MetricName) + if (item.EventType == HeimdallDlrEvent.MetricName) { - HeimdallEventEnvelope? envelope = JsonSerializer.Deserialize(item.Data, HeimdallJsonSerializerOptions.Default); + HeimdallEventEnvelope? envelope = JsonSerializer.Deserialize(item.Data, HeimdallStreamJsonSerializerOptions.Default); if (envelope is null) continue; @@ -115,69 +139,4 @@ private static async IAsyncEnumerable ConnectAndReadAsync } } } - - private static TimeSpan GetRetryDelay(int failedAttempts) - { - if (failedAttempts <= 0) - return InitialRetryDelay; - - var exponential = InitialRetryDelay * Math.Pow(2, Math.Min(failedAttempts, 10)); - var capped = exponential < MaxRetryDelay ? exponential : MaxRetryDelay; - - // Jitter avoids a thundering herd of clients reconnecting simultaneously. - return capped * (0.8 + (Random.Shared.NextDouble() * 0.4)); - } -} - - - -public record HeimdallEventEnvelope( - string SchemaVersion, - string Metric, - string Unit, - JsonElement Data) -{ - private HeimdallDlr? _heimdallDlr; - - public HeimdallDlr? HeimdallDlr - { - get - { - return _heimdallDlr ??= Data.Deserialize(HeimdallJsonSerializerOptions.Default); - } - } -} - -public record HeimdallDlr( - Guid AtLineId, - Guid AtSpanId, - DateTimeOffset Timestamp, - double Value, - bool IsFallback) -{ - public const string MetricName = "Heimdall DLR"; -} - -public static class HeimdallJsonSerializerOptions -{ - private static JsonSerializerOptions? _jsonOptions; - - private static JsonSerializerOptions CreateJsonOptions() - { - var options = new JsonSerializerOptions - { - PropertyNameCaseInsensitive = true, - PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower - }; - options.Converters.Add(new JsonStringEnumConverter()); - return options; - } - - public static JsonSerializerOptions Default - { - get - { - return _jsonOptions ??= CreateJsonOptions(); - } - } } diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamJsonSerializerOptions.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamJsonSerializerOptions.cs new file mode 100644 index 0000000..301e0c5 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamJsonSerializerOptions.cs @@ -0,0 +1,28 @@ +using System.Text.Json; +using System.Text.Json.Serialization; + +namespace HeimdallPower.Api.Client.Stream; + +internal static class HeimdallStreamJsonSerializerOptions +{ + private static JsonSerializerOptions? _jsonOptions; + + private static JsonSerializerOptions CreateJsonOptions() + { + var options = new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true, + PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower + }; + options.Converters.Add(new JsonStringEnumConverter()); + return options; + } + + public static JsonSerializerOptions Default + { + get + { + return _jsonOptions ??= CreateJsonOptions(); + } + } +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/IHeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs similarity index 58% rename from dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/IHeimdallStreamClient.cs rename to dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs index 20d27e3..88a7e38 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/IHeimdallStreamClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs @@ -1,8 +1,17 @@ -namespace HeimdallPower.Api.Client; +namespace HeimdallPower.Api.Client.Stream; /// /// Interface for a Heimdall stream client that receives events from the Heimdall Stream API. /// +/// +/// Authentication: handled internally. The token is acquired and refreshed automatically +/// before connecting and after an unauthorized (401) response, using the same client credentials +/// flow as . +/// +/// Reconnection: connection drops and transient failures are retried internally with +/// exponential backoff. Callers only ever see a continuous sequence of events. +/// +/// public interface IHeimdallStreamClient { /// diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs new file mode 100644 index 0000000..5641d53 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs @@ -0,0 +1,22 @@ +namespace HeimdallPower.Api.Client.Stream; + +/// +/// Computes the exponential backoff (with jitter) delay between stream reconnect attempts. +/// +internal sealed class StreamConnectionRetryPolicy(TimeSpan? initialDelay = null, TimeSpan? maxDelay = null) +{ + private readonly TimeSpan _initialDelay = initialDelay ?? TimeSpan.FromSeconds(1); + private readonly TimeSpan _maxDelay = maxDelay ?? TimeSpan.FromSeconds(30); + + public TimeSpan GetDelay(int failedAttempts) + { + if (failedAttempts <= 0) + return _initialDelay; + + var exponential = _initialDelay * Math.Pow(2, Math.Min(failedAttempts, 10)); + var capped = exponential < _maxDelay ? exponential : _maxDelay; + + // Jitter avoids a thundering herd of clients reconnecting simultaneously. + return capped * (0.8 + (Random.Shared.NextDouble() * 0.4)); + } +} diff --git a/dotnet/README.md b/dotnet/README.md index fee273e..33988ac 100644 --- a/dotnet/README.md +++ b/dotnet/README.md @@ -75,6 +75,40 @@ services.AddHeimdallPowerApiClient(options => When no explicit `Address` is set, the SDK falls back to `HTTPS_PROXY`/`HTTP_PROXY`/`NO_PROXY` environment variables. The proxy applies to both API calls and token acquisition. +## Streaming + +`HeimdallStreamClient` consumes the Heimdall Stream API (Server-Sent Events), transparently reconnecting with exponential backoff. It authenticates the same way as `HeimdallApiClient`, using the same client credentials. + +```csharp +using HeimdallPower.Api.Client.Stream; + +var streamClient = new HeimdallStreamClient(clientId, clientSecret); + +await foreach (var envelope in streamClient.ReceiveAsync(gridOwnerId: null, infoLogger: Console.WriteLine, cancellationToken)) +{ + if (envelope.HeimdallDlr is { } dlr) + { + Console.WriteLine($"{dlr.Value} {envelope.Unit} at {dlr.Timestamp}"); + } +} +``` + +Using the `HeimdallPower.Api.Client.Extensions` package: + +```csharp +services.AddHeimdallPowerStreamClient(options => +{ + options.ClientId = "your-client-id"; + options.ClientSecret = "your-client-secret"; +}); + +var streamClient = provider.GetRequiredService(); +``` + +`AddHeimdallPowerStreamClient` reuses the same proxy configuration (`ProxyOptions`) as `AddHeimdallPowerApiClient`, but does **not** apply the standard resilience handler — reconnection for the long-lived stream connection is handled internally by `HeimdallStreamClient` instead. + +See the full example in [`examples/Api.Client.StreamExample`](examples/Api.Client.StreamExample). + ## Error Handling ### Resilience and retry diff --git a/dotnet/examples/Api.Client.StreamExample/Api.Client.StreamExample.csproj b/dotnet/examples/Api.Client.StreamExample/Api.Client.StreamExample.csproj new file mode 100644 index 0000000..014037c --- /dev/null +++ b/dotnet/examples/Api.Client.StreamExample/Api.Client.StreamExample.csproj @@ -0,0 +1,14 @@ + + + + Exe + net10.0 + enable + enable + + + + + + + diff --git a/dotnet/examples/Api.Client.StreamExample/Program.cs b/dotnet/examples/Api.Client.StreamExample/Program.cs new file mode 100644 index 0000000..9747897 --- /dev/null +++ b/dotnet/examples/Api.Client.StreamExample/Program.cs @@ -0,0 +1,40 @@ +using HeimdallPower.Api.Client.Stream; + +// Configuration setup +const string clientId = "insert-your-client-id-here"; +const string clientSecret = "insert-your-client-secret-here"; + +Console.WriteLine("Initiating Heimdall Stream client"); + +// Note: direct instantiation does NOT include automatic HTTP resilience. +// Reconnection with backoff for dropped/failed connections is handled internally by HeimdallStreamClient. +var streamClient = new HeimdallStreamClient(clientId, clientSecret); + +using var cts = new CancellationTokenSource(); +Console.CancelKeyPress += (_, e) => +{ + e.Cancel = true; // Let the loop below exit gracefully instead of killing the process immediately. + cts.Cancel(); +}; + +Console.WriteLine("Listening for events. Press Ctrl+C to stop."); + +try +{ + // Pass a specific grid owner ID to only receive events for that grid owner. + await foreach (var envelope in streamClient.ReceiveAsync(gridOwnerId: null, infoLogger: Console.WriteLine, cts.Token)) + { + if (envelope.HeimdallDlr is { } dlr) + { + Console.WriteLine($"- Heimdall DLR: {dlr.Value} {envelope.Unit} for line {dlr.AtLineId} at {dlr.Timestamp} (IsFallback={dlr.IsFallback})"); + } + else + { + Console.WriteLine($"- {envelope.Metric}: {envelope.Data}"); + } + } +} +catch (OperationCanceledException) +{ + Console.WriteLine("Stream stopped."); +} diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenComputingRetryDelay/WhenComputingRetryDelay.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenComputingRetryDelay/WhenComputingRetryDelay.cs new file mode 100644 index 0000000..f33eb7a --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenComputingRetryDelay/WhenComputingRetryDelay.cs @@ -0,0 +1,59 @@ +using HeimdallPower.Api.Client.Stream; + +namespace HeimdallPower.Api.Client.UnitTests.WhenComputingRetryDelay; + +/// +/// Data-driven tests for 's exponential backoff + jitter math. +/// Jitter is random, so assertions check bounds rather than exact values. +/// +[Trait("Category", "Unit")] +public class WhenComputingRetryDelay +{ + private static readonly TimeSpan InitialDelay = TimeSpan.FromSeconds(1); + private static readonly TimeSpan MaxDelay = TimeSpan.FromSeconds(30); + + private readonly StreamConnectionRetryPolicy _policy = new(InitialDelay, MaxDelay); + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void ShouldReturnInitialDelay_WhenNoFailedAttempts(int failedAttempts) + { + var delay = _policy.GetDelay(failedAttempts); + + Assert.Equal(InitialDelay, delay); + } + + [Theory] + [InlineData(1, 1.6, 2.4)] // 1s * 2^1 = 2s, +/-20% jitter + [InlineData(2, 3.2, 4.8)] // 1s * 2^2 = 4s, +/-20% jitter + [InlineData(3, 6.4, 9.6)] // 1s * 2^3 = 8s, +/-20% jitter + public void ShouldDoubleDelayPerAttempt_WithinJitterBand(int failedAttempts, double minSeconds, double maxSeconds) + { + var delay = _policy.GetDelay(failedAttempts); + + Assert.InRange(delay.TotalSeconds, minSeconds, maxSeconds); + } + + [Theory] + [InlineData(10)] + [InlineData(100)] + public void ShouldCapAtMaxDelay_WithinJitterBand(int failedAttempts) + { + var delay = _policy.GetDelay(failedAttempts); + + Assert.InRange(delay.TotalSeconds, MaxDelay.TotalSeconds * 0.8, MaxDelay.TotalSeconds * 1.2); + } + + [Fact] + public void ShouldNeverExceedMaxDelayPlusJitter_AcrossManyAttempts() + { + for (var attempt = 1; attempt <= 20; attempt++) + { + var delay = _policy.GetDelay(attempt); + + Assert.True(delay <= MaxDelay * 1.2, $"Delay {delay} at attempt {attempt} exceeded max+jitter bound"); + Assert.True(delay >= InitialDelay * 0.8, $"Delay {delay} at attempt {attempt} was below the initial*0.8 floor"); + } + } +} diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/CountingAccessTokenProvider.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/CountingAccessTokenProvider.cs new file mode 100644 index 0000000..5943514 --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/CountingAccessTokenProvider.cs @@ -0,0 +1,25 @@ +namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming.Fakes; + +/// +/// A test double for that tracks how many times a token +/// was acquired, so tests can assert on refresh/reuse behavior. +/// +internal sealed class CountingAccessTokenProvider(DateTimeOffset? expiresOn = null) : IAccessTokenProvider +{ + public int AcquireTokenCallCount { get; private set; } + + public Task AcquireTokenAsync(CancellationToken cancellationToken = default) + { + AcquireTokenCallCount++; + return Task.CompletedTask; + } + + public DateTimeOffset GetTokenExpiry() => expiresOn ?? DateTimeOffset.UtcNow.AddHours(1); + + public IDictionary GetAccessHeaders() => + new Dictionary + { + { "Authorization", $"Bearer stub-token-{AcquireTokenCallCount}" }, + { "x-region", "stub-region" }, + }; +} diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/NeverEndingSseStream.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/NeverEndingSseStream.cs new file mode 100644 index 0000000..6d8064d --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/NeverEndingSseStream.cs @@ -0,0 +1,41 @@ +using System.Text; + +namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming.Fakes; + +/// +/// A stream that yields a fixed initial payload, then blocks (respecting cancellation) as if the +/// connection were still open with no further data - used to test clean cancellation mid-stream. +/// +internal sealed class NeverEndingSseStream(string initialBody) : System.IO.Stream +{ + private readonly byte[] _initial = Encoding.UTF8.GetBytes(initialBody); + private int _position; + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => throw new NotSupportedException(); + public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + + public override int Read(byte[] buffer, int offset, int count) => + ReadAsync(buffer, offset, count, CancellationToken.None).GetAwaiter().GetResult(); + + public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + if (_position < _initial.Length) + { + var toCopy = Math.Min(count, _initial.Length - _position); + Array.Copy(_initial, _position, buffer, offset, toCopy); + _position += toCopy; + return toCopy; + } + + await Task.Delay(Timeout.Infinite, cancellationToken); + return 0; + } + + public override void Flush() { } + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); +} diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/SseTestData.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/SseTestData.cs new file mode 100644 index 0000000..ac9a3a1 --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/SseTestData.cs @@ -0,0 +1,43 @@ +using System.Net; +using System.Text; +using System.Text.Json; + +namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming.Fakes; + +/// +/// Builds canned SSE (text/event-stream) response bodies and HTTP responses for stream client tests. +/// +internal static class SseTestData +{ + public static string HeartbeatEvent => "event: heartbeat\ndata: \n\n"; + + public static string DlrEvent(Guid lineId, Guid spanId, DateTimeOffset timestamp, double value, bool isFallback = false) + { + var payload = new + { + schema_version = "1.0", + metric = "Heimdall DLR", + unit = "Ampere", + data = new { at_line_id = lineId, at_span_id = spanId, timestamp, value, is_fallback = isFallback }, + }; + var json = JsonSerializer.Serialize(payload); + return $"event: Heimdall DLR\ndata: {json}\n\n"; + } + + public static string UnknownEvent => "event: something_else\ndata: {\"foo\":\"bar\"}\n\n"; + + public static string MalformedDlrEvent => "event: Heimdall DLR\ndata: not-json\n\n"; + + public static HttpResponseMessage OkResponse(string sseBody) => new(HttpStatusCode.OK) + { + Content = new StringContent(sseBody, Encoding.UTF8, "text/event-stream"), + }; + + public static HttpResponseMessage OkResponseNeverEnding(string initialSseBody) => new(HttpStatusCode.OK) + { + Content = new StreamContent(new NeverEndingSseStream(initialSseBody)) + { + Headers = { ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("text/event-stream") }, + }, + }; +} diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs new file mode 100644 index 0000000..946574b --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs @@ -0,0 +1,107 @@ +using HeimdallPower.Api.Client.Stream; +using HeimdallPower.Api.Client.UnitTests.WhenStreaming.Fakes; +using HeimdallPower.Api.Client.UnitTests.WhenUsingResilienceExtensions.Fakes; + +namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming; + +/// +/// Verifies that acquires, reuses, and refreshes its access +/// token correctly - no network calls, no real MSAL/AAD. +/// +[Trait("Category", "Unit")] +public class WhenAuthenticating +{ + private static readonly StreamConnectionRetryPolicy ZeroDelay = new(TimeSpan.Zero, TimeSpan.Zero); + + [Fact] + public async Task ShouldAttachAuthAndClientHeaders_BeforeFirstConnect() + { + HttpRequestMessage? capturedRequest = null; + var handler = new CountingHttpMessageHandler(request => + { + capturedRequest = request; + return SseTestData.OkResponse(SseTestData.DlrEvent(Guid.NewGuid(), Guid.NewGuid(), DateTimeOffset.UtcNow, 1.0)); + }); + + var client = new HeimdallStreamClient( + new CountingAccessTokenProvider(), + new HttpClient(handler) { BaseAddress = new Uri("https://fake-stream.example.com") }, + retryPolicy: ZeroDelay); + + using var cts = new CancellationTokenSource(); + await foreach (var _ in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + { + break; + } + + Assert.NotNull(capturedRequest); + Assert.True(capturedRequest!.Headers.Contains("Authorization")); + Assert.True(capturedRequest.Headers.Contains("x-region")); + Assert.True(capturedRequest.Headers.Contains("x-client-name")); + Assert.True(capturedRequest.Headers.Contains("x-client-version")); + } + + [Fact] + public async Task ShouldReuseToken_AcrossReconnects_WhileStillFresh() + { + var lineId = Guid.NewGuid(); + var spanId = Guid.NewGuid(); + var handler = new CountingHttpMessageHandler(_ => + SseTestData.OkResponse(SseTestData.DlrEvent(lineId, spanId, DateTimeOffset.UtcNow, 1.0))); + + var tokenProvider = new CountingAccessTokenProvider(expiresOn: DateTimeOffset.UtcNow.AddHours(1)); + var client = new HeimdallStreamClient( + tokenProvider, + new HttpClient(handler) { BaseAddress = new Uri("https://fake-stream.example.com") }, + retryPolicy: ZeroDelay); + + using var cts = new CancellationTokenSource(); + var receivedCount = 0; + await foreach (var _ in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + { + receivedCount++; + if (receivedCount == 2) + break; // Received one event from each of two connects (server closes after each canned response). + } + + Assert.Equal(2, receivedCount); + Assert.Equal(1, tokenProvider.AcquireTokenCallCount); + } + + [Fact] + public async Task ShouldForceRefreshToken_OnUnauthorizedResponse() + { + var lineId = Guid.NewGuid(); + var spanId = Guid.NewGuid(); + + var callCount = 0; + var handler = new CountingHttpMessageHandler(_ => + { + callCount++; + return callCount == 1 + ? new HttpResponseMessage(System.Net.HttpStatusCode.Unauthorized) + : SseTestData.OkResponse(SseTestData.DlrEvent(lineId, spanId, DateTimeOffset.UtcNow, 1.0)); + }); + + var tokenProvider = new CountingAccessTokenProvider(); + var client = new HeimdallStreamClient( + tokenProvider, + new HttpClient(handler) { BaseAddress = new Uri("https://fake-stream.example.com") }, + retryPolicy: ZeroDelay); + + using var cts = new CancellationTokenSource(); + var logMessages = new List(); + + HeimdallEventEnvelope? received = null; + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, cts.Token)) + { + received = envelope; + break; + } + + Assert.NotNull(received); + Assert.Equal(2, callCount); + Assert.Equal(2, tokenProvider.AcquireTokenCallCount); // Initial acquire + forced refresh after the 401. + Assert.Contains(logMessages, m => m.Contains("unauthorized", StringComparison.OrdinalIgnoreCase)); + } +} diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs new file mode 100644 index 0000000..8f0f22e --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs @@ -0,0 +1,124 @@ +using HeimdallPower.Api.Client.Stream; +using HeimdallPower.Api.Client.UnitTests.WhenStreaming.Fakes; +using HeimdallPower.Api.Client.UnitTests.WhenUsingResilienceExtensions.Fakes; + +namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming; + +/// +/// Verifies SSE event parsing behavior of - no network calls, no auth. +/// +[Trait("Category", "Unit")] +public class WhenParsingEvents +{ + private static HeimdallStreamClient CreateClient(HttpMessageHandler handler) => + new(new CountingAccessTokenProvider(), + new HttpClient(handler) { BaseAddress = new Uri("https://fake-stream.example.com") }, + retryPolicy: new StreamConnectionRetryPolicy(TimeSpan.Zero, TimeSpan.Zero)); + + [Fact] + public async Task ShouldYieldDeserializedDlrEvent_ForHeimdallDlrEventType() + { + var lineId = Guid.NewGuid(); + var spanId = Guid.NewGuid(); + var timestamp = new DateTimeOffset(2026, 1, 1, 12, 0, 0, TimeSpan.Zero); + + var handler = new CountingHttpMessageHandler(_ => + SseTestData.OkResponse(SseTestData.DlrEvent(lineId, spanId, timestamp, 123.4))); + + var client = CreateClient(handler); + using var cts = new CancellationTokenSource(); + + HeimdallEventEnvelope? received = null; + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + { + received = envelope; + break; + } + + Assert.NotNull(received); + Assert.Equal("Heimdall DLR", received!.Metric); + Assert.NotNull(received.HeimdallDlr); + Assert.Equal(lineId, received.HeimdallDlr!.AtLineId); + Assert.Equal(spanId, received.HeimdallDlr.AtSpanId); + Assert.Equal(123.4, received.HeimdallDlr.Value); + Assert.Equal(timestamp, received.HeimdallDlr.Timestamp); + Assert.False(received.HeimdallDlr.IsFallback); + } + + [Fact] + public async Task ShouldNotYieldHeartbeats_ButShouldLogThem() + { + var lineId = Guid.NewGuid(); + var spanId = Guid.NewGuid(); + var body = SseTestData.HeartbeatEvent + SseTestData.DlrEvent(lineId, spanId, DateTimeOffset.UtcNow, 1.0); + + var handler = new CountingHttpMessageHandler(_ => SseTestData.OkResponse(body)); + var client = CreateClient(handler); + using var cts = new CancellationTokenSource(); + var logMessages = new List(); + + var events = new List(); + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, cts.Token)) + { + events.Add(envelope); + break; + } + + Assert.Single(events); + Assert.Contains(logMessages, m => m.Contains("Heartbeat", StringComparison.OrdinalIgnoreCase)); + } + + [Fact] + public async Task ShouldSkipUnknownEventTypes() + { + var lineId = Guid.NewGuid(); + var spanId = Guid.NewGuid(); + var body = SseTestData.UnknownEvent + SseTestData.DlrEvent(lineId, spanId, DateTimeOffset.UtcNow, 1.0); + + var handler = new CountingHttpMessageHandler(_ => SseTestData.OkResponse(body)); + var client = CreateClient(handler); + using var cts = new CancellationTokenSource(); + + var events = new List(); + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + { + events.Add(envelope); + break; + } + + Assert.Single(events); + Assert.Equal("Heimdall DLR", events[0].Metric); + } + + [Fact] + public async Task ShouldRecoverAndReconnect_WhenDataIsMalformed() + { + var lineId = Guid.NewGuid(); + var spanId = Guid.NewGuid(); + + var callCount = 0; + var handler = new CountingHttpMessageHandler(_ => + { + callCount++; + return callCount == 1 + ? SseTestData.OkResponse(SseTestData.MalformedDlrEvent) + : SseTestData.OkResponse(SseTestData.DlrEvent(lineId, spanId, DateTimeOffset.UtcNow, 42.0)); + }); + + var client = CreateClient(handler); + using var cts = new CancellationTokenSource(); + var logMessages = new List(); + + HeimdallEventEnvelope? received = null; + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, cts.Token)) + { + received = envelope; + break; + } + + Assert.NotNull(received); + Assert.Equal(42.0, received!.HeimdallDlr!.Value); + Assert.Contains(logMessages, m => m.Contains("Stream error", StringComparison.OrdinalIgnoreCase)); + Assert.Equal(2, callCount); + } +} diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs new file mode 100644 index 0000000..cfae99f --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs @@ -0,0 +1,104 @@ +using HeimdallPower.Api.Client.Stream; +using HeimdallPower.Api.Client.UnitTests.WhenStreaming.Fakes; +using HeimdallPower.Api.Client.UnitTests.WhenUsingResilienceExtensions.Fakes; + +namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming; + +/// +/// Verifies 's transparent reconnection and cancellation behavior. +/// +[Trait("Category", "Unit")] +public class WhenReconnecting +{ + private static readonly StreamConnectionRetryPolicy ZeroDelay = new(TimeSpan.Zero, TimeSpan.Zero); + + private static HeimdallStreamClient CreateClient(HttpMessageHandler handler) => + new(new CountingAccessTokenProvider(), + new HttpClient(handler) { BaseAddress = new Uri("https://fake-stream.example.com") }, + retryPolicy: ZeroDelay); + + [Fact] + public async Task ShouldRecover_AfterTransientConnectionFailure() + { + var lineId = Guid.NewGuid(); + var spanId = Guid.NewGuid(); + + var callCount = 0; + var handler = new CountingHttpMessageHandler(_ => + { + callCount++; + if (callCount == 1) + throw new HttpRequestException("Connection refused"); + + return SseTestData.OkResponse(SseTestData.DlrEvent(lineId, spanId, DateTimeOffset.UtcNow, 7.0)); + }); + + var client = CreateClient(handler); + using var cts = new CancellationTokenSource(); + var logMessages = new List(); + + HeimdallEventEnvelope? received = null; + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, cts.Token)) + { + received = envelope; + break; + } + + Assert.NotNull(received); + Assert.Equal(2, callCount); + Assert.Contains(logMessages, m => m.Contains("Stream error", StringComparison.OrdinalIgnoreCase)); + } + + [Fact] + public async Task ShouldContinueYieldingEvents_AcrossReconnectBoundary() + { + var lineId = Guid.NewGuid(); + var spanId = Guid.NewGuid(); + + var callCount = 0; + var handler = new CountingHttpMessageHandler(_ => + { + callCount++; + // Each connect's canned body ends after one event, forcing a reconnect for the next. + return SseTestData.OkResponse(SseTestData.DlrEvent(lineId, spanId, DateTimeOffset.UtcNow, callCount)); + }); + + var client = CreateClient(handler); + using var cts = new CancellationTokenSource(); + + var values = new List(); + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + { + values.Add(envelope.HeimdallDlr!.Value); + if (values.Count == 3) + break; + } + + Assert.Equal([1.0, 2.0, 3.0], values); + } + + [Fact] + public async Task ShouldStopCleanly_WhenCancellationRequested() + { + var handler = new CountingHttpMessageHandler(_ => SseTestData.OkResponseNeverEnding(SseTestData.HeartbeatEvent)); + var client = CreateClient(handler); + + using var cts = new CancellationTokenSource(); + var events = new List(); + + var enumerationTask = Task.Run(async () => + { + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + { + events.Add(envelope); + } + }); + + cts.CancelAfter(TimeSpan.FromMilliseconds(100)); + + // Should complete (not throw) once cancelled - no exception surfaces to the caller. + await enumerationTask; + + Assert.Empty(events); + } +} From 49f55ae7496df67143f92b585604d5d84cf811eb Mon Sep 17 00:00:00 2001 From: Anders Johan Holmefjord Date: Thu, 27 Aug 2026 19:11:01 +0200 Subject: [PATCH 3/8] Refacoring and xml comments --- .../HeimdallStreamClientExtensions.cs | 12 +++++----- .../HeimdallApiEndpoints.cs | 1 + .../Lines/HeimdallDlrEvent.cs | 23 +++++++++++++++++++ .../Stream/HeimdallDlrEvent.cs | 11 --------- .../Stream/HeimdallEventEnvelope.cs | 12 ++++++++++ .../Stream/HeimdallStreamClient.cs | 20 +++++++++++++++- .../HeimdallStreamJsonSerializerOptions.cs | 7 ++++++ .../Stream/IHeimdallStreamClient.cs | 15 ++++++++---- .../Stream/StreamConnectionRetryPolicy.cs | 5 ++++ .../Api.Client.StreamExample.csproj | 14 ----------- .../Program.cs | 0 .../Stream.Client.Examples.csproj | 14 +++++++++++ 12 files changed, 97 insertions(+), 37 deletions(-) create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Lines/HeimdallDlrEvent.cs delete mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallDlrEvent.cs delete mode 100644 dotnet/examples/Api.Client.StreamExample/Api.Client.StreamExample.csproj rename dotnet/examples/{Api.Client.StreamExample => Stream.Client.Examples}/Program.cs (100%) create mode 100644 dotnet/examples/Stream.Client.Examples/Stream.Client.Examples.csproj diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs index 9db568b..d208ac4 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs @@ -10,13 +10,14 @@ namespace HeimdallPower.Api.Client.Extensions; public static class HeimdallStreamClientExtensions { /// - /// Adds the Heimdall Power stream client to the service collection. + /// Registers as a singleton, configured via . /// /// - /// Unlike , no standard resilience - /// handler is applied: its default total-request timeout would tear down the long-lived streaming - /// connection. Reconnection is instead handled internally by . + /// The stream connection reconnects with exponential backoff internally, handled by . /// + /// The service collection to add the client to. + /// Callback used to set the , such as client credentials and optional proxy settings. + /// The same service collection, for chaining. public static IServiceCollection AddHeimdallPowerStreamClient(this IServiceCollection services, Action configureOptions) { const string clientName = "HeimdallPowerStream"; @@ -26,8 +27,7 @@ public static IServiceCollection AddHeimdallPowerStreamClient(this IServiceColle services.AddHttpClient(clientName) .ConfigureHttpClient((_, client) => { - client.BaseAddress = new Uri("https://external-api.heimdallcloud.com"); - client.DefaultRequestHeaders.Add("Accept", "text/event-stream"); + client.BaseAddress = new Uri("https://stream-api.heimdallcloud.com"); }) .ConfigurePrimaryHttpMessageHandler(sp => { diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiEndpoints.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiEndpoints.cs index 7a0c61b..58226eb 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiEndpoints.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiEndpoints.cs @@ -6,6 +6,7 @@ namespace HeimdallPower.Api.Client; internal static class HeimdallApiEndpoints { public const string ApiUrl = "https://external-api.heimdallcloud.com"; + public const string StreamUrl = "https://stream-api.heimdallcloud.com"; private const string Policy = "B2C_1A_CLIENTCREDENTIALSFLOW"; private const string Instance = "https://hpadb2cprod.b2clogin.com"; private const string Domain = "hpadb2cprod.onmicrosoft.com"; diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Lines/HeimdallDlrEvent.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Lines/HeimdallDlrEvent.cs new file mode 100644 index 0000000..14934e2 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Lines/HeimdallDlrEvent.cs @@ -0,0 +1,23 @@ +namespace HeimdallPower.Api.Client.Stream.CapacityMonitoring.Lines; + +/// +/// A Heimdall Dynamic Line Rating (DLR) reading for a span, as carried in +/// when equals . +/// +/// The ID of the line the span belongs to. +/// The ID of the span the reading applies to. +/// The time the reading was taken. +/// The DLR value, in the unit specified by . +/// Whether this is a fallback rating, used when a live DLR value is unavailable. +public record HeimdallDlrEvent( + Guid AtLineId, + Guid AtSpanId, + DateTimeOffset Timestamp, + double Value, + bool IsFallback) +{ + /// + /// The value of for Heimdall DLR events. + /// + public const string MetricName = "Heimdall DLR"; +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallDlrEvent.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallDlrEvent.cs deleted file mode 100644 index b04c402..0000000 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallDlrEvent.cs +++ /dev/null @@ -1,11 +0,0 @@ -namespace HeimdallPower.Api.Client.Stream; - -public record HeimdallDlrEvent( - Guid AtLineId, - Guid AtSpanId, - DateTimeOffset Timestamp, - double Value, - bool IsFallback) -{ - public const string MetricName = "Heimdall DLR"; -} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs index b0e63b2..9069386 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs @@ -1,7 +1,15 @@ +using HeimdallPower.Api.Client.Stream.CapacityMonitoring.Lines; using System.Text.Json; namespace HeimdallPower.Api.Client.Stream; +/// +/// A single event received from the Heimdall Stream API. +/// +/// The version of the envelope schema. +/// The kind of event carried in , e.g. . +/// The unit of the value carried in . +/// The raw, metric-specific event payload. Use to access it as a typed . public record HeimdallEventEnvelope( string SchemaVersion, string Metric, @@ -10,6 +18,10 @@ public record HeimdallEventEnvelope( { private HeimdallDlrEvent? _heimdallDlr; + /// + /// deserialized as a , or if it cannot be deserialized as one. + /// Only meaningful when equals . + /// public HeimdallDlrEvent? HeimdallDlr { get diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs index 0489ad0..cf63982 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs @@ -1,3 +1,4 @@ +using HeimdallPower.Api.Client.Stream.CapacityMonitoring.Lines; using System.Net; using System.Net.ServerSentEvents; using System.Runtime.CompilerServices; @@ -5,6 +6,10 @@ namespace HeimdallPower.Api.Client.Stream; +/// +/// Default implementation, backed by an that +/// consumes the Heimdall Stream API's Server-Sent Events endpoint. +/// public class HeimdallStreamClient : IHeimdallStreamClient { private readonly HttpClient _httpClient; @@ -15,10 +20,15 @@ public class HeimdallStreamClient : IHeimdallStreamClient /// A client that lets you consume the Heimdall Stream API. /// Throws on non-transient errors. /// + /// The client ID used to authenticate with the Heimdall Power API. + /// The client secret used to authenticate with the Heimdall Power API. + /// An optional pre-configured . When omitted, one is created with the default stream base address. + /// Optional additional metadata to include in request headers. + /// An optional message handler used to route token acquisition through a proxy. public HeimdallStreamClient(string clientId, string clientSecret, HttpClient? httpClient = null, Dictionary? clientMetadata = null, HttpMessageHandler? proxyHandler = null) : this( new AccessTokenProvider(clientId, clientSecret, HeimdallApiEndpoints.Authority, HeimdallApiEndpoints.Scope, proxyHandler), - httpClient ?? new HttpClient { BaseAddress = new Uri(HeimdallApiEndpoints.ApiUrl) }, + httpClient ?? new HttpClient { BaseAddress = new Uri(HeimdallApiEndpoints.StreamUrl) }, clientMetadata) { } @@ -35,6 +45,10 @@ internal HeimdallStreamClient(IAccessTokenProvider accessTokenProvider, HttpClie /// Streams events, transparently reconnecting with exponential backoff when the /// connection drops or fails. The consumer only ever sees a continuous sequence of events. /// + /// The grid owner to receive events for, or to receive events for the authenticated grid owner. + /// Callback invoked with diagnostic messages (e.g. heartbeats, reconnect attempts). + /// A token used to stop receiving events and end the stream. + /// An asynchronous stream of Heimdall event envelopes that runs until cancelled. public async IAsyncEnumerable ReceiveAsync( Guid? gridOwnerId, Action infoLogger, @@ -97,6 +111,10 @@ public async IAsyncEnumerable ReceiveAsync( } } + /// + /// Opens a single SSE connection and yields the events read from it until the connection ends or fails. + /// Does not reconnect; that is handled by the caller, . + /// private async IAsyncEnumerable ConnectAndReadAsync( Guid? gridOwnerId, Action infoLogger, diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamJsonSerializerOptions.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamJsonSerializerOptions.cs index 301e0c5..d60e98e 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamJsonSerializerOptions.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamJsonSerializerOptions.cs @@ -3,6 +3,10 @@ namespace HeimdallPower.Api.Client.Stream; +/// +/// Shared, lazily-created used to (de)serialize Stream API payloads +/// (snake_case property names, case-insensitive matching, string enums). +/// internal static class HeimdallStreamJsonSerializerOptions { private static JsonSerializerOptions? _jsonOptions; @@ -18,6 +22,9 @@ private static JsonSerializerOptions CreateJsonOptions() return options; } + /// + /// The shared instance, created on first access. + /// public static JsonSerializerOptions Default { get diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs index 88a7e38..9b3f14f 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs @@ -15,11 +15,16 @@ namespace HeimdallPower.Api.Client.Stream; public interface IHeimdallStreamClient { /// - /// Receives Heimdall events from the Heimdall Stream API as an asynchronous stream. + /// Receives Heimdall events from the Heimdall Stream API as a continuous asynchronous stream. /// - /// The ID of the grid owner. - /// A logger action for informational messages. - /// A cancellation token. - /// An asynchronous stream of Heimdall event envelopes. + /// + /// The returned sequence does not complete on its own: it keeps yielding events, transparently + /// reconnecting behind the scenes, until is cancelled. Cancelling the + /// token ends enumeration gracefully rather than throwing. + /// + /// The grid owner to receive events for, or to receive events for all grid owners accessible to the authenticated client. + /// Callback invoked with diagnostic messages (e.g. heartbeats, reconnect attempts). Not used for the actual event data. + /// A token used to stop receiving events and end the stream. + /// An asynchronous stream of Heimdall event envelopes that runs until cancelled. IAsyncEnumerable ReceiveAsync(Guid? gridOwnerId, Action infoLogger, CancellationToken token); } diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs index 5641d53..dad92f1 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs @@ -8,6 +8,11 @@ internal sealed class StreamConnectionRetryPolicy(TimeSpan? initialDelay = null, private readonly TimeSpan _initialDelay = initialDelay ?? TimeSpan.FromSeconds(1); private readonly TimeSpan _maxDelay = maxDelay ?? TimeSpan.FromSeconds(30); + /// + /// Gets the delay to wait before the next reconnect attempt. + /// + /// The number of consecutive failed connection attempts so far. + /// The backoff delay, with jitter applied, capped at the configured maximum delay. public TimeSpan GetDelay(int failedAttempts) { if (failedAttempts <= 0) diff --git a/dotnet/examples/Api.Client.StreamExample/Api.Client.StreamExample.csproj b/dotnet/examples/Api.Client.StreamExample/Api.Client.StreamExample.csproj deleted file mode 100644 index 014037c..0000000 --- a/dotnet/examples/Api.Client.StreamExample/Api.Client.StreamExample.csproj +++ /dev/null @@ -1,14 +0,0 @@ - - - - Exe - net10.0 - enable - enable - - - - - - - diff --git a/dotnet/examples/Api.Client.StreamExample/Program.cs b/dotnet/examples/Stream.Client.Examples/Program.cs similarity index 100% rename from dotnet/examples/Api.Client.StreamExample/Program.cs rename to dotnet/examples/Stream.Client.Examples/Program.cs diff --git a/dotnet/examples/Stream.Client.Examples/Stream.Client.Examples.csproj b/dotnet/examples/Stream.Client.Examples/Stream.Client.Examples.csproj new file mode 100644 index 0000000..3250428 --- /dev/null +++ b/dotnet/examples/Stream.Client.Examples/Stream.Client.Examples.csproj @@ -0,0 +1,14 @@ + + + + Exe + net10.0 + enable + enable + + + + + + + From e1251e207b2d18975175e8bccba275d923d53a19 Mon Sep 17 00:00:00 2001 From: Anders Johan Holmefjord Date: Fri, 28 Aug 2026 11:05:50 +0200 Subject: [PATCH 4/8] Use correct name for stream example project --- dotnet/HeimdallPower.Api.Client.slnx | 2 +- dotnet/README.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/dotnet/HeimdallPower.Api.Client.slnx b/dotnet/HeimdallPower.Api.Client.slnx index 42f3630..665202a 100644 --- a/dotnet/HeimdallPower.Api.Client.slnx +++ b/dotnet/HeimdallPower.Api.Client.slnx @@ -6,7 +6,7 @@ - + diff --git a/dotnet/README.md b/dotnet/README.md index 33988ac..50face3 100644 --- a/dotnet/README.md +++ b/dotnet/README.md @@ -107,7 +107,7 @@ var streamClient = provider.GetRequiredService(); `AddHeimdallPowerStreamClient` reuses the same proxy configuration (`ProxyOptions`) as `AddHeimdallPowerApiClient`, but does **not** apply the standard resilience handler — reconnection for the long-lived stream connection is handled internally by `HeimdallStreamClient` instead. -See the full example in [`examples/Api.Client.StreamExample`](examples/Api.Client.StreamExample). +See the full example in [`examples/Stream.Client.Examples`](examples/Stream.Client.Examples). ## Error Handling From 16eeba27b9413f096aa824a744f4ed3813fcd182 Mon Sep 17 00:00:00 2001 From: Anders Johan Holmefjord Date: Fri, 28 Aug 2026 16:16:48 +0200 Subject: [PATCH 5/8] Stop client after max retries, and include quantity in url --- .../HeimdallStreamClientExtensions.cs | 2 +- .../HeimdallStreamClientOptions.cs | 7 ++ .../Stream/CapacityMonitoring/Quantity.cs | 27 ++++++++ .../Stream/HeimdallStreamClient.cs | 61 ++++++++++++----- .../Stream/IHeimdallStreamClient.cs | 10 ++- .../Stream/StreamConnectionRetryPolicy.cs | 39 ++++++++++- .../Stream/UrlBuilder.cs | 24 +++++++ .../Stream.Client.Examples/Program.cs | 26 ++++---- .../WhenStreaming/WhenAuthenticating.cs | 8 +-- .../WhenStreaming/WhenBuildingUrls.cs | 65 +++++++++++++++++++ .../WhenComputingRetryDelay.cs | 6 +- .../WhenStreaming/WhenParsingEvents.cs | 13 ++-- .../WhenStreaming/WhenReconnecting.cs | 24 +++++-- 13 files changed, 259 insertions(+), 53 deletions(-) create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Quantity.cs create mode 100644 dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/UrlBuilder.cs create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenBuildingUrls.cs rename dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/{WhenComputingRetryDelay => WhenStreaming}/WhenComputingRetryDelay.cs (80%) diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs index d208ac4..013db1c 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientExtensions.cs @@ -42,7 +42,7 @@ public static IServiceCollection AddHeimdallPowerStreamClient(this IServiceColle var httpClient = httpClientFactory.CreateClient(clientName); var proxyHandler = ProxyHandlerFactory.CreateHandler(options.Proxy); - return new HeimdallStreamClient(options.ClientId, options.ClientSecret, httpClient, options.ClientMetadata, proxyHandler); + return new HeimdallStreamClient(options.ClientId, options.ClientSecret, httpClient, options.ClientMetadata, proxyHandler, options.RetryPolicy); }); services.AddSingleton(sp => sp.GetRequiredService()); diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientOptions.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientOptions.cs index eefa2d6..7dc4970 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientOptions.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientOptions.cs @@ -1,3 +1,5 @@ +using HeimdallPower.Api.Client.Stream; + namespace HeimdallPower.Api.Client.Extensions; /// @@ -25,4 +27,9 @@ public class HeimdallStreamClientOptions /// are routed through the specified proxy. /// public ProxyOptions? Proxy { get; set; } + + /// + /// Optional configuration for the reconnect backoff behavior used when the stream connection drops. + /// + public StreamConnectionRetryPolicyOptions? RetryPolicy { get; set; } } diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Quantity.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Quantity.cs new file mode 100644 index 0000000..2fc66a1 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Quantity.cs @@ -0,0 +1,27 @@ +namespace HeimdallPower.Api.Client.Stream.CapacityMonitoring; + +/// +/// Controls which physical quantity is returned by a rating endpoint. +/// +public enum Quantity +{ + /// + /// Values in amperes (default). + /// + Current, + + /// + /// Values converted to three-phase apparent power in MVA using S = sqrt(3) * V * I / 1,000,000. + /// The line's operational voltage is used when set and positive; otherwise the nominal voltage is used. + /// + ApparentPower, +} + +internal static class QuantityExtensions +{ + public static string ToQueryValue(this Quantity quantity) => quantity switch + { + Quantity.ApparentPower => "apparent_power", + _ => "current", + }; +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs index cf63982..f8ea43c 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs @@ -1,3 +1,5 @@ +using HeimdallPower.Api.Client; +using HeimdallPower.Api.Client.Stream.CapacityMonitoring; using HeimdallPower.Api.Client.Stream.CapacityMonitoring.Lines; using System.Net; using System.Net.ServerSentEvents; @@ -25,11 +27,13 @@ public class HeimdallStreamClient : IHeimdallStreamClient /// An optional pre-configured . When omitted, one is created with the default stream base address. /// Optional additional metadata to include in request headers. /// An optional message handler used to route token acquisition through a proxy. - public HeimdallStreamClient(string clientId, string clientSecret, HttpClient? httpClient = null, Dictionary? clientMetadata = null, HttpMessageHandler? proxyHandler = null) + /// Optional configuration for the reconnect backoff behavior. + public HeimdallStreamClient(string clientId, string clientSecret, HttpClient? httpClient = null, Dictionary? clientMetadata = null, HttpMessageHandler? proxyHandler = null, StreamConnectionRetryPolicyOptions? retryPolicyOptions = null) : this( new AccessTokenProvider(clientId, clientSecret, HeimdallApiEndpoints.Authority, HeimdallApiEndpoints.Scope, proxyHandler), httpClient ?? new HttpClient { BaseAddress = new Uri(HeimdallApiEndpoints.StreamUrl) }, - clientMetadata) + clientMetadata, + new StreamConnectionRetryPolicy(retryPolicyOptions)) { } @@ -42,25 +46,31 @@ internal HeimdallStreamClient(IAccessTokenProvider accessTokenProvider, HttpClie } /// - /// Streams events, transparently reconnecting with exponential backoff when the - /// connection drops or fails. The consumer only ever sees a continuous sequence of events. + /// Streams events for the specified grid owner with the specified quantity, transparently reconnecting with exponential backoff when the + /// connection drops or fails. The consumer only ever sees a continuous sequence of events. All event types are returned in a single stream, and the consumer can filter them by type if desired. + /// The stream runs until the provided cancellation token is cancelled, or an unrecoverable error occurs. /// /// The grid owner to receive events for, or to receive events for the authenticated grid owner. - /// Callback invoked with diagnostic messages (e.g. heartbeats, reconnect attempts). + /// The physical quantity to receive events for, Current (default) or ApparentPower. + /// Callback invoked with diagnostic messages (e.g. errors, reconnect attempts), or if no logging is desired. Not used for any event data. + /// Callback invoked with trace messages (e.g. heartbeats, received events), or if no logging is desired. Does not log the detailed event data. /// A token used to stop receiving events and end the stream. /// An asynchronous stream of Heimdall event envelopes that runs until cancelled. public async IAsyncEnumerable ReceiveAsync( Guid? gridOwnerId, - Action infoLogger, - [EnumeratorCancellation] CancellationToken token) + Quantity quantity = Quantity.Current, + Action? infoLogger = null, + Action? traceLogger = null, + [EnumeratorCancellation] CancellationToken token = default) { var failedAttempts = 0; + infoLogger?.Invoke("Starting Heimdall stream..."); while (!token.IsCancellationRequested) { // The inner iterator is driven manually so that try/catch can wrap MoveNextAsync // without ever wrapping a yield return (which the compiler forbids). - await using var enumerator = ConnectAndReadAsync(gridOwnerId, infoLogger, token) + await using var enumerator = ConnectAndReadAsync(gridOwnerId, quantity, traceLogger, token) .GetAsyncEnumerator(token); while (true) @@ -82,14 +92,16 @@ public async IAsyncEnumerable ReceiveAsync( catch (UnauthorizedAccessException) { failedAttempts++; - infoLogger($"Stream error: unauthorized. Refreshing token and reconnecting... (attempt #{failedAttempts})"); + infoLogger?.Invoke($"Stream error: unauthorized. Refreshing token and reconnecting... (attempt #{failedAttempts})"); await _tokenRefresher.ForceRefreshAsync(token); break; } catch (Exception ex) { + if (ex is HeimdallApiException) throw; + failedAttempts++; - infoLogger($"Stream error: {ex.Message}. Reconnecting... (attempt #{failedAttempts})"); + infoLogger?.Invoke($"Stream error: {ex.Message}. Reconnecting... (attempt #{failedAttempts})"); break; } @@ -102,7 +114,15 @@ public async IAsyncEnumerable ReceiveAsync( try { // Should we break the loop after a number of failed attempts? - await Task.Delay(_retryPolicy.GetDelay(failedAttempts), token); + if (_retryPolicy.ShouldRetry(failedAttempts)) + { + await Task.Delay(_retryPolicy.GetDelay(failedAttempts), token); + } + else + { + infoLogger?.Invoke($"Stream error: exceeded maximum retry attempts ({failedAttempts}). Stopping stream."); + yield break; + } } catch (OperationCanceledException) { @@ -117,21 +137,31 @@ public async IAsyncEnumerable ReceiveAsync( /// private async IAsyncEnumerable ConnectAndReadAsync( Guid? gridOwnerId, - Action infoLogger, + Quantity quantity, + Action? traceLogger, [EnumeratorCancellation] CancellationToken token) { await _tokenRefresher.EnsureFreshTokenAsync(token); + var url = UrlBuilder.BuildStreamUrl(version: 1, gridOwnerId, quantity); + using var request = new HttpRequestMessage( HttpMethod.Get, - $"/v1/stream?gridownerid={gridOwnerId}"); + url); using var response = await _httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, token); if (response.StatusCode == HttpStatusCode.Unauthorized) throw new UnauthorizedAccessException("Unauthorized access. Please check your credentials."); - response.EnsureSuccessStatusCode(); + if (!response.IsSuccessStatusCode) + { + var requestUrl = response.RequestMessage?.RequestUri?.ToString() ?? string.Empty; + throw new HeimdallApiException( + $"Stream request failed with status code {(int)response.StatusCode} {response.StatusCode}.", + response.StatusCode, + requestUrl); + } await using var stream = await response.Content.ReadAsStreamAsync(token); @@ -139,7 +169,7 @@ private async IAsyncEnumerable ConnectAndReadAsync( { if (item.EventType == "heartbeat") { - infoLogger("Heartbeat received..");// This is debug information, should check if the user wants to log it or not + traceLogger?.Invoke("Heartbeat received."); continue; } @@ -153,6 +183,7 @@ private async IAsyncEnumerable ConnectAndReadAsync( if (envelope is null) continue; + traceLogger?.Invoke("Received event: " + envelope.Metric); yield return envelope; } } diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs index 9b3f14f..2307d84 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs @@ -1,3 +1,5 @@ +using HeimdallPower.Api.Client.Stream.CapacityMonitoring; + namespace HeimdallPower.Api.Client.Stream; /// @@ -22,9 +24,11 @@ public interface IHeimdallStreamClient /// reconnecting behind the scenes, until is cancelled. Cancelling the /// token ends enumeration gracefully rather than throwing. /// - /// The grid owner to receive events for, or to receive events for all grid owners accessible to the authenticated client. - /// Callback invoked with diagnostic messages (e.g. heartbeats, reconnect attempts). Not used for the actual event data. + /// The grid owner to receive events for, or to receive events for the authenticated grid owner. + /// The physical quantity to receive events for, Current (default) or ApparentPower. + /// Callback invoked with diagnostic messages (e.g. errors, reconnect attempts), or if no logging is desired. Not used for any event data. + /// Callback invoked with trace messages (e.g. heartbeats, received events), or if no logging is desired. Does not log the detailed event data. /// A token used to stop receiving events and end the stream. /// An asynchronous stream of Heimdall event envelopes that runs until cancelled. - IAsyncEnumerable ReceiveAsync(Guid? gridOwnerId, Action infoLogger, CancellationToken token); + IAsyncEnumerable ReceiveAsync(Guid? gridOwnerId, Quantity quantity = Quantity.Current, Action? infoLogger = null, Action? traceLogger = null, CancellationToken token = default); } diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs index dad92f1..14b2bba 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs @@ -1,12 +1,35 @@ namespace HeimdallPower.Api.Client.Stream; +/// +/// Configures the exponential backoff (with jitter) used between stream reconnect attempts. +/// +public sealed class StreamConnectionRetryPolicyOptions +{ + /// + /// The delay before the first reconnect attempt. Defaults to 1 second. + /// + public TimeSpan? InitialDelay { get; init; } + + /// + /// The maximum delay between reconnect attempts. Defaults to 30 seconds. + /// + public TimeSpan? MaxDelay { get; init; } + + /// + /// The maximum number of consecutive reconnect attempts before giving up. Defaults to 5. + /// + public int? MaxRetries { get; init; } +} + /// /// Computes the exponential backoff (with jitter) delay between stream reconnect attempts. +/// Also holds the configured maximum number of retries before giving up. /// -internal sealed class StreamConnectionRetryPolicy(TimeSpan? initialDelay = null, TimeSpan? maxDelay = null) +internal sealed class StreamConnectionRetryPolicy(StreamConnectionRetryPolicyOptions? options = null) { - private readonly TimeSpan _initialDelay = initialDelay ?? TimeSpan.FromSeconds(1); - private readonly TimeSpan _maxDelay = maxDelay ?? TimeSpan.FromSeconds(30); + private readonly TimeSpan _initialDelay = options?.InitialDelay ?? TimeSpan.FromSeconds(1); + private readonly TimeSpan _maxDelay = options?.MaxDelay ?? TimeSpan.FromSeconds(30); + private readonly int _maxRetries = options?.MaxRetries ?? 5; /// /// Gets the delay to wait before the next reconnect attempt. @@ -24,4 +47,14 @@ public TimeSpan GetDelay(int failedAttempts) // Jitter avoids a thundering herd of clients reconnecting simultaneously. return capped * (0.8 + (Random.Shared.NextDouble() * 0.4)); } + + /// + /// Determines whether another retry attempt should be made based on the number of failed attempts. + /// + /// The number of consecutive failed connection attempts so far. + /// true if another retry should be attempted; otherwise, false. + public bool ShouldRetry(int failedAttempts) + { + return failedAttempts <= _maxRetries; + } } diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/UrlBuilder.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/UrlBuilder.cs new file mode 100644 index 0000000..eddcad8 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/UrlBuilder.cs @@ -0,0 +1,24 @@ +using System.Collections.Specialized; +using HeimdallPower.Api.Client.Stream.CapacityMonitoring; + +namespace HeimdallPower.Api.Client.Stream; + +internal static class UrlBuilder +{ + private const string Stream = "stream"; + + public static string BuildStreamUrl(int version, Guid? gridOwnerId, Quantity quantity = Quantity.Current) + { + var queryParams = new NameValueCollection(); + + if (gridOwnerId.HasValue) + queryParams.AddQueryParam("gridownerid", gridOwnerId.Value.ToString()); + + if (quantity != Quantity.Current) + queryParams.AddQueryParam("quantity", quantity.ToQueryValue()); + + var url = $"/v{version}/{Stream}"; + + return queryParams.Count > 0 ? url + queryParams.ToQueryString() : url; + } +} diff --git a/dotnet/examples/Stream.Client.Examples/Program.cs b/dotnet/examples/Stream.Client.Examples/Program.cs index 9747897..4ef1e5e 100644 --- a/dotnet/examples/Stream.Client.Examples/Program.cs +++ b/dotnet/examples/Stream.Client.Examples/Program.cs @@ -1,4 +1,5 @@ using HeimdallPower.Api.Client.Stream; +using HeimdallPower.Api.Client.Stream.CapacityMonitoring; // Configuration setup const string clientId = "insert-your-client-id-here"; @@ -19,22 +20,17 @@ Console.WriteLine("Listening for events. Press Ctrl+C to stop."); -try +// Pass a specific grid owner ID to only receive events for that grid owner. +await foreach (var envelope in streamClient.ReceiveAsync(gridOwnerId: null, quantity: Quantity.Current, infoLogger: Console.WriteLine, traceLogger: Console.WriteLine, token: cts.Token)) { - // Pass a specific grid owner ID to only receive events for that grid owner. - await foreach (var envelope in streamClient.ReceiveAsync(gridOwnerId: null, infoLogger: Console.WriteLine, cts.Token)) + if (envelope.HeimdallDlr is { } dlr) { - if (envelope.HeimdallDlr is { } dlr) - { - Console.WriteLine($"- Heimdall DLR: {dlr.Value} {envelope.Unit} for line {dlr.AtLineId} at {dlr.Timestamp} (IsFallback={dlr.IsFallback})"); - } - else - { - Console.WriteLine($"- {envelope.Metric}: {envelope.Data}"); - } + Console.WriteLine($"- Heimdall DLR: {dlr.Value} {envelope.Unit} for line {dlr.AtLineId} at {dlr.Timestamp} (IsFallback={dlr.IsFallback})"); + } + else + { + Console.WriteLine($"- {envelope.Metric}: {envelope.Data}"); } } -catch (OperationCanceledException) -{ - Console.WriteLine("Stream stopped."); -} + +Console.WriteLine("Stream stopped."); diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs index 946574b..6c849a5 100644 --- a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs @@ -11,7 +11,7 @@ namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming; [Trait("Category", "Unit")] public class WhenAuthenticating { - private static readonly StreamConnectionRetryPolicy ZeroDelay = new(TimeSpan.Zero, TimeSpan.Zero); + private static readonly StreamConnectionRetryPolicy ZeroDelay = new(new StreamConnectionRetryPolicyOptions { InitialDelay = TimeSpan.Zero, MaxDelay = TimeSpan.Zero }); [Fact] public async Task ShouldAttachAuthAndClientHeaders_BeforeFirstConnect() @@ -29,7 +29,7 @@ public async Task ShouldAttachAuthAndClientHeaders_BeforeFirstConnect() retryPolicy: ZeroDelay); using var cts = new CancellationTokenSource(); - await foreach (var _ in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + await foreach (var _ in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) { break; } @@ -57,7 +57,7 @@ public async Task ShouldReuseToken_AcrossReconnects_WhileStillFresh() using var cts = new CancellationTokenSource(); var receivedCount = 0; - await foreach (var _ in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + await foreach (var _ in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) { receivedCount++; if (receivedCount == 2) @@ -93,7 +93,7 @@ public async Task ShouldForceRefreshToken_OnUnauthorizedResponse() var logMessages = new List(); HeimdallEventEnvelope? received = null; - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, cts.Token)) + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, token: cts.Token)) { received = envelope; break; diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenBuildingUrls.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenBuildingUrls.cs new file mode 100644 index 0000000..2bae137 --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenBuildingUrls.cs @@ -0,0 +1,65 @@ +using HeimdallPower.Api.Client.Stream.CapacityMonitoring; +using StreamUrlBuilder = HeimdallPower.Api.Client.Stream.UrlBuilder; + +namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming; + +/// +/// Verifies the URL built for the Heimdall Stream API's SSE endpoint. +/// +[Trait("Category", "Unit")] +public class WhenBuildingUrls +{ + [Fact] + public void ShouldBuildUrlWithoutQueryString_WhenGridOwnerIdIsNullAndQuantityIsDefault() + { + var url = StreamUrlBuilder.BuildStreamUrl(version: 1, gridOwnerId: null); + + Assert.Equal("/v1/stream", url); + } + + [Fact] + public void ShouldIncludeGridOwnerId_WhenProvided() + { + var gridOwnerId = Guid.NewGuid(); + + var url = StreamUrlBuilder.BuildStreamUrl(version: 1, gridOwnerId: gridOwnerId); + + Assert.Equal($"/v1/stream?gridownerid={gridOwnerId}", url); + } + + [Fact] + public void ShouldIncludeQuantity_WhenNotCurrent() + { + var url = StreamUrlBuilder.BuildStreamUrl(version: 1, gridOwnerId: null, quantity: Quantity.ApparentPower); + + Assert.Equal("/v1/stream?quantity=apparent_power", url); + } + + [Fact] + public void ShouldOmitQuantity_WhenCurrent() + { + var gridOwnerId = Guid.NewGuid(); + + var url = StreamUrlBuilder.BuildStreamUrl(version: 1, gridOwnerId: gridOwnerId, quantity: Quantity.Current); + + Assert.Equal($"/v1/stream?gridownerid={gridOwnerId}", url); + } + + [Fact] + public void ShouldIncludeBothGridOwnerIdAndQuantity_WhenBothProvided() + { + var gridOwnerId = Guid.NewGuid(); + + var url = StreamUrlBuilder.BuildStreamUrl(version: 1, gridOwnerId: gridOwnerId, quantity: Quantity.ApparentPower); + + Assert.Equal($"/v1/stream?gridownerid={gridOwnerId}&quantity=apparent_power", url); + } + + [Fact] + public void ShouldUseGivenVersion() + { + var url = StreamUrlBuilder.BuildStreamUrl(version: 2, gridOwnerId: null); + + Assert.Equal("/v2/stream", url); + } +} diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenComputingRetryDelay/WhenComputingRetryDelay.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenComputingRetryDelay.cs similarity index 80% rename from dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenComputingRetryDelay/WhenComputingRetryDelay.cs rename to dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenComputingRetryDelay.cs index f33eb7a..d8fb5a2 100644 --- a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenComputingRetryDelay/WhenComputingRetryDelay.cs +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenComputingRetryDelay.cs @@ -1,10 +1,12 @@ using HeimdallPower.Api.Client.Stream; -namespace HeimdallPower.Api.Client.UnitTests.WhenComputingRetryDelay; +namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming; /// /// Data-driven tests for 's exponential backoff + jitter math. /// Jitter is random, so assertions check bounds rather than exact values. +/// Also verifies that the policy never exceeds the configured maximum delay, even after many failed attempts. +/// And checks the maximum limit of retries, which is a safety feature to prevent infinite retry loops in case of persistent failures. /// [Trait("Category", "Unit")] public class WhenComputingRetryDelay @@ -12,7 +14,7 @@ public class WhenComputingRetryDelay private static readonly TimeSpan InitialDelay = TimeSpan.FromSeconds(1); private static readonly TimeSpan MaxDelay = TimeSpan.FromSeconds(30); - private readonly StreamConnectionRetryPolicy _policy = new(InitialDelay, MaxDelay); + private readonly StreamConnectionRetryPolicy _policy = new(new StreamConnectionRetryPolicyOptions { InitialDelay = InitialDelay, MaxDelay = MaxDelay }); [Theory] [InlineData(0)] diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs index 8f0f22e..39bd6eb 100644 --- a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs @@ -13,7 +13,7 @@ public class WhenParsingEvents private static HeimdallStreamClient CreateClient(HttpMessageHandler handler) => new(new CountingAccessTokenProvider(), new HttpClient(handler) { BaseAddress = new Uri("https://fake-stream.example.com") }, - retryPolicy: new StreamConnectionRetryPolicy(TimeSpan.Zero, TimeSpan.Zero)); + retryPolicy: new StreamConnectionRetryPolicy(new StreamConnectionRetryPolicyOptions { InitialDelay = TimeSpan.Zero, MaxDelay = TimeSpan.Zero })); [Fact] public async Task ShouldYieldDeserializedDlrEvent_ForHeimdallDlrEventType() @@ -29,7 +29,7 @@ public async Task ShouldYieldDeserializedDlrEvent_ForHeimdallDlrEventType() using var cts = new CancellationTokenSource(); HeimdallEventEnvelope? received = null; - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) { received = envelope; break; @@ -56,16 +56,17 @@ public async Task ShouldNotYieldHeartbeats_ButShouldLogThem() var client = CreateClient(handler); using var cts = new CancellationTokenSource(); var logMessages = new List(); + var traceMessages = new List(); var events = new List(); - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, cts.Token)) + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, traceLogger: traceMessages.Add, token: cts.Token)) { events.Add(envelope); break; } Assert.Single(events); - Assert.Contains(logMessages, m => m.Contains("Heartbeat", StringComparison.OrdinalIgnoreCase)); + Assert.Contains(traceMessages, m => m.Contains("Heartbeat", StringComparison.OrdinalIgnoreCase)); } [Fact] @@ -80,7 +81,7 @@ public async Task ShouldSkipUnknownEventTypes() using var cts = new CancellationTokenSource(); var events = new List(); - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) { events.Add(envelope); break; @@ -110,7 +111,7 @@ public async Task ShouldRecoverAndReconnect_WhenDataIsMalformed() var logMessages = new List(); HeimdallEventEnvelope? received = null; - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, cts.Token)) + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, token: cts.Token)) { received = envelope; break; diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs index cfae99f..e6f6f5b 100644 --- a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs @@ -10,7 +10,8 @@ namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming; [Trait("Category", "Unit")] public class WhenReconnecting { - private static readonly StreamConnectionRetryPolicy ZeroDelay = new(TimeSpan.Zero, TimeSpan.Zero); + private const int _maxRetriesInTest = 3; + private static readonly StreamConnectionRetryPolicy ZeroDelay = new(new StreamConnectionRetryPolicyOptions { InitialDelay = TimeSpan.Zero, MaxDelay = TimeSpan.Zero, MaxRetries = _maxRetriesInTest }); private static HeimdallStreamClient CreateClient(HttpMessageHandler handler) => new(new CountingAccessTokenProvider(), @@ -38,7 +39,7 @@ public async Task ShouldRecover_AfterTransientConnectionFailure() var logMessages = new List(); HeimdallEventEnvelope? received = null; - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, cts.Token)) + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, token: cts.Token)) { received = envelope; break; @@ -67,7 +68,7 @@ public async Task ShouldContinueYieldingEvents_AcrossReconnectBoundary() using var cts = new CancellationTokenSource(); var values = new List(); - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) { values.Add(envelope.HeimdallDlr!.Value); if (values.Count == 3) @@ -88,7 +89,7 @@ public async Task ShouldStopCleanly_WhenCancellationRequested() var enumerationTask = Task.Run(async () => { - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, cts.Token)) + await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) { events.Add(envelope); } @@ -101,4 +102,19 @@ public async Task ShouldStopCleanly_WhenCancellationRequested() Assert.Empty(events); } + + + [Fact] + public void ShouldRetryMaxTimes_ThenStop() + { + for (var failedAttempts = 0; failedAttempts <= _maxRetriesInTest; failedAttempts++) + { + bool shouldRetry = ZeroDelay.ShouldRetry(failedAttempts); + + Assert.True(shouldRetry, $"ShouldRetry returned false for failedAttempts={failedAttempts}"); + } + + bool shouldRetryAfterThree = ZeroDelay.ShouldRetry(_maxRetriesInTest + 1); + Assert.False(shouldRetryAfterThree, $"ShouldRetry returned true for failedAttempts={_maxRetriesInTest + 1}"); + } } From 7f5629543223706793c262e935410cc4cfd8e017 Mon Sep 17 00:00:00 2001 From: Anders Johan Holmefjord Date: Mon, 31 Aug 2026 09:28:44 +0200 Subject: [PATCH 6/8] Throw exception after max attempts, plus comment improvements --- .../Stream/HeimdallStreamClient.cs | 18 ++++++++++++------ dotnet/README.md | 2 +- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs index f8ea43c..d87375f 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs @@ -20,8 +20,11 @@ public class HeimdallStreamClient : IHeimdallStreamClient /// /// A client that lets you consume the Heimdall Stream API. - /// Throws on non-transient errors. /// + /// + /// This client will automatically attempt to reconnect with exponential backoff when transient errors occur. + /// The number of retry attempts is determined by the . + /// /// The client ID used to authenticate with the Heimdall Power API. /// The client secret used to authenticate with the Heimdall Power API. /// An optional pre-configured . When omitted, one is created with the default stream base address. @@ -50,12 +53,13 @@ internal HeimdallStreamClient(IAccessTokenProvider accessTokenProvider, HttpClie /// connection drops or fails. The consumer only ever sees a continuous sequence of events. All event types are returned in a single stream, and the consumer can filter them by type if desired. /// The stream runs until the provided cancellation token is cancelled, or an unrecoverable error occurs. /// + /// Thrown on non-transient errors after exhausting all retry attempts. /// The grid owner to receive events for, or to receive events for the authenticated grid owner. /// The physical quantity to receive events for, Current (default) or ApparentPower. /// Callback invoked with diagnostic messages (e.g. errors, reconnect attempts), or if no logging is desired. Not used for any event data. /// Callback invoked with trace messages (e.g. heartbeats, received events), or if no logging is desired. Does not log the detailed event data. /// A token used to stop receiving events and end the stream. - /// An asynchronous stream of Heimdall event envelopes that runs until cancelled. + /// An asynchronous stream of that runs until cancelled. public async IAsyncEnumerable ReceiveAsync( Guid? gridOwnerId, Quantity quantity = Quantity.Current, @@ -98,9 +102,12 @@ public async IAsyncEnumerable ReceiveAsync( } catch (Exception ex) { - if (ex is HeimdallApiException) throw; - failedAttempts++; + + if (ex is HeimdallApiException + && !_retryPolicy.ShouldRetry(failedAttempts)) + throw; + infoLogger?.Invoke($"Stream error: {ex.Message}. Reconnecting... (attempt #{failedAttempts})"); break; } @@ -113,7 +120,6 @@ public async IAsyncEnumerable ReceiveAsync( try { - // Should we break the loop after a number of failed attempts? if (_retryPolicy.ShouldRetry(failedAttempts)) { await Task.Delay(_retryPolicy.GetDelay(failedAttempts), token); @@ -143,7 +149,7 @@ private async IAsyncEnumerable ConnectAndReadAsync( { await _tokenRefresher.EnsureFreshTokenAsync(token); - var url = UrlBuilder.BuildStreamUrl(version: 1, gridOwnerId, quantity); + string url = UrlBuilder.BuildStreamUrl(version: 1, gridOwnerId, quantity); using var request = new HttpRequestMessage( HttpMethod.Get, diff --git a/dotnet/README.md b/dotnet/README.md index 50face3..53dd422 100644 --- a/dotnet/README.md +++ b/dotnet/README.md @@ -84,7 +84,7 @@ using HeimdallPower.Api.Client.Stream; var streamClient = new HeimdallStreamClient(clientId, clientSecret); -await foreach (var envelope in streamClient.ReceiveAsync(gridOwnerId: null, infoLogger: Console.WriteLine, cancellationToken)) +await foreach (var envelope in streamClient.ReceiveAsync()) { if (envelope.HeimdallDlr is { } dlr) { From 7286e4d170f4acc3cf97adbd49617071df94a37c Mon Sep 17 00:00:00 2001 From: Anders Johan Holmefjord Date: Tue, 1 Sep 2026 14:28:58 +0200 Subject: [PATCH 7/8] Use correct event type name in parsing --- .../Stream/CapacityMonitoring/Lines/HeimdallDlrEvent.cs | 6 +++--- .../Stream/HeimdallEventEnvelope.cs | 4 ++-- .../HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs | 2 +- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Lines/HeimdallDlrEvent.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Lines/HeimdallDlrEvent.cs index 14934e2..0800700 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Lines/HeimdallDlrEvent.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Lines/HeimdallDlrEvent.cs @@ -2,7 +2,7 @@ namespace HeimdallPower.Api.Client.Stream.CapacityMonitoring.Lines; /// /// A Heimdall Dynamic Line Rating (DLR) reading for a span, as carried in -/// when equals . +/// when equals . /// /// The ID of the line the span belongs to. /// The ID of the span the reading applies to. @@ -17,7 +17,7 @@ public record HeimdallDlrEvent( bool IsFallback) { /// - /// The value of for Heimdall DLR events. + /// The value of SSE 'event:' field for Heimdall DLR events. /// - public const string MetricName = "Heimdall DLR"; + public const string EventName = "heimdall_dlr"; } diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs index 9069386..774db70 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs @@ -7,7 +7,7 @@ namespace HeimdallPower.Api.Client.Stream; /// A single event received from the Heimdall Stream API. /// /// The version of the envelope schema. -/// The kind of event carried in , e.g. . +/// The kind of event carried in , e.g. . /// The unit of the value carried in . /// The raw, metric-specific event payload. Use to access it as a typed . public record HeimdallEventEnvelope( @@ -20,7 +20,7 @@ public record HeimdallEventEnvelope( /// /// deserialized as a , or if it cannot be deserialized as one. - /// Only meaningful when equals . + /// Only meaningful when equals "Heimdall DLR" />. /// public HeimdallDlrEvent? HeimdallDlr { diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs index d87375f..f936bbc 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs @@ -182,7 +182,7 @@ private async IAsyncEnumerable ConnectAndReadAsync( if (string.IsNullOrWhiteSpace(item.Data)) continue; - if (item.EventType == HeimdallDlrEvent.MetricName) + if (item.EventType == HeimdallDlrEvent.EventName) { HeimdallEventEnvelope? envelope = JsonSerializer.Deserialize(item.Data, HeimdallStreamJsonSerializerOptions.Default); From a64ec61291d359f7695d9a688f1d1265bd03ef98 Mon Sep 17 00:00:00 2001 From: Anders Johan Holmefjord Date: Thu, 3 Sep 2026 11:08:13 +0200 Subject: [PATCH 8/8] Remove grid owner id from url, update tests to fit --- .../Stream/HeimdallStreamClient.cs | 9 ++--- .../Stream/IHeimdallStreamClient.cs | 3 +- .../Stream/UrlBuilder.cs | 5 +-- .../Stream.Client.Examples/Program.cs | 3 +- .../WhenStreaming/Fakes/SseTestData.cs | 5 +-- .../WhenStreaming/WhenAuthenticating.cs | 6 ++-- .../WhenStreaming/WhenBuildingUrls.cs | 34 ++++--------------- .../WhenStreaming/WhenParsingEvents.cs | 9 ++--- .../WhenStreaming/WhenReconnecting.cs | 6 ++-- 9 files changed, 26 insertions(+), 54 deletions(-) diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs index f936bbc..356d73c 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs @@ -49,19 +49,17 @@ internal HeimdallStreamClient(IAccessTokenProvider accessTokenProvider, HttpClie } /// - /// Streams events for the specified grid owner with the specified quantity, transparently reconnecting with exponential backoff when the + /// Streams events for the authenticated grid owner with the specified quantity, transparently reconnecting with exponential backoff when the /// connection drops or fails. The consumer only ever sees a continuous sequence of events. All event types are returned in a single stream, and the consumer can filter them by type if desired. /// The stream runs until the provided cancellation token is cancelled, or an unrecoverable error occurs. /// /// Thrown on non-transient errors after exhausting all retry attempts. - /// The grid owner to receive events for, or to receive events for the authenticated grid owner. /// The physical quantity to receive events for, Current (default) or ApparentPower. /// Callback invoked with diagnostic messages (e.g. errors, reconnect attempts), or if no logging is desired. Not used for any event data. /// Callback invoked with trace messages (e.g. heartbeats, received events), or if no logging is desired. Does not log the detailed event data. /// A token used to stop receiving events and end the stream. /// An asynchronous stream of that runs until cancelled. public async IAsyncEnumerable ReceiveAsync( - Guid? gridOwnerId, Quantity quantity = Quantity.Current, Action? infoLogger = null, Action? traceLogger = null, @@ -74,7 +72,7 @@ public async IAsyncEnumerable ReceiveAsync( { // The inner iterator is driven manually so that try/catch can wrap MoveNextAsync // without ever wrapping a yield return (which the compiler forbids). - await using var enumerator = ConnectAndReadAsync(gridOwnerId, quantity, traceLogger, token) + await using var enumerator = ConnectAndReadAsync(quantity, traceLogger, token) .GetAsyncEnumerator(token); while (true) @@ -142,14 +140,13 @@ public async IAsyncEnumerable ReceiveAsync( /// Does not reconnect; that is handled by the caller, . /// private async IAsyncEnumerable ConnectAndReadAsync( - Guid? gridOwnerId, Quantity quantity, Action? traceLogger, [EnumeratorCancellation] CancellationToken token) { await _tokenRefresher.EnsureFreshTokenAsync(token); - string url = UrlBuilder.BuildStreamUrl(version: 1, gridOwnerId, quantity); + string url = UrlBuilder.BuildStreamUrl(version: 1, quantity); using var request = new HttpRequestMessage( HttpMethod.Get, diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs index 2307d84..d7882f7 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs @@ -24,11 +24,10 @@ public interface IHeimdallStreamClient /// reconnecting behind the scenes, until is cancelled. Cancelling the /// token ends enumeration gracefully rather than throwing. /// - /// The grid owner to receive events for, or to receive events for the authenticated grid owner. /// The physical quantity to receive events for, Current (default) or ApparentPower. /// Callback invoked with diagnostic messages (e.g. errors, reconnect attempts), or if no logging is desired. Not used for any event data. /// Callback invoked with trace messages (e.g. heartbeats, received events), or if no logging is desired. Does not log the detailed event data. /// A token used to stop receiving events and end the stream. /// An asynchronous stream of Heimdall event envelopes that runs until cancelled. - IAsyncEnumerable ReceiveAsync(Guid? gridOwnerId, Quantity quantity = Quantity.Current, Action? infoLogger = null, Action? traceLogger = null, CancellationToken token = default); + IAsyncEnumerable ReceiveAsync(Quantity quantity = Quantity.Current, Action? infoLogger = null, Action? traceLogger = null, CancellationToken token = default); } diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/UrlBuilder.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/UrlBuilder.cs index eddcad8..c926342 100644 --- a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/UrlBuilder.cs +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/UrlBuilder.cs @@ -7,13 +7,10 @@ internal static class UrlBuilder { private const string Stream = "stream"; - public static string BuildStreamUrl(int version, Guid? gridOwnerId, Quantity quantity = Quantity.Current) + public static string BuildStreamUrl(int version, Quantity quantity = Quantity.Current) { var queryParams = new NameValueCollection(); - if (gridOwnerId.HasValue) - queryParams.AddQueryParam("gridownerid", gridOwnerId.Value.ToString()); - if (quantity != Quantity.Current) queryParams.AddQueryParam("quantity", quantity.ToQueryValue()); diff --git a/dotnet/examples/Stream.Client.Examples/Program.cs b/dotnet/examples/Stream.Client.Examples/Program.cs index 4ef1e5e..31474b6 100644 --- a/dotnet/examples/Stream.Client.Examples/Program.cs +++ b/dotnet/examples/Stream.Client.Examples/Program.cs @@ -20,8 +20,7 @@ Console.WriteLine("Listening for events. Press Ctrl+C to stop."); -// Pass a specific grid owner ID to only receive events for that grid owner. -await foreach (var envelope in streamClient.ReceiveAsync(gridOwnerId: null, quantity: Quantity.Current, infoLogger: Console.WriteLine, traceLogger: Console.WriteLine, token: cts.Token)) +await foreach (var envelope in streamClient.ReceiveAsync(quantity: Quantity.Current, infoLogger: Console.WriteLine, traceLogger: Console.WriteLine, token: cts.Token)) { if (envelope.HeimdallDlr is { } dlr) { diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/SseTestData.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/SseTestData.cs index ac9a3a1..1c63c2e 100644 --- a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/SseTestData.cs +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/SseTestData.cs @@ -1,6 +1,7 @@ using System.Net; using System.Text; using System.Text.Json; +using HeimdallPower.Api.Client.Stream.CapacityMonitoring.Lines; namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming.Fakes; @@ -21,12 +22,12 @@ public static string DlrEvent(Guid lineId, Guid spanId, DateTimeOffset timestamp data = new { at_line_id = lineId, at_span_id = spanId, timestamp, value, is_fallback = isFallback }, }; var json = JsonSerializer.Serialize(payload); - return $"event: Heimdall DLR\ndata: {json}\n\n"; + return $"event: {HeimdallDlrEvent.EventName}\ndata: {json}\n\n"; } public static string UnknownEvent => "event: something_else\ndata: {\"foo\":\"bar\"}\n\n"; - public static string MalformedDlrEvent => "event: Heimdall DLR\ndata: not-json\n\n"; + public static string MalformedDlrEvent => $"event: {HeimdallDlrEvent.EventName}\ndata: not-json\n\n"; public static HttpResponseMessage OkResponse(string sseBody) => new(HttpStatusCode.OK) { diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs index 6c849a5..e5933a0 100644 --- a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenAuthenticating.cs @@ -29,7 +29,7 @@ public async Task ShouldAttachAuthAndClientHeaders_BeforeFirstConnect() retryPolicy: ZeroDelay); using var cts = new CancellationTokenSource(); - await foreach (var _ in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) + await foreach (var _ in client.ReceiveAsync(infoLogger: _ => { }, token: cts.Token)) { break; } @@ -57,7 +57,7 @@ public async Task ShouldReuseToken_AcrossReconnects_WhileStillFresh() using var cts = new CancellationTokenSource(); var receivedCount = 0; - await foreach (var _ in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) + await foreach (var _ in client.ReceiveAsync(infoLogger: _ => { }, token: cts.Token)) { receivedCount++; if (receivedCount == 2) @@ -93,7 +93,7 @@ public async Task ShouldForceRefreshToken_OnUnauthorizedResponse() var logMessages = new List(); HeimdallEventEnvelope? received = null; - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, token: cts.Token)) + await foreach (var envelope in client.ReceiveAsync(infoLogger: logMessages.Add, token: cts.Token)) { received = envelope; break; diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenBuildingUrls.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenBuildingUrls.cs index 2bae137..420b74a 100644 --- a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenBuildingUrls.cs +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenBuildingUrls.cs @@ -10,27 +10,17 @@ namespace HeimdallPower.Api.Client.UnitTests.WhenStreaming; public class WhenBuildingUrls { [Fact] - public void ShouldBuildUrlWithoutQueryString_WhenGridOwnerIdIsNullAndQuantityIsDefault() + public void ShouldBuildUrlWithoutQueryString_WhenQuantityIsDefault() { - var url = StreamUrlBuilder.BuildStreamUrl(version: 1, gridOwnerId: null); + var url = StreamUrlBuilder.BuildStreamUrl(version: 1); Assert.Equal("/v1/stream", url); } - [Fact] - public void ShouldIncludeGridOwnerId_WhenProvided() - { - var gridOwnerId = Guid.NewGuid(); - - var url = StreamUrlBuilder.BuildStreamUrl(version: 1, gridOwnerId: gridOwnerId); - - Assert.Equal($"/v1/stream?gridownerid={gridOwnerId}", url); - } - [Fact] public void ShouldIncludeQuantity_WhenNotCurrent() { - var url = StreamUrlBuilder.BuildStreamUrl(version: 1, gridOwnerId: null, quantity: Quantity.ApparentPower); + var url = StreamUrlBuilder.BuildStreamUrl(version: 1, quantity: Quantity.ApparentPower); Assert.Equal("/v1/stream?quantity=apparent_power", url); } @@ -38,27 +28,15 @@ public void ShouldIncludeQuantity_WhenNotCurrent() [Fact] public void ShouldOmitQuantity_WhenCurrent() { - var gridOwnerId = Guid.NewGuid(); - - var url = StreamUrlBuilder.BuildStreamUrl(version: 1, gridOwnerId: gridOwnerId, quantity: Quantity.Current); - - Assert.Equal($"/v1/stream?gridownerid={gridOwnerId}", url); - } - - [Fact] - public void ShouldIncludeBothGridOwnerIdAndQuantity_WhenBothProvided() - { - var gridOwnerId = Guid.NewGuid(); - - var url = StreamUrlBuilder.BuildStreamUrl(version: 1, gridOwnerId: gridOwnerId, quantity: Quantity.ApparentPower); + var url = StreamUrlBuilder.BuildStreamUrl(version: 1, quantity: Quantity.Current); - Assert.Equal($"/v1/stream?gridownerid={gridOwnerId}&quantity=apparent_power", url); + Assert.Equal("/v1/stream", url); } [Fact] public void ShouldUseGivenVersion() { - var url = StreamUrlBuilder.BuildStreamUrl(version: 2, gridOwnerId: null); + var url = StreamUrlBuilder.BuildStreamUrl(version: 2); Assert.Equal("/v2/stream", url); } diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs index 39bd6eb..e7f7da7 100644 --- a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs @@ -29,7 +29,7 @@ public async Task ShouldYieldDeserializedDlrEvent_ForHeimdallDlrEventType() using var cts = new CancellationTokenSource(); HeimdallEventEnvelope? received = null; - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) + await foreach (var envelope in client.ReceiveAsync(infoLogger: _ => { }, token: cts.Token)) { received = envelope; break; @@ -59,7 +59,7 @@ public async Task ShouldNotYieldHeartbeats_ButShouldLogThem() var traceMessages = new List(); var events = new List(); - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, traceLogger: traceMessages.Add, token: cts.Token)) + await foreach (var envelope in client.ReceiveAsync(infoLogger: logMessages.Add, traceLogger: traceMessages.Add, token: cts.Token)) { events.Add(envelope); break; @@ -67,6 +67,7 @@ public async Task ShouldNotYieldHeartbeats_ButShouldLogThem() Assert.Single(events); Assert.Contains(traceMessages, m => m.Contains("Heartbeat", StringComparison.OrdinalIgnoreCase)); + Assert.Contains(logMessages, m => !m.Contains("Heartbeat", StringComparison.OrdinalIgnoreCase)); } [Fact] @@ -81,7 +82,7 @@ public async Task ShouldSkipUnknownEventTypes() using var cts = new CancellationTokenSource(); var events = new List(); - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) + await foreach (var envelope in client.ReceiveAsync(infoLogger: _ => { }, token: cts.Token)) { events.Add(envelope); break; @@ -111,7 +112,7 @@ public async Task ShouldRecoverAndReconnect_WhenDataIsMalformed() var logMessages = new List(); HeimdallEventEnvelope? received = null; - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, token: cts.Token)) + await foreach (var envelope in client.ReceiveAsync(infoLogger: logMessages.Add, token: cts.Token)) { received = envelope; break; diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs index e6f6f5b..baa6164 100644 --- a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs @@ -39,7 +39,7 @@ public async Task ShouldRecover_AfterTransientConnectionFailure() var logMessages = new List(); HeimdallEventEnvelope? received = null; - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: logMessages.Add, token: cts.Token)) + await foreach (var envelope in client.ReceiveAsync(infoLogger: logMessages.Add, token: cts.Token)) { received = envelope; break; @@ -68,7 +68,7 @@ public async Task ShouldContinueYieldingEvents_AcrossReconnectBoundary() using var cts = new CancellationTokenSource(); var values = new List(); - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) + await foreach (var envelope in client.ReceiveAsync(infoLogger: _ => { }, token: cts.Token)) { values.Add(envelope.HeimdallDlr!.Value); if (values.Count == 3) @@ -89,7 +89,7 @@ public async Task ShouldStopCleanly_WhenCancellationRequested() var enumerationTask = Task.Run(async () => { - await foreach (var envelope in client.ReceiveAsync(gridOwnerId: null, infoLogger: _ => { }, token: cts.Token)) + await foreach (var envelope in client.ReceiveAsync(infoLogger: _ => { }, token: cts.Token)) { events.Add(envelope); }