diff --git a/dotnet/HeimdallPower.Api.Client.slnx b/dotnet/HeimdallPower.Api.Client.slnx index f321152..665202a 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..013db1c --- /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 +{ + /// + /// Registers as a singleton, configured via . + /// + /// + /// 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"; + + services.Configure(configureOptions); + + services.AddHttpClient(clientName) + .ConfigureHttpClient((_, client) => + { + client.BaseAddress = new Uri("https://stream-api.heimdallcloud.com"); + }) + .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, options.RetryPolicy); + }); + + 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..7dc4970 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client.Extensions/HeimdallStreamClientOptions.cs @@ -0,0 +1,35 @@ +using HeimdallPower.Api.Client.Stream; + +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; } + + /// + /// 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/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..58226eb --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/HeimdallApiEndpoints.cs @@ -0,0 +1,15 @@ +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"; + 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"; + 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/CapacityMonitoring/Lines/HeimdallDlrEvent.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/CapacityMonitoring/Lines/HeimdallDlrEvent.cs new file mode 100644 index 0000000..0800700 --- /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 SSE 'event:' field for Heimdall DLR events. + /// + public const string EventName = "heimdall_dlr"; +} 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/HeimdallEventEnvelope.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs new file mode 100644 index 0000000..774db70 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallEventEnvelope.cs @@ -0,0 +1,32 @@ +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, + string Unit, + JsonElement Data) +{ + private HeimdallDlrEvent? _heimdallDlr; + + /// + /// deserialized as a , or if it cannot be deserialized as one. + /// Only meaningful when equals "Heimdall DLR" />. + /// + public HeimdallDlrEvent? HeimdallDlr + { + get + { + return _heimdallDlr ??= Data.Deserialize(HeimdallStreamJsonSerializerOptions.Default); + } + } +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs new file mode 100644 index 0000000..356d73c --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamClient.cs @@ -0,0 +1,194 @@ +using HeimdallPower.Api.Client; +using HeimdallPower.Api.Client.Stream.CapacityMonitoring; +using HeimdallPower.Api.Client.Stream.CapacityMonitoring.Lines; +using System.Net; +using System.Net.ServerSentEvents; +using System.Runtime.CompilerServices; +using System.Text.Json; + +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; + private readonly AccessTokenHeaderRefresher _tokenRefresher; + private readonly StreamConnectionRetryPolicy _retryPolicy; + + /// + /// A client that lets you consume the Heimdall Stream API. + /// + /// + /// 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. + /// Optional additional metadata to include in request headers. + /// An optional message handler used to route token acquisition through a proxy. + /// 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, + new StreamConnectionRetryPolicy(retryPolicyOptions)) + { + } + + // 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 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 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( + 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(quantity, traceLogger, 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 (UnauthorizedAccessException) + { + failedAttempts++; + infoLogger?.Invoke($"Stream error: unauthorized. Refreshing token and reconnecting... (attempt #{failedAttempts})"); + await _tokenRefresher.ForceRefreshAsync(token); + break; + } + catch (Exception ex) + { + failedAttempts++; + + if (ex is HeimdallApiException + && !_retryPolicy.ShouldRetry(failedAttempts)) + throw; + + infoLogger?.Invoke($"Stream error: {ex.Message}. Reconnecting... (attempt #{failedAttempts})"); + break; + } + + yield return envelope; + } + + if (token.IsCancellationRequested) + yield break; + + try + { + 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) + { + yield break; + } + } + } + + /// + /// 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( + Quantity quantity, + Action? traceLogger, + [EnumeratorCancellation] CancellationToken token) + { + await _tokenRefresher.EnsureFreshTokenAsync(token); + + string url = UrlBuilder.BuildStreamUrl(version: 1, quantity); + + using var request = new HttpRequestMessage( + HttpMethod.Get, + 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."); + + 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); + + await foreach (SseItem item in SseParser.Create(stream).EnumerateAsync(token)) + { + if (item.EventType == "heartbeat") + { + traceLogger?.Invoke("Heartbeat received."); + continue; + } + + if (string.IsNullOrWhiteSpace(item.Data)) + continue; + + if (item.EventType == HeimdallDlrEvent.EventName) + { + HeimdallEventEnvelope? envelope = JsonSerializer.Deserialize(item.Data, HeimdallStreamJsonSerializerOptions.Default); + + 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/HeimdallStreamJsonSerializerOptions.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamJsonSerializerOptions.cs new file mode 100644 index 0000000..d60e98e --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/HeimdallStreamJsonSerializerOptions.cs @@ -0,0 +1,35 @@ +using System.Text.Json; +using System.Text.Json.Serialization; + +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; + + private static JsonSerializerOptions CreateJsonOptions() + { + var options = new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true, + PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower + }; + options.Converters.Add(new JsonStringEnumConverter()); + return options; + } + + /// + /// The shared instance, created on first access. + /// + public static JsonSerializerOptions Default + { + get + { + return _jsonOptions ??= CreateJsonOptions(); + } + } +} diff --git a/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs new file mode 100644 index 0000000..d7882f7 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/IHeimdallStreamClient.cs @@ -0,0 +1,33 @@ +using HeimdallPower.Api.Client.Stream.CapacityMonitoring; + +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 +{ + /// + /// Receives Heimdall events from the Heimdall Stream API as a continuous asynchronous stream. + /// + /// + /// 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 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(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 new file mode 100644 index 0000000..14b2bba --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/StreamConnectionRetryPolicy.cs @@ -0,0 +1,60 @@ +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(StreamConnectionRetryPolicyOptions? options = null) +{ + 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. + /// + /// 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) + 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)); + } + + /// + /// 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..c926342 --- /dev/null +++ b/dotnet/HeimdallPower.Api.Client/HeimdallPower.Api.Client/Stream/UrlBuilder.cs @@ -0,0 +1,21 @@ +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, Quantity quantity = Quantity.Current) + { + var queryParams = new NameValueCollection(); + + 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/README.md b/dotnet/README.md index fee273e..53dd422 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()) +{ + 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/Stream.Client.Examples`](examples/Stream.Client.Examples). + ## Error Handling ### Resilience and retry diff --git a/dotnet/examples/Stream.Client.Examples/Program.cs b/dotnet/examples/Stream.Client.Examples/Program.cs new file mode 100644 index 0000000..31474b6 --- /dev/null +++ b/dotnet/examples/Stream.Client.Examples/Program.cs @@ -0,0 +1,35 @@ +using HeimdallPower.Api.Client.Stream; +using HeimdallPower.Api.Client.Stream.CapacityMonitoring; + +// 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."); + +await foreach (var envelope in streamClient.ReceiveAsync(quantity: Quantity.Current, infoLogger: Console.WriteLine, traceLogger: Console.WriteLine, token: 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}"); + } +} + +Console.WriteLine("Stream stopped."); 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 + + + + + + + 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..1c63c2e --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/Fakes/SseTestData.cs @@ -0,0 +1,44 @@ +using System.Net; +using System.Text; +using System.Text.Json; +using HeimdallPower.Api.Client.Stream.CapacityMonitoring.Lines; + +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: {HeimdallDlrEvent.EventName}\ndata: {json}\n\n"; + } + + public static string UnknownEvent => "event: something_else\ndata: {\"foo\":\"bar\"}\n\n"; + + public static string MalformedDlrEvent => $"event: {HeimdallDlrEvent.EventName}\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..e5933a0 --- /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(new StreamConnectionRetryPolicyOptions { InitialDelay = TimeSpan.Zero, MaxDelay = 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(infoLogger: _ => { }, token: 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(infoLogger: _ => { }, token: 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(infoLogger: logMessages.Add, token: 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/WhenBuildingUrls.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenBuildingUrls.cs new file mode 100644 index 0000000..420b74a --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenBuildingUrls.cs @@ -0,0 +1,43 @@ +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_WhenQuantityIsDefault() + { + var url = StreamUrlBuilder.BuildStreamUrl(version: 1); + + Assert.Equal("/v1/stream", url); + } + + [Fact] + public void ShouldIncludeQuantity_WhenNotCurrent() + { + var url = StreamUrlBuilder.BuildStreamUrl(version: 1, quantity: Quantity.ApparentPower); + + Assert.Equal("/v1/stream?quantity=apparent_power", url); + } + + [Fact] + public void ShouldOmitQuantity_WhenCurrent() + { + var url = StreamUrlBuilder.BuildStreamUrl(version: 1, quantity: Quantity.Current); + + Assert.Equal("/v1/stream", url); + } + + [Fact] + public void ShouldUseGivenVersion() + { + var url = StreamUrlBuilder.BuildStreamUrl(version: 2); + + Assert.Equal("/v2/stream", url); + } +} diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenComputingRetryDelay.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenComputingRetryDelay.cs new file mode 100644 index 0000000..d8fb5a2 --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenComputingRetryDelay.cs @@ -0,0 +1,61 @@ +using HeimdallPower.Api.Client.Stream; + +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 +{ + private static readonly TimeSpan InitialDelay = TimeSpan.FromSeconds(1); + private static readonly TimeSpan MaxDelay = TimeSpan.FromSeconds(30); + + private readonly StreamConnectionRetryPolicy _policy = new(new StreamConnectionRetryPolicyOptions { InitialDelay = InitialDelay, MaxDelay = 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/WhenParsingEvents.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs new file mode 100644 index 0000000..e7f7da7 --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenParsingEvents.cs @@ -0,0 +1,126 @@ +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(new StreamConnectionRetryPolicyOptions { InitialDelay = TimeSpan.Zero, MaxDelay = 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(infoLogger: _ => { }, token: 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 traceMessages = new List(); + + var events = new List(); + await foreach (var envelope in client.ReceiveAsync(infoLogger: logMessages.Add, traceLogger: traceMessages.Add, token: cts.Token)) + { + events.Add(envelope); + break; + } + + Assert.Single(events); + Assert.Contains(traceMessages, m => m.Contains("Heartbeat", StringComparison.OrdinalIgnoreCase)); + 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(infoLogger: _ => { }, token: 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(infoLogger: logMessages.Add, token: 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..baa6164 --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenStreaming/WhenReconnecting.cs @@ -0,0 +1,120 @@ +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 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(), + 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(infoLogger: logMessages.Add, token: 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(infoLogger: _ => { }, token: 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(infoLogger: _ => { }, token: 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); + } + + + [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}"); + } +}