Samuele Locatelli
409 lines
16 KiB
C#
409 lines
16 KiB
C#
using EgwCoreLib.Lux.Core.Generic;
|
|
using EgwCoreLib.Lux.Core.RestPayload;
|
|
using EgwMultiEngineManager.Data;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Newtonsoft.Json;
|
|
using NLog;
|
|
using StackExchange.Redis;
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using System.Diagnostics;
|
|
using static Egw.Window.Data.Enums;
|
|
|
|
namespace EgwCoreLib.Lux.Data.Services
|
|
{
|
|
public class ProdService : BaseServ
|
|
{
|
|
#region Public Constructors
|
|
|
|
public ProdService(IConfiguration configuration, IConnectionMultiplexer RedisConn, IRedisService redisService) : base(configuration, RedisConn)
|
|
{
|
|
// conf redis service
|
|
_redisService = redisService;
|
|
_db = redisConn.GetDatabase();
|
|
chPub = _config.GetValue<string>("ServerConf:ChannelPub") ?? "";
|
|
Log.Info($"ProdService | Init OK");
|
|
}
|
|
|
|
private readonly IDatabase _db;
|
|
#endregion Public Constructors
|
|
|
|
#region Public Enums
|
|
|
|
public enum QueueType
|
|
{
|
|
waiting,
|
|
running,
|
|
done
|
|
}
|
|
|
|
#endregion Public Enums
|
|
|
|
#region Public Methods
|
|
|
|
/// <summary>
|
|
/// Accoda una richiesta di calcolo
|
|
/// </summary>
|
|
/// <param name="reqType"></param>
|
|
/// <param name="reqUid"></param>
|
|
/// <param name="currRequest"></param>
|
|
/// <returns></returns>
|
|
public async Task<bool> EnqueueRequest(string reqType, string reqUid, CalcRequestDTO currRequest)
|
|
{
|
|
bool done = false;
|
|
int nId = 1;
|
|
// salvo su cache x successivo reinvio da currRequest
|
|
QuestionDTO calcRequest = new QuestionDTO(nId, currRequest.EnvType, currRequest.DictExec);
|
|
// salvo in cache contenuto della richiesta x UID
|
|
string currKey = $"{redisBaseKey}:{currRequest.EnvType}:{reqUid.Replace("/", ":")}";
|
|
done = await _redisService.SetAsync(currKey, calcRequest.sProcessArgs);
|
|
// invio ed accodo!
|
|
var qWaitKey = qKey(currRequest.EnvType, QueueType.waiting);
|
|
_redisService.QueuePush(qWaitKey, (RedisValue)reqUid);
|
|
// dizionario richieste: è il serializzato dell'elenco degli UID da calcolare...
|
|
List<RedisValue> currList = await _redisService.QueueListAllAsync(qWaitKey);
|
|
Dictionary<string, string> calcDict = new Dictionary<string, string>();
|
|
calcDict.Add("ReqLen", $"{calcDict.Count}");
|
|
string listReq = JsonConvert.SerializeObject(currList);
|
|
calcDict.Add("ReqList", listReq);
|
|
// preparo richiesta di calcolo x UID da inviare
|
|
QuestionDTO chRequest = new QuestionDTO(nId, currRequest.EnvType, calcDict);
|
|
// invio sul channel redis della richiesta di processing
|
|
await _redisService.PublishAsync(chPub, chRequest.sProcessArgs);
|
|
// ritorno
|
|
return done;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Restituzione singolo Job specifico dalla coda ORDINI
|
|
/// </summary>
|
|
/// <param name="cEnv">environment richiesto</param>
|
|
/// <param name="id"></param>
|
|
/// <returns></returns>
|
|
public async Task<string> GetJob(Constants.EXECENVIRONMENTS cEnv, string id)
|
|
{
|
|
Stopwatch sw = new Stopwatch();
|
|
sw.Start();
|
|
string result = "";
|
|
// recupero richiesta serializzata
|
|
string currKey = $"{redisBaseKey}:{cEnv}:{id.Replace("/", ":")}";
|
|
var rawRes = await _redisService.GetAsync(currKey);
|
|
if (!string.IsNullOrEmpty(rawRes))
|
|
{
|
|
result = rawRes;
|
|
}
|
|
sw.Stop();
|
|
Log.Info($"GetJob | {id} | {sw.Elapsed.TotalMilliseconds:N3} ms");
|
|
return result;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Restituisce il contenuto della prox richiesta da eseguire x Environment
|
|
/// </summary>
|
|
/// <param name="cEnv">environment richiesto</param>
|
|
/// <param name="qType">tipo coda (default waiting)</param>
|
|
/// <returns></returns>
|
|
public async Task<string> GetNext(Constants.EXECENVIRONMENTS cEnv, QueueType qType = QueueType.waiting)
|
|
{
|
|
Stopwatch sw = new Stopwatch();
|
|
sw.Start();
|
|
string result = "";
|
|
RedisKey rKey = qKey(cEnv, qType);
|
|
// prendo dalla coda primo job (rimuovendolo...)
|
|
var rawReq = await _redisService.QueuePopAsync(rKey);
|
|
if (rawReq.HasValue)
|
|
{
|
|
string reqUid = $"{rawReq}";
|
|
// metto UID in coda running nella successiva.. se non ultima...
|
|
if (qType < QueueType.running)
|
|
{
|
|
// FixMe ToDo !!!: salvataggio data-ora per indicare avvio calcolo...
|
|
_redisService.QueuePush(qKey(cEnv, qType + 1), (RedisValue)reqUid);
|
|
}
|
|
// recupero richiesta serializzata
|
|
string currKey = $"{redisBaseKey}:{cEnv}:{reqUid.Replace("/", ":")}";
|
|
var rawRes = await _redisService.GetAsync(currKey);
|
|
if (!string.IsNullOrEmpty(rawRes))
|
|
{
|
|
result = rawRes;
|
|
}
|
|
}
|
|
sw.Stop();
|
|
Log.Info($"GetNext | {rKey} | {sw.Elapsed.TotalMilliseconds:N3} ms");
|
|
return result;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifica un job in una coda, se trovato lo "avanza" a fatto (per estimation/balance, ...)
|
|
/// </summary>
|
|
/// <param name="cEnv">environment richiesto</param>
|
|
/// <param name="codJob">chiave da cercare x avanzarla</param>
|
|
/// <param name="qType">coda di default a run --> done</param>
|
|
/// <returns></returns>
|
|
public async Task<bool> SetDone(Constants.EXECENVIRONMENTS cEnv, string codJob, QueueType qType = QueueType.running)
|
|
{
|
|
Stopwatch sw = new Stopwatch();
|
|
sw.Start();
|
|
bool fatto = false;
|
|
// recupero lista completa SENZA eliminare...
|
|
var listRunning = await _redisService.QueueListAllAsync(qKey(cEnv, QueueType.running));
|
|
if (listRunning != null && listRunning.Contains(codJob))
|
|
{
|
|
var numRem = await _redisService.QueueRemoveAsync(qKey(cEnv, QueueType.running), (RedisValue)codJob);
|
|
if (numRem > 0)
|
|
{
|
|
// elimino eventuali duplicati sulla coda di destinazione
|
|
var listDone = await _redisService.QueueListAllAsync(qKey(cEnv, QueueType.done));
|
|
if (listDone != null && listDone.Contains(codJob))
|
|
{
|
|
await _redisService.QueueRemoveAsync(qKey(cEnv, QueueType.done), (RedisValue)codJob);
|
|
}
|
|
// accodo nuovi!
|
|
var numAdd = _redisService.QueuePush(qKey(cEnv, QueueType.done), codJob);
|
|
fatto = numAdd > 0;
|
|
}
|
|
}
|
|
sw.Stop();
|
|
Log.Info($"SetDone | env: {cEnv} | qFrom: {qKey(cEnv, QueueType.running)} | reqUid: {codJob} | fatto: {fatto} | {sw.Elapsed.TotalMilliseconds:N3} ms");
|
|
return fatto;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Numero di Job in attesa su coda x environment e tipo
|
|
/// </summary>
|
|
/// <param name="cEnv">environment richiesto</param>
|
|
/// <param name="qType">tipo coda (default waiting)</param>
|
|
/// <returns></returns>
|
|
public async Task<long> QueueLen(Constants.EXECENVIRONMENTS cEnv, QueueType qType = QueueType.waiting)
|
|
{
|
|
Stopwatch sw = new Stopwatch();
|
|
sw.Start();
|
|
RedisKey rKey = qKey(cEnv, qType);
|
|
long numWaiting = await _redisService.QueueCountAsync(rKey);
|
|
sw.Stop();
|
|
Log.Info($"QueueLen | {rKey} | {sw.Elapsed.TotalMilliseconds:N3} ms");
|
|
return numWaiting;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Restituisce elenco corrente delle richieste x environment e tipo
|
|
/// </summary>
|
|
/// <param name="cEnv">environment richiesto</param>
|
|
/// <param name="qType">tipo coda</param>
|
|
/// <returns></returns>
|
|
public async Task<List<RedisValue>> QueueListAllAsync(Constants.EXECENVIRONMENTS cEnv, QueueType qType)
|
|
{
|
|
RedisKey rKey = qKey(cEnv, qType);
|
|
List<RedisValue> currList = await _redisService.QueueListAllAsync(rKey);
|
|
return currList;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reset coda del tipo richiesto...
|
|
/// </summary>
|
|
/// <param name="cEnv">environment richiesto</param>
|
|
/// <param name="qType">tipo coda</param>
|
|
/// <returns></returns>
|
|
public bool QueueReset(Constants.EXECENVIRONMENTS cEnv, QueueType qType)
|
|
{
|
|
RedisKey rKey = qKey(cEnv, qType);
|
|
bool answ = _redisService.QueueReset(rKey);
|
|
return answ;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reset coda del tipo richiestoversione Async...
|
|
/// </summary>
|
|
/// <param name="cEnv">environment richiesto</param>
|
|
/// <param name="qType">tipo coda</param>
|
|
/// <returns></returns>
|
|
public async Task<bool> QueueResetAsync(Constants.EXECENVIRONMENTS cEnv, QueueType qType)
|
|
{
|
|
RedisKey rKey = qKey(cEnv, qType);
|
|
bool answ = await _redisService.QueueResetAsync(rKey);
|
|
return answ;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reset coda running (riaccoda job su waiting)...
|
|
/// </summary>
|
|
/// <param name="cEnv">environment richiesto</param>
|
|
/// <returns></returns>
|
|
public async Task<bool> ResetRunning(Constants.EXECENVIRONMENTS cEnv)
|
|
{
|
|
bool fatto = false;
|
|
RedisKey rKey = qKey(cEnv, QueueType.running);
|
|
var listRunning = await _redisService.QueuePopAllAsync(rKey);
|
|
if (listRunning != null)
|
|
{
|
|
foreach (var item in listRunning)
|
|
{
|
|
if (item.HasValue)
|
|
{
|
|
// FixMe ToDo !!!: salvataggio data-ora per indicare avvio calcolo...
|
|
_redisService.QueuePush(qKey(cEnv, QueueType.waiting), item);
|
|
}
|
|
}
|
|
}
|
|
return fatto;
|
|
}
|
|
|
|
|
|
/// <summary>
|
|
/// Rimette un run un singolo job eliminando da coda Run e mettendo in coda waiting
|
|
/// </summary>
|
|
/// <param name="cEnv">environment richiesto</param>
|
|
/// <param name="codJob">cod Job da resettare</param>
|
|
/// <returns></returns>
|
|
public async Task<bool> ReRunJob(Constants.EXECENVIRONMENTS cEnv, string codJob)
|
|
{
|
|
bool fatto = false;
|
|
RedisKey rKey = qKey(cEnv, QueueType.running);
|
|
// recupero lista completa SENZA eliminare...
|
|
var listRunning = await _redisService.QueueListAllAsync(rKey);
|
|
// se non trova usa coda Done...
|
|
if (listRunning == null || !listRunning.Contains(codJob))
|
|
{
|
|
rKey = qKey(cEnv, QueueType.done);
|
|
// recupero lista completa SENZA eliminare...
|
|
listRunning = await _redisService.QueueListAllAsync(rKey);
|
|
}
|
|
if (listRunning != null && listRunning.Contains(codJob))
|
|
{
|
|
var numRem = await _redisService.QueueRemoveAsync(rKey, (RedisValue)codJob);
|
|
if (numRem > 0)
|
|
{
|
|
var numAdd = _redisService.QueuePush(qKey(cEnv, QueueType.waiting), codJob);
|
|
fatto = numAdd > 0;
|
|
}
|
|
}
|
|
return fatto;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Richiesta di esecuzione di un job di tipo Balance
|
|
/// </summary>
|
|
/// <param name="codJob"></param>
|
|
/// <param name="balData"></param>
|
|
/// <returns></returns>
|
|
public async Task<bool> BalanceJob(string codJob, BalanceReqDto balData)
|
|
{
|
|
bool fatto = false;
|
|
#if false
|
|
RedisKey rKey = qKey(QueueType.running);
|
|
// recupero lista completa SENZA eliminare...
|
|
var listRunning = await _redisService.QueueListAllAsync(rKey);
|
|
if (listRunning != null && listRunning.Contains(codJob))
|
|
{
|
|
var numRem = await _redisService.QueueRemoveAsync(rKey, (RedisValue)codJob);
|
|
if (numRem > 0)
|
|
{
|
|
var numAdd = _redisService.QueuePush(qKey(QueueType.waiting), codJob);
|
|
fatto = numAdd > 0;
|
|
}
|
|
}
|
|
#endif
|
|
return fatto;
|
|
}
|
|
|
|
#endregion Public Methods
|
|
|
|
#region Private Fields
|
|
|
|
private static Logger Log = LogManager.GetCurrentClassLogger();
|
|
|
|
private readonly IRedisService _redisService;
|
|
|
|
private readonly string chPub = "";
|
|
|
|
|
|
private string redisBaseKey = "Lux:Prod";
|
|
|
|
|
|
#endregion Private Fields
|
|
|
|
#region Private Methods
|
|
|
|
/// <summary>
|
|
/// Calcolo coda dato Environment e tipo coda
|
|
/// </summary>
|
|
/// <param name="cEnvir"></param>
|
|
/// <param name="qType"></param>
|
|
/// <returns></returns>
|
|
private string qKey(Constants.EXECENVIRONMENTS cEnvir, QueueType qType)
|
|
{
|
|
RedisKey rKey = (RedisKey)$"{redisBaseKey}:{cEnvir}:CalcQueue";
|
|
switch (qType)
|
|
{
|
|
case QueueType.waiting:
|
|
rKey = (RedisKey)$"{redisBaseKey}:{cEnvir}:CalcQueue:Wait";
|
|
break;
|
|
|
|
case QueueType.running:
|
|
rKey = (RedisKey)$"{redisBaseKey}:{cEnvir}:CalcQueue:Run";
|
|
break;
|
|
|
|
case QueueType.done:
|
|
rKey = (RedisKey)$"{redisBaseKey}:{cEnvir}:CalcQueue:Done";
|
|
break;
|
|
|
|
default:
|
|
//rKey = (RedisKey)$"{redisBaseKey}:{cEnvir}:CalcQueue";
|
|
break;
|
|
}
|
|
return rKey;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Esegue avanzamento nella coda logica tra wait/run/done se possibile...
|
|
/// </summary>
|
|
/// <param name="hashKey">Chiave hash del contenuto richiesta RUID</param>
|
|
/// <returns></returns>
|
|
public async Task UpdateQueue(Constants.EXECENVIRONMENTS cEnvir, string hashKey)
|
|
{
|
|
// recupero info accessorie da redis
|
|
var tipo = await _db.HashGetAsync(hashKey, "tipo");
|
|
|
|
// verifico SE si tratta di un create/estimate e nel caso processo coda req x sistemare
|
|
if (!tipo.IsNull && !string.IsNullOrEmpty(tipo))
|
|
{
|
|
string rKey = "";
|
|
var UID = await _db.HashGetAsync(hashKey, "UID");
|
|
var parti = ((string)tipo).Split('-');
|
|
if (parti.Length == 2)
|
|
{
|
|
if (parti[0] == "ORDER")
|
|
{
|
|
if (parti.Length == 2)
|
|
{
|
|
rKey = $"{parti[0]}:{parti[1]}:{UID}";
|
|
}
|
|
}
|
|
// cerco solo tipo create / 5...
|
|
else if (parti[0] == "5")
|
|
{
|
|
if (int.TryParse(parti[0], out int iMode) && int.TryParse(parti[1], out int iSubMode))
|
|
{
|
|
// Validazione e Cast in un colpo solo
|
|
bool modeValido = Enum.IsDefined(typeof(QuestionModes), iMode);
|
|
bool subModeValido = Enum.IsDefined(typeof(QuestionOrderSubModes), iSubMode);
|
|
|
|
if (modeValido && subModeValido)
|
|
{
|
|
var cMode = (QuestionModes)iMode;
|
|
var cSubMode = (QuestionOrderSubModes)iSubMode;
|
|
rKey = $"{cMode}:{cSubMode}:{UID}";
|
|
}
|
|
}
|
|
}
|
|
if (!string.IsNullOrEmpty(rKey))
|
|
{
|
|
await SetDone(cEnvir, rKey, QueueType.running);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#endregion Private Methods
|
|
}
|
|
} |