Kleidukos/Finlytic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core): dynamic settings service, IFinlyticLogger, log broadcaster, and persistent Yahoo auth

Browse Source
This commit is contained in:
Kleidukos
2026-08-15 21:29:38 +02:00
parent 34fa774cbf
commit 3dbee36ca0
15 changed files with 920 additions and 392 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FinlyticCore/Util/ManagedMqttClient.cs
+69 -19
View File
@@ -6,6 +6,8 @@ using System.Text.Json.Serialization;
using System.Threading;
using System.Threading.Tasks;
using FinlyticCore.Models;
using FinlyticCore.Models.Settings;
using FinlyticCore.Services;
using Microsoft.Extensions.Logging;
using MQTTnet;
@@ -14,10 +16,13 @@ namespace FinlyticCore.Util;
/// <summary>
/// An abstract, resilient MQTT client wrapper designed for microservice architectures.
/// Handles automatic reconnection, structured JSON publishing, thread-safe subscription management, and synchronous Request-Reply (RPC).
/// Supports channel-controlled logging via <see cref="CoreSettingKeys.MqttChannel"/>.
/// </summary>
public abstract class ManagedMqttClient : IDisposable
{
private readonly ILogger<ManagedMqttClient> _logger;
private readonly ISettingsService? _settingsService;
private readonly IFinlyticLogger<ManagedMqttClient>? _finlyticLogger;
private readonly IMqttClient _mqttClient;
private CancellationTokenSource? _cts;
@@ -29,15 +34,58 @@ public abstract class ManagedMqttClient : IDisposable
/// </summary>
public bool IsConnected => _mqttClient.IsConnected;
protected ManagedMqttClient(ILogger<ManagedMqttClient> logger)
protected ManagedMqttClient(
ILogger<ManagedMqttClient> logger,
ISettingsService? settingsService = null,
IFinlyticLogger<ManagedMqttClient>? finlyticLogger = null)
{
_logger = logger;
_settingsService = settingsService;
_finlyticLogger = finlyticLogger;
_mqttClient = new MqttClientFactory().CreateMqttClient();
_mqttClient.ApplicationMessageReceivedAsync += HandleIncomingMessageAsync;
_mqttClient.DisconnectedAsync += HandleDisconnectAsync;
}
private async Task LogMqttInfoAsync(string message, params object[] args)
{
if (_finlyticLogger != null)
{
await _finlyticLogger.LogInfoAsync(CoreSettingKeys.MqttChannel, message, args);
}
else if (_settingsService != null)
{
if (await _settingsService.GetSettingAsync(CoreSettingKeys.MqttChannel))
{
_logger.LogInformation(message, args);
}
}
else
{
_logger.LogInformation(message, args);
}
}
private async Task LogMqttDebugAsync(string message, params object[] args)
{
if (_finlyticLogger != null)
{
await _finlyticLogger.LogDebugAsync(CoreSettingKeys.MqttChannel, message, args);
}
else if (_settingsService != null)
{
if (await _settingsService.GetSettingAsync(CoreSettingKeys.MqttChannel))
{
_logger.LogDebug(message, args);
}
}
else
{
_logger.LogDebug(message, args);
}
}
/// <summary>
/// Establishes a connection to the MQTT broker and initializes the background auto-reconnection loop.
/// </summary>
@@ -61,12 +109,12 @@ public abstract class ManagedMqttClient : IDisposable
var options = optionsBuilder.Build();
_logger.LogInformation("Connecting to MQTT broker at {Host}:{Port}...", config.Host, config.Port);
await LogMqttInfoAsync("Connecting to MQTT broker at {Host}:{Port}...", config.Host, config.Port);
try
{
await _mqttClient.ConnectAsync(options, _cts.Token);
_logger.LogInformation("Successfully connected to MQTT broker.");
await LogMqttInfoAsync("Successfully connected to MQTT broker.");
await OnConnectedAsync();
}
@@ -94,7 +142,7 @@ public abstract class ManagedMqttClient : IDisposable
{
Reason = MqttClientDisconnectOptionsReason.NormalDisconnection
});
_logger.LogInformation("MQTT connection gracefully closed.");
await LogMqttInfoAsync("MQTT connection gracefully closed.");
}
catch (Exception ex)
{
@@ -127,7 +175,7 @@ public abstract class ManagedMqttClient : IDisposable
.Build();
await _mqttClient.SubscribeAsync(subscribeOptions, CancellationToken.None);
_logger.LogDebug("Successfully subscribed to topic: {Topic} (NoLocal: {NoLocal})", topic, noLocal);
await LogMqttDebugAsync("Successfully subscribed to topic: {Topic} (NoLocal: {NoLocal})", topic, noLocal);
}
/// <summary>
@@ -180,15 +228,21 @@ public abstract class ManagedMqttClient : IDisposable
return _mqttClient.PublishAsync(message, CancellationToken.None);
}
/// <summary>
/// Sends a parameterless request to an RPC channel and asynchronously blocks until a matching response arrives.
/// </summary>
public Task<TResponse?> SendRpcRequestAsync<TResponse>(
string channel,
TimeSpan? timeout = null)
where TResponse : class
{
return SendRpcRequestAsync<TResponse, string>(channel, string.Empty, timeout);
}
/// <summary>
/// Sends a generic request payload to an RPC channel and asynchronously blocks until a matching response arrives.
/// Uses the topic conventions: <c>services/request/{channel}/{correlationId}</c> and <c>services/response/{channel}/{correlationId}</c>.
/// </summary>
/// <typeparam name="TResponse">The expected strongly-typed object type of the reply.</typeparam>
/// <typeparam name="TRequest">The type of the payload being transmitted.</typeparam>
/// <param name="channel">The target sub-channel or service name (e.g., "sentix", "assets").</param>
/// <param name="requestData">The object that will be serialized to JSON and sent.</param>
/// <param name="timeout">Optional. Maximum time to wait before returning null. Defaults to 10 seconds.</param>
public async Task<TResponse?> SendRpcRequestAsync<TResponse, TRequest>(
string channel,
TRequest requestData,
@@ -209,7 +263,7 @@ public abstract class ManagedMqttClient : IDisposable
// 2. Serialize and dispatch via the existing JSON helper
await PublishAsync(requestTopic, requestData);
_logger.LogInformation("RPC request published to '{Topic}' [CorrelationId: {Id}]", requestTopic, correlationId);
await LogMqttInfoAsync("RPC request published to '{Topic}' [CorrelationId: {Id}]", requestTopic, correlationId);
try
{
@@ -250,7 +304,7 @@ public abstract class ManagedMqttClient : IDisposable
{
var topic = e.ApplicationMessage.Topic;
var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
_logger.LogInformation("MQTT message received on topic '{Topic}', length={Length}", topic, payload?.Length ?? 0);
await LogMqttDebugAsync("MQTT message received on topic '{Topic}', length={Length}", topic, payload?.Length ?? 0);
// Intercept message if it belongs to the RPC response convention
if (topic.StartsWith("services/response/"))
@@ -282,7 +336,6 @@ public abstract class ManagedMqttClient : IDisposable
private async Task HandleDisconnectAsync(MqttClientDisconnectedEventArgs e)
{
// Prevent trigger during deliberate connection shutdowns
if (_cts == null || _cts.IsCancellationRequested)
return;
@@ -296,19 +349,19 @@ public abstract class ManagedMqttClient : IDisposable
try
{
_logger.LogInformation("Reconnect attempt {Attempt} in {Delay}s...", attempt, delaySeconds);
await LogMqttInfoAsync("Reconnect attempt {Attempt} in {Delay}s...", attempt, delaySeconds);
await Task.Delay(TimeSpan.FromSeconds(delaySeconds), _cts.Token);
await _mqttClient.ReconnectAsync(_cts.Token);
if (_mqttClient.IsConnected)
{
_logger.LogInformation("MQTT client reconnected successfully after {Attempt} attempt(s).", attempt);
await LogMqttInfoAsync("MQTT client reconnected successfully after {Attempt} attempt(s).", attempt);
await OnConnectedAsync();
return;
}
}
catch (OperationCanceledException) { return; /* Expected on application shutdown */ }
catch (OperationCanceledException) { return; }
catch (Exception ex)
{
_logger.LogWarning(ex, "Reconnect attempt {Attempt} to the MQTT broker failed.", attempt);
@@ -318,15 +371,12 @@ public abstract class ManagedMqttClient : IDisposable
/// <summary>
/// Fired automatically whenever a connection or reconnection is successfully established.
/// Ideal place to trigger <see cref="SubscribeAsync"/> operations.
/// </summary>
protected abstract Task OnConnectedAsync();
/// <summary>
/// Fired whenever a new message lands on a registered subscription channel.
/// </summary>
/// <param name="topic">The specific topic where the message was broadcasted.</param>
/// <param name="payload">The deserialized UTF-8 payload string.</param>
protected abstract Task OnMessageReceivedAsync(string topic, string payload);
/// <summary>