using Microsoft.AspNetCore.Identity.UI.Services; using Microsoft.Extensions.Configuration; using Newtonsoft.Json; using NLog; using StackExchange.Redis; using System.Diagnostics; using WebDoorCreator.Core; using WebDoorCreator.Data; namespace WebDoorCreator.Data.Services { public class QueueDataService : IDisposable { #region Public Constructors /// /// Init classe /// /// /// /// /// public QueueDataService(IConfiguration configuration, IEmailSender emailSender, IConnectionMultiplexer redisConn) { _configuration = configuration; _emailSender = emailSender; // setup componenti REDIS var redConnString = _configuration.GetConnectionString("Redis"); if (redConnString == null) { Log.Error("REDIS ConnString empty!"); } else { this.redisConn = ConnectionMultiplexer.Connect(redConnString); redisDb = this.redisConn.GetDatabase(); } // setup canali pub/sub calcReqPipe = new MessagePipe(redisConn, Constants.CALC_REQ_QUEUE); calcDonePipe = new MessagePipe(redisConn, Constants.CALC_DONE_QUEUE); } #endregion Public Constructors #region Public Properties /// /// Message pipe esecuzione elaborazione CAM --> UI /// public MessagePipe calcDonePipe { get; set; } = null!; /// /// Message pipe richieste elaborazione UI --> CAM /// public MessagePipe calcReqPipe { get; set; } = null!; #endregion Public Properties #region Public Methods public void Dispose() { redisConn.Dispose(); } /// /// /// /// /// Contenuto completo del DDF /// indice/versione del calcolo DDF public async Task sendCalcReq(int DoorId, string FullDDF) { int currVers = 1; Stopwatch stopWatch = new Stopwatch(); stopWatch.Start(); // accumulo richieste di calcolo in HashTable redis // per prima cosa controllo se ho GIA' in coda qualcosa come richeiste var numPending = await NumRequestPending(); // se coda vuota --> aggiungo e stop if (numPending == 0) { } // se coda non vuota: cerco la DoorId corrente, se c'è sovrascrivo versione altrimenti // inserisco in coda else { } // salvo nell'archivio REDIS delle porte il DDF corrente (door+ vers numb), potrebbe // venire buono anche x eventuale UNDO... // invio sul canale dei messaggi il numero di items in coda attuali x chiedere esecuzione... #if false // cerco da cache RedisKey currKey = new RedisKey(Constants.CALC_REQ_PEND); currVers = redisDb.HashLength(currKey); if (currVers > 0) { var rawData = await redisDb.HashGetAllAsync(currKey); foreach (var item in rawData) { dictResult.Add($"{item.Name}", $"{item.Value}"); } } #endif stopWatch.Stop(); TimeSpan ts = stopWatch.Elapsed; Log.Debug($"EnqueueDoorDDF | DoorId: {DoorId} enqueued in: {ts.TotalMilliseconds} ms"); // to do // calcolo quanti emssaggi ho in coda string message = "999"; // invio sul channel redis la richiesta, salvando il valore anche in cache bool answ = calcReqPipe.saveAndSendMessage(Constants.LAST_CALC_REQ_KEY, message); // simulazione ritorno dati... await Task.Delay(200); // invio il FINTO messaggio di ritorno... string retMess = ""; string fileName = Path.Combine("temp", $"Logo{idxSim:00}.svg"); if (File.Exists(fileName)) { retMess = File.ReadAllText(fileName); idxSim++; idxSim = idxSim % 4; } calcDonePipe.saveAndSendMessage(Constants.LAST_CALC_DONE_KEY, retMess); return currVers; } public async Task FlushRedisCache() { await Task.Delay(1); RedisValue pattern = new RedisValue($"{Constants.BASE_HASH}:Cache*"); bool answ = await ExecFlushRedisPattern(pattern); return answ; } /// /// Get # of calculation request pending /// public async Task NumRequestPending() { long numReq = 0; string source = "REDIS"; Dictionary dictResult = new Dictionary(); // cerco da cache RedisKey currKey = new RedisKey(Constants.CALC_REQ_PEND); Stopwatch stopWatch = new Stopwatch(); stopWatch.Start(); numReq = await redisDb.HashLengthAsync(currKey); stopWatch.Stop(); TimeSpan ts = stopWatch.Elapsed; Log.Debug($"NumRequestPending | {source} in: {ts.TotalMilliseconds} ms"); return numReq; } /// /// Get # of calculation request processing /// public async Task NumRequestProcessing() { long numReq = 0; string source = "REDIS"; Dictionary dictResult = new Dictionary(); // cerco da cache RedisKey currKey = new RedisKey(Constants.CALC_REQ_PROC); Stopwatch stopWatch = new Stopwatch(); stopWatch.Start(); numReq = await redisDb.HashLengthAsync(currKey); stopWatch.Stop(); TimeSpan ts = stopWatch.Elapsed; Log.Debug($"NumRequestProcessing | {source} in: {ts.TotalMilliseconds} ms"); return numReq; } /// /// Get Queue request pending /// /// Dictionary of DoorId, saveVersNumb public async Task> RequestPending() { string source = "REDIS"; long numReq = 0; Dictionary dictResult = new Dictionary(); // cerco da cache RedisKey currKey = new RedisKey(Constants.CALC_REQ_PEND); Stopwatch stopWatch = new Stopwatch(); stopWatch.Start(); numReq = redisDb.HashLength(currKey); if (numReq > 0) { var rawData = await redisDb.HashGetAllAsync(currKey); foreach (var item in rawData) { dictResult.Add($"{item.Name}", $"{item.Value}"); } } stopWatch.Stop(); TimeSpan ts = stopWatch.Elapsed; Log.Debug($"RequestPending | {source} in: {ts.TotalMilliseconds} ms"); return dictResult; } /// /// Get Queue request processing /// /// Dictionary of DoorId, saveVersNumb public async Task> RequestProcessing() { string source = "REDIS"; long numReq = 0; Dictionary dictResult = new Dictionary(); // cerco da cache RedisKey currKey = new RedisKey(Constants.CALC_REQ_PROC); Stopwatch stopWatch = new Stopwatch(); stopWatch.Start(); numReq = redisDb.HashLength(currKey); if (numReq > 0) { var rawData = await redisDb.HashGetAllAsync(currKey); foreach (var item in rawData) { dictResult.Add($"{item.Name}", $"{item.Value}"); } } stopWatch.Stop(); TimeSpan ts = stopWatch.Elapsed; Log.Debug($"RequestProcessing | {source} in: {ts.TotalMilliseconds} ms"); return dictResult; } /// /// Get Queue request pending, removing from queue and putting on processing queue /// /// Dictionary of DoorId, saveVersNumb public async Task> TakeProcessingItems(int numItems) { string source = "REDIS"; long numReq = 0; Dictionary dictResult = new Dictionary(); // cerco da cache RedisKey currKey = new RedisKey(Constants.CALC_REQ_PEND); Stopwatch stopWatch = new Stopwatch(); stopWatch.Start(); numReq = redisDb.HashLength(currKey); if (numReq > 0) { var rawData = await redisDb.HashGetAllAsync(currKey); foreach (var item in rawData) { dictResult.Add($"{item.Name}", $"{item.Value}"); } } stopWatch.Stop(); TimeSpan ts = stopWatch.Elapsed; Log.Debug($"TakeProcessingItems | {source} in: {ts.TotalMilliseconds} ms"); return dictResult; } #endregion Public Methods #region Protected Fields protected const string rKeyCalcOreFase = "Check:OreFasi"; protected const string rKeyFasiAct = "Check:FasiAct"; protected const string rKeyProjAct = "Check:ProjAct"; /// /// TTL da 1 min x cache Redis /// protected const int shortTTL = 60 * 5; protected int idxSim = 0; protected Random rnd = new Random(); #endregion Protected Fields #region Protected Methods /// /// Recupero chiave da redis /// /// /// protected async Task getRSV(string rKey) { string answ = ""; var rawData = await redisDb.StringGetAsync(rKey); if (rawData.HasValue) { answ = $"{rawData}"; } return answ; } /// /// Salvataggio chiave in redis /// /// /// /// /// protected async Task setRSV(string rKey, string rVal, int ttlSec) { bool fatto = false; await redisDb.StringSetAsync(rKey, rVal, TimeSpan.FromSeconds(ttlSec)); fatto = true; return fatto; } /// /// Salvataggio chiave in redis /// /// /// /// /// protected async Task setRSV(string rKey, int rValInt, int ttlSec) { bool fatto = false; await redisDb.StringSetAsync(rKey, rValInt, TimeSpan.FromSeconds(ttlSec)); fatto = true; return fatto; } /// /// Registra in cache chiave se non fosse già in elenco /// /// protected void trackCache(string newKey) { if (!cachedDataList.Contains(newKey)) { cachedDataList.Add(newKey); } } #endregion Protected Methods #region Private Fields private static IConfiguration _configuration = null!; private static JsonSerializerSettings? JSSettings; private static Logger Log = LogManager.GetCurrentClassLogger(); private readonly IEmailSender _emailSender; /// /// Elenco obj in cache /// private List cachedDataList = new List(); /// /// Durata cache lunga IN SECONDI /// private int cacheTtlLong = 60 * 5; /// /// Durata cache breve IN SECONDI /// private int cacheTtlShort = 60 * 1; /// /// Oggetto per connessione a REDIS /// private ConnectionMultiplexer redisConn = null!; /// /// Oggetto DB redis da impiegare x chiamate R/W /// private IDatabase redisDb = null!; #endregion Private Fields #region Private Properties /// /// Durata cache lunga (+ perturbazione percentuale +/-10%) /// private TimeSpan FastCache { get => TimeSpan.FromSeconds(cacheTtlShort * rnd.Next(900, 1100) / 1000); } /// /// Durata cache lunga (+ perturbazione percentuale +/-10%) /// private TimeSpan LongCache { get => TimeSpan.FromSeconds(cacheTtlLong * rnd.Next(900, 1100) / 1000); } /// /// Durata cache lunga (+ perturbazione percentuale +/-10%) /// private TimeSpan UltraLongCache { get => TimeSpan.FromSeconds(cacheTtlLong * 10 * rnd.Next(900, 1100) / 1000); } #endregion Private Properties #region Private Methods /// /// Esegue flush memoria redis dato pattern /// /// /// private async Task ExecFlushRedisPattern(RedisValue pattern) { bool answ = false; var listEndpoints = redisConn.GetEndPoints(); foreach (var endPoint in listEndpoints) { //var server = redisConnAdmin.GetServer(listEndpoints[0]); var server = redisConn.GetServer(endPoint); if (server != null) { var keyList = server.Keys(redisDb.Database, pattern); foreach (var item in keyList) { await redisDb.KeyDeleteAsync(item); } // brutalmente rimuovo intero contenuto DB... DANGER //await server.FlushDatabaseAsync(); answ = true; } } return answ; } #endregion Private Methods } }