Init

FinlyticAssets/Util/AssetsMqttClient.cs

@@ -0,0 +1,93 @@
+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);
+        }
+    }
+}