92 lines
3.2 KiB
C#
92 lines
3.2 KiB
C#
using FinlyticBackend.Database;
|
|
using FinlyticBackend.Hubs;
|
|
using FinlyticBackend.Util;
|
|
using FinlyticCore.Dtos;
|
|
using FinlyticCore.Dtos.TechnicalAnalysis;
|
|
using Microsoft.AspNetCore.SignalR;
|
|
using Microsoft.EntityFrameworkCore;
|
|
|
|
namespace FinlyticBackend.Services;
|
|
|
|
public class FavoritesPriceBackgroundService(
|
|
IHubContext<FavoritesPriceHub> hubContext,
|
|
WebMqttClient mqttClient,
|
|
IServiceScopeFactory scopeFactory,
|
|
ILogger<FavoritesPriceBackgroundService> logger) : BackgroundService
|
|
{
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
await Task.Delay(4000, stoppingToken);
|
|
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
var activeUserIds = FavoritesPriceHub.GetActiveUserIds();
|
|
if (activeUserIds.Length > 0)
|
|
{
|
|
await ProcessActiveUserPricesAsync(activeUserIds, stoppingToken);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
logger.LogWarning(ex, "[FavoritesPriceBackgroundService] Error broadcasting price updates.");
|
|
}
|
|
|
|
await Task.Delay(10000, stoppingToken);
|
|
}
|
|
}
|
|
|
|
private async Task ProcessActiveUserPricesAsync(string[] activeUserIds, CancellationToken cancellationToken)
|
|
{
|
|
using var scope = scopeFactory.CreateScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<BackendDbContext>();
|
|
|
|
// 1. Direkt per EF nach aktiven Usern filtern & laden
|
|
var userFavorites = await db.UserFavoriteAssets
|
|
.AsNoTracking()
|
|
.Where(f => activeUserIds.Contains(f.UserId.ToString()))
|
|
.ToListAsync(cancellationToken);
|
|
|
|
if (userFavorites.Count == 0) return;
|
|
|
|
// 2. Pro User einfach die Kurse abfragen und senden
|
|
foreach (var userGroup in userFavorites.GroupBy(f => f.UserId.ToString()))
|
|
{
|
|
var userId = userGroup.Key;
|
|
var priceUpdates = new Dictionary<string, object>();
|
|
|
|
foreach (var fav in userGroup)
|
|
{
|
|
var cleanIsin = fav.Isin.Trim().ToUpperInvariant();
|
|
|
|
if (!mqttClient.IsConnected) continue;
|
|
|
|
try
|
|
{
|
|
var livePrice = await mqttClient.SendRpcRequestAsync<LivePriceDto, IsinRequest>(
|
|
"tr_GetLivePrice",
|
|
new IsinRequest(cleanIsin),
|
|
TimeSpan.FromSeconds(2)
|
|
);
|
|
|
|
if (livePrice != null)
|
|
{
|
|
priceUpdates[cleanIsin] = new
|
|
{
|
|
currentPrice = (double)livePrice.CurrentPrice,
|
|
dailyChangePercent = (double)livePrice.DailyChangePercent
|
|
};
|
|
}
|
|
}
|
|
catch { /* Ignorieren bei Einzel-Timeout */ }
|
|
}
|
|
|
|
if (priceUpdates.Count > 0)
|
|
{
|
|
await hubContext.Clients.Group(FavoritesPriceHub.GetGroupName(userId))
|
|
.SendAsync("ReceiveFavoritePrices", priceUpdates, cancellationToken);
|
|
}
|
|
}
|
|
}
|
|
} |