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

feat(assets): dynamic settings, IFinlyticLogger, live log streaming, and EF migration

Browse Source
This commit is contained in:
Kleidukos
2026-08-15 21:30:46 +02:00
parent 57554a9582
commit 1f9d66405a
11 changed files with 855 additions and 384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FinlyticAssets/Util/AssetsMqttClient.cs
+130 -31
View File
@@ -1,75 +1,94 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using FinlyticAssets.Entities;
using FinlyticAssets.Services;
using FinlyticCore.Dtos.Settings;
using FinlyticCore.Models;
using FinlyticCore.Services;
using FinlyticCore.Util;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
namespace FinlyticAssets.Util;
/// <summary>
/// Represents a managed MQTT client acting as a server-side RPC provider within the asset microservice.
/// It subscribes to request topics, processes incoming JSON payloads via the database and index service,
/// and publishes the requested asset entities or logo files back to the response topic.
/// Also implements <see cref="IHostedService"/> to manage its own lifecycle connections.
/// </summary>
public class AssetsMqttClient(
ILogger<AssetsMqttClient> logger,
IServiceScopeFactory scopeFactory,
IConfiguration configuration) : ManagedMqttClient(logger), IHostedService
public class AssetsMqttClient : ManagedMqttClient, IHostedService
{
private readonly ILogger<AssetsMqttClient> _logger;
private readonly IServiceScopeFactory _scopeFactory;
private readonly IConfiguration _configuration;
public AssetsMqttClient(
ILogger<AssetsMqttClient> logger,
IServiceScopeFactory scopeFactory,
IConfiguration configuration) : base(logger)
{
_logger = logger;
_scopeFactory = scopeFactory;
_configuration = configuration;
}
/// <summary>
/// Starts the MQTT client and connects to the configured broker.
/// </summary>
/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
/// <returns>A task representing the asynchronous start operation.</returns>
public async Task StartAsync(CancellationToken cancellationToken)
{
var config = new MqttConfiguration()
{
Host = configuration["MQTT:Host"] ?? configuration["MQTT__Host"]!,
Port = Convert.ToInt32(configuration["MQTT:Port"] ?? configuration["MQTT__Port"]!),
ClientId = $"{(configuration["MQTT:ClientId"] ?? configuration["MQTT__ClientId"]!)}_{Guid.NewGuid()}"
Host = _configuration["MQTT:Host"] ?? _configuration["MQTT__Host"] ?? "localhost",
Port = Convert.ToInt32(_configuration["MQTT:Port"] ?? _configuration["MQTT__Port"] ?? "1883"),
ClientId = $"{(_configuration["MQTT:ClientId"] ?? _configuration["MQTT__ClientId"] ?? "FinlyticAssets")}_{Guid.NewGuid()}"
};
_logger.LogInformation("Starting Assets MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
await ConnectAsync(config);
}
/// <summary>
/// Gracefully stops and disconnects the MQTT client.
/// </summary>
/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
/// <returns>A task representing the asynchronous stop operation.</returns>
public async Task StopAsync(CancellationToken cancellationToken)
{
_logger.LogInformation("Stopping Assets MQTT client.");
await DisconnectAsync();
}
/// <summary>
/// Invoked automatically once the connection to the MQTT broker is successfully established or restored.
/// Registers subscriptions for asset validation, search, and logo download requests.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous subscription operation.</returns>
protected override async Task OnConnectedAsync()
{
_logger.LogInformation("Assets MQTT Client connected. Subscribing to topics...");
await SubscribeAsync("services/request/assets_Get/#");
await SubscribeAsync("services/request/assets_Search/#");
await SubscribeAsync("services/request/assets_GetDiscovery/#");
await SubscribeAsync("services/request/assets_GetDerivatives/#");
await SubscribeAsync("services/request/assets_FetchLogo/#");
await SubscribeAsync("services/request/assets_settings_GetAll/#");
await SubscribeAsync("services/request/assets_settings_Update/#");
await SubscribeAsync("services/request/health_Ping/#");
await SubscribeAsync("services/config/updated/#");
FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
{
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticAssets", StringComparison.OrdinalIgnoreCase))
{
await PublishAsync("finlytic/logs/FinlyticAssets", logDto);
}
};
}
/// <summary>
/// Processes incoming messages on the subscribed topics, executes the corresponding service methods,
/// and publishes the result to the response topic while preserving the correlation ID.
/// Processes incoming messages on the subscribed topics.
/// </summary>
/// <param name="topic">The MQTT topic on which the message was received.</param>
/// <param name="payload">The incoming message as a UTF-8 encoded JSON string.</param>
/// <returns>A <see cref="Task"/> representing the asynchronous message processing operation.</returns>
protected override async Task OnMessageReceivedAsync(string topic, string payload)
{
if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase))
@@ -90,9 +109,21 @@ public class AssetsMqttClient(
return;
}
if (topic.StartsWith("services/request/assets_settings_GetAll", StringComparison.OrdinalIgnoreCase))
{
await HandleSettingsGetAllAsync(correlationId);
return;
}
if (topic.StartsWith("services/request/assets_settings_Update", StringComparison.OrdinalIgnoreCase))
{
await HandleSettingsUpdateAsync(payload, correlationId);
return;
}
try
{
using var scope = scopeFactory.CreateScope();
using var scope = _scopeFactory.CreateScope();
var dbService = scope.ServiceProvider.GetRequiredService<IAssetsDbService>();
var indexService = scope.ServiceProvider.GetRequiredService<IAssetsIndexService>();
@@ -129,26 +160,90 @@ public class AssetsMqttClient(
}
}
private async Task HandleSettingsGetAllAsync(string correlationId)
{
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_GetAll] Retrieving all dynamic settings via reflection [CorrelationId: {CorrelationId}]", correlationId);
try
{
var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
var responseTopic = $"services/response/assets_settings_GetAll/{correlationId}";
await PublishAsync(responseTopic, settings);
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic);
}
catch (Exception ex)
{
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticAssets] [Settings_GetAll] Failed to retrieve settings.");
}
}
private async Task HandleSettingsUpdateAsync(string payload, string correlationId)
{
if (string.IsNullOrWhiteSpace(payload)) return;
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId);
try
{
Dictionary<string, object?>? updates = null;
try
{
updates = JsonSerializer.Deserialize<Dictionary<string, object?>>(payload);
}
catch
{
var list = JsonSerializer.Deserialize<List<DynamicSettingDto>>(payload);
if (list != null)
{
updates = new Dictionary<string, object?>();
foreach (var item in list) updates[item.Key] = item.Value;
}
}
if (updates != null && updates.Count > 0)
{
await settingsService.UpdateSettingsAsync(updates);
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
}
var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
var responseTopic = $"services/response/assets_settings_Update/{correlationId}";
await PublishAsync(responseTopic, currentSettings);
}
catch (Exception ex)
{
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticAssets] [Settings_Update] Failed to update settings.");
}
}
private async Task HandleConfigUpdatedAsync(string topic, string payload)
{
if (!topic.EndsWith("FinlyticAssets", StringComparison.OrdinalIgnoreCase))
return;
logger.LogInformation("[{Channel}] [AssetsMqttClient] Received config update event for FinlyticAssets.", "AssetsChannel");
try
{
var updatePayload = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.ServiceConfigUpdatePayload);
if (updatePayload?.Settings != null && updatePayload.Settings.Count > 0)
{
using var scope = scopeFactory.CreateScope();
var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
await settingsDb.UpdateSettingsFromDictionary(updatePayload.Settings);
logger.LogInformation("[{Channel}] [AssetsMqttClient] Persisted {Count} updated settings to FinlyticAssets database.", "AssetsChannel", updatePayload.Settings.Count);
using var scope = _scopeFactory.CreateScope();
var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
var dict = updatePayload.Settings.ToDictionary(k => k.Key, v => (object?)v.Value);
await settings.UpdateSettingsAsync(dict);
}
}
catch (Exception ex)
{
logger.LogError(ex, "[{Channel}] [AssetsMqttClient] Error processing MQTT config update event.", "AssetsChannel");
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
await finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsMqttClient] Error processing MQTT config update event.");
}
}
@@ -162,7 +257,9 @@ public class AssetsMqttClient(
{
string respTopic = $"services/response/health_Ping/{correlationId}";
await PublishAsync(respTopic, new FinlyticCore.Dtos.ServiceHealthResponse("FinlyticAssets", "Online", DateTime.UtcNow, "Connected"));
logger.LogInformation("[{Channel}] [AssetsMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", "AssetsChannel", correlationId);
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[AssetsMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
}
}
@@ -236,7 +333,9 @@ public class AssetsMqttClient(
}
catch (Exception ex)
{
logger.LogError(ex, "[{Channel}] Error parsing GetDerivativesRequest payload.", "AssetsChannel");
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
await finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsMqttClient] Error parsing GetDerivativesRequest payload.");
}
return [];