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

feat(core): update DTOs, Trade Republic client, Yahoo scrapers, and dynamic settings

Browse Source
This commit is contained in:
Kleidukos
2026-08-14 23:55:02 +02:00
parent 3d8af3940b
commit 4f733bf8c3
511 changed files with 1990 additions and 1080548 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FinlyticCore/Util/ManagedMqttClient.cs
+28 -23
View File
@@ -242,37 +242,42 @@ public abstract class ManagedMqttClient : IDisposable
}
}
private async Task HandleIncomingMessageAsync(MqttApplicationMessageReceivedEventArgs e)
private Task HandleIncomingMessageAsync(MqttApplicationMessageReceivedEventArgs e)
{
try
_ = Task.Run(async () =>
{
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);
// Intercept message if it belongs to the RPC response convention
if (topic.StartsWith("services/response/"))
try
{
var lastSlashIndex = topic.LastIndexOf('/');
if (lastSlashIndex != -1)
{
string correlationId = topic[(lastSlashIndex + 1)..];
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);
if (_pendingRequests.TryRemove(correlationId, out var tcs))
// Intercept message if it belongs to the RPC response convention
if (topic.StartsWith("services/response/"))
{
var lastSlashIndex = topic.LastIndexOf('/');
if (lastSlashIndex != -1)
{
tcs.SetResult(payload);
return; // Sinks the message, avoiding triggering OnMessageReceivedAsync for active RPC handles
string correlationId = topic[(lastSlashIndex + 1)..];
if (_pendingRequests.TryRemove(correlationId, out var tcs))
{
tcs.SetResult(payload ?? string.Empty);
return; // Sinks the message, avoiding triggering OnMessageReceivedAsync for active RPC handles
}
}
}
}
// Regular Pub/Sub message propagation
await OnMessageReceivedAsync(topic, payload);
}
catch (Exception ex)
{
OnError(ex);
}
// Regular Pub/Sub message propagation
await OnMessageReceivedAsync(topic, payload ?? string.Empty);
}
catch (Exception ex)
{
OnError(ex);
}
});
return Task.CompletedTask;
}
private async Task HandleDisconnectAsync(MqttClientDisconnectedEventArgs e)