lux/EgwCoreLib.Lux.Data/Services/ProdService.cs

using EgwCoreLib.Lux.Core.RestPayload;
using EgwCoreLib.Lux.Data.Controllers;
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 System.Linq;
using System.Text;
using System.Threading.Tasks;

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;
            chPub = _config.GetValue<string>("ServerConf:ChannelPub") ?? "";
            queueKey = (RedisKey)$"{redisBaseKey}:OrderQueue";
            redisOrderReqKey = $"{redisBaseKey}:OrderReq";
            redisOrderRunKey = $"{redisBaseKey}:OrderRun";
            redisOrderDoneKey = $"{redisBaseKey}:OrderDone";
            Log.Info($"ProdService | Init OK");
        }

        #endregion Public Constructors

        #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 = $"{redisOrderReqKey}:{reqUid.Replace("/", ":")}";
            done = await _redisService.SetAsync(currKey, calcRequest.sProcessArgs);
            // accodo la nuova richiesta
            //RedisKey queueKey = (RedisKey)$"{redisBaseKey}:OrderQueue:{reqType}";
            _redisService.QueuePush(queueKey, (RedisValue)reqUid);
            // dizionario richieste: è il serializzato dell'elenco degli UID da calcolare...
            var currList = await _redisService.QueueListAllAsync(queueKey);
            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
        /// </summary>
        /// <param name="id"></param>
        /// <returns></returns>
        public async Task<string> GetJob(string id)
        {
            Stopwatch sw = new Stopwatch();
            sw.Start();
            string result = "";
            // recupero richiesta serializzata
            string currKey = $"{redisOrderReqKey}:{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
        /// </summary>
        /// <returns></returns>
        public async Task<string> GetNext()
        {
            Stopwatch sw = new Stopwatch();
            sw.Start();
            string result = "";
            // prendo dalla coda primo job (rimuovendolo...)
            var rawReq = await _redisService.QueuePopAsync(queueKey);
            if (rawReq.HasValue)
            {
                string reqUid = $"{rawReq}";
                // metto UID in coda running
                _redisService.QueuePush(queueKey, (RedisValue)reqUid);
                // FixMe ToDo !!!: salvataggio data-ora per indicare avvio calcolo...

                // recupero richiesta serializzata
                string currKey = $"{redisOrderReqKey}:{reqUid.Replace("/", ":")}";
                var rawRes = await _redisService.GetAsync(currKey);
                if (!string.IsNullOrEmpty(rawRes))
                {
                    result = rawRes;
                }
            }
            sw.Stop();
            Log.Info($"GetNext | {sw.Elapsed.TotalMilliseconds:N3} ms");
            return result;
        }

        /// <summary>
        /// Numero di Job in attesa su coda
        /// </summary>
        /// <returns></returns>
        public async Task<long> QueueLen()
        {
            Stopwatch sw = new Stopwatch();
            sw.Start();
            long numWaiting = await _redisService.QueueCountAsync(queueKey);
            sw.Stop();
            Log.Info($"QueueLen | {sw.Elapsed.TotalMilliseconds:N3} ms");
            return numWaiting;
        }

        #endregion Public Methods

        #region Private Fields

        private static Logger Log = LogManager.GetCurrentClassLogger();
        private readonly IRedisService _redisService;
        private readonly string chPub = "";

        /// <summary>
        /// Key della coda redis delle richieste x PROD Engine
        /// </summary>
        private RedisKey queueKey;

        private string redisBaseKey = "Lux:Prod";
        private string redisOrderDoneKey = "";
        private string redisOrderReqKey = "";
        private string redisOrderRunKey = "";

        #endregion Private Fields
    }
}