93 lines
3.7 KiB
C#
93 lines
3.7 KiB
C#
using System.Text.Json;
|
|
using FinlyticAssets.Services;
|
|
using FinlyticCore.Entities.Assets;
|
|
using FinlyticCore.Models.Assets;
|
|
using FinlyticCore.Util;
|
|
|
|
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 service, and publishes
|
|
/// the requested asset entities back to the corresponding response topic.
|
|
/// </summary>
|
|
public class AssetsMqttClient : ManagedMqttClient
|
|
{
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
private readonly ILogger<AssetsMqttClient> _logger;
|
|
|
|
/// <summary>
|
|
/// Initializes a new instance of the <see cref="AssetsMqttClient"/> class.
|
|
/// </summary>
|
|
/// <param name="logger">The logger used to record connection, error, and status messages.</param>
|
|
/// <param name="dbService">The database service used for querying and validating assets.</param>
|
|
public AssetsMqttClient(ILogger<AssetsMqttClient> logger, IServiceScopeFactory scopeFactory)
|
|
: base(logger)
|
|
{
|
|
_scopeFactory = scopeFactory;
|
|
_logger = logger;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Invoked automatically once the connection to the MQTT broker is successfully established or restored.
|
|
/// Registers the required wildcard subscriptions for incoming asset validation and search requests.
|
|
/// </summary>
|
|
/// <returns>A <see cref="Task"/> representing the asynchronous subscription operation.</returns>
|
|
protected override async Task OnConnectedAsync()
|
|
{
|
|
await SubscribeAsync("services/request/assets_Get/#");
|
|
await SubscribeAsync("services/request/assets_Search/#");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Processes incoming messages on the subscribed topics, executes the corresponding database query,
|
|
/// and publishes the result to the response topic while preserving the correlation ID.
|
|
/// </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)
|
|
{
|
|
var segments = topic.Split('/');
|
|
if (segments.Length < 4) return;
|
|
|
|
var channel = segments[2];
|
|
var correlationId = segments[3];
|
|
|
|
try
|
|
{
|
|
List<AssetEntity> responseData = [];
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<IAssetsDbService>();
|
|
|
|
switch (channel)
|
|
{
|
|
case "assets_Get":
|
|
var validReq = JsonSerializer.Deserialize<GetValidAssetRequest>(payload);
|
|
if (validReq != null)
|
|
{
|
|
responseData = await dbService.GetValidAssetsByIsinAsync(validReq.Isin);
|
|
}
|
|
break;
|
|
|
|
case "assets_Search":
|
|
var searchReq = JsonSerializer.Deserialize<SearchAssetsRequest>(payload);
|
|
if (searchReq != null)
|
|
{
|
|
responseData = await dbService.FindAffectedActiveAssetsAsync(searchReq.SearchQuery);
|
|
}
|
|
break;
|
|
}
|
|
|
|
{
|
|
string responseTopic = $"services/response/{channel}/{correlationId}";
|
|
await PublishAsync(responseTopic, responseData.ToDtoList());
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
OnError(ex);
|
|
}
|
|
}
|
|
} |