namespace EgwCoreLib.Lux.Data.Services.General { /// /// Implementazione interfaccia REDIS: /// - gestione PubSub /// - gestione caching /// public class RedisService : IRedisService { #region Public Constructors public RedisService(IConnectionMultiplexer connection) { _connection = connection; _db = connection.GetDatabase(); _subscriber = connection.GetSubscriber(); // json serializer... FIX errore loop circolare https://www.ryadel.com/en/jsonserializationexception-self-referencing-loop-detected-error-fix-entity-framework-asp-net-core/ JSSettings = new JsonSerializerSettings() { ReferenceLoopHandling = ReferenceLoopHandling.Ignore }; Log.Info("Startup completed"); } #endregion Public Constructors #region Public Methods /// /// /// public bool Delete(string key) { Log.Trace($"Set request for {key}"); return _db.KeyDelete(key); } /// /// /// public async Task DeleteAsync(string key) { Log.Trace($"SetAsync request for {key}"); return await _db.KeyDeleteAsync(key); } /// /// /// public async Task FlushPatternAsync(RedisValue pattern) { try { var connServ = _connection.GetEndPoints() .Select(endpoint => _connection.GetServer(endpoint)) .Where(x => x.IsConnected && !x.IsReplica) .FirstOrDefault(); if (connServ == null) { Log.Error("Master REDIS server non trovato"); return false; } if (pattern.Equals(RedisValue.Null) || pattern.ToString() == "*") { connServ.FlushDatabase(database: _db.Database); return true; } var keys = connServ.Keys(database: _db.Database, pattern: pattern, pageSize: 1000); var deleteTasks = new List(); foreach (var key in keys) { deleteTasks.Add(_db.KeyDeleteAsync(key)); if (deleteTasks.Count >= 1000) { await Task.WhenAll(deleteTasks); deleteTasks.Clear(); } } if (deleteTasks.Count > 0) await Task.WhenAll(deleteTasks); return true; } catch (Exception ex) { Log.Error($"Eccezione durante FlushPatternAsync: {ex}"); return false; } } /// /// /// public string? Get(string key) { Log.Trace($"Get request for {key}"); var value = _db.StringGet(key); return value.HasValue ? value.ToString() : null; } /// /// /// public async Task GetAsync(string key) { Log.Trace($"GetAsync request for {key}"); var value = await _db.StringGetAsync(key); return value.HasValue ? value.ToString() : null; } /// /// /// public async Task> HashGetAllAsync(string hashKey) { Log.Trace($"HashGetAllAsync request for {hashKey}"); var results = await _db.HashGetAllAsync(hashKey); return results.ToList(); } /// /// /// public async Task HashGetAsync(string hashKey, string field) { Log.Trace($"HashGetAsync request for {hashKey}.{field}"); return await _db.HashGetAsync(hashKey, field); } /// /// /// public long Publish(string channel, string message) { Log.Trace($"Publish: channel {channel}"); RedisChannel rChannel = new RedisChannel(channel, RedisChannel.PatternMode.Literal); long numCli = _subscriber.Publish(rChannel, message); return numCli; } /// /// /// public async Task PublishAsync(string channel, string message) { Log.Trace($"PublishAsync: channel {channel}"); RedisChannel rChannel = new RedisChannel(channel, RedisChannel.PatternMode.Literal); long numCli = await _subscriber.PublishAsync(rChannel, message); return numCli; } /// /// /// public long QueueCount(RedisKey queueName) { return _db.ListLength(queueName); } /// /// /// public async Task QueueCountAsync(RedisKey queueName) { return await _db.ListLengthAsync(queueName); } /// /// /// public List QueueListAll(RedisKey queueName) { // lettura + reset in blocco var listData = _db.ListRange(queueName, 0, -1).ToList(); return listData; } /// /// /// public async Task> QueueListAllAsync(RedisKey queueName) { // lettura + reset in blocco var listData = await _db.ListRangeAsync(queueName, 0, -1); return listData.ToList(); } /// /// /// public RedisValue QueuePop(RedisKey queueName) { return _db.ListLeftPop(queueName); } /// /// /// public List QueuePopAll(RedisKey queueName) { // lettura + reset in blocco var listData = _db.ListRange(queueName, 0, -1).ToList(); if (listData.Count > 0) { _db.KeyDelete(queueName); // remove the entire list } return listData; } /// /// /// public async Task> QueuePopAllAsync(RedisKey queueName) { // lettura + reset in blocco var rawData = await _db.ListRangeAsync(queueName, 0, -1); var listData = rawData.ToList(); if (listData.Count > 0) { _db.KeyDelete(queueName); // remove the entire list } return listData; } /// /// /// public async Task QueuePopAsync(RedisKey queueName) { return await _db.ListLeftPopAsync(queueName); } /// /// /// public List QueuePopList(RedisKey queueName, int maxElem) { // nuovo metodo con rimozione var results = new List(maxElem); for (int i = 0; i < maxElem; i++) { var item = _db.ListLeftPop(queueName); if (item.IsNull) break; // queue empty results.Add(item); } return results; } /// /// /// public async Task> QueuePopListAsync(RedisKey queueName, int maxElem) { // nuovo metodo con rimozione var results = new List(maxElem); for (int i = 0; i < maxElem; i++) { var item = await _db.ListLeftPopAsync(queueName); if (item.IsNull) break; // queue empty results.Add(item); } return results; } /// /// /// public long QueuePush(RedisKey queueName, RedisValue value) { long qLen = _db.ListRightPush(queueName, value); return qLen; } /// /// /// public async Task QueuePushAsync(RedisKey queueName, RedisValue value) { long qLen = await _db.ListRightPushAsync(queueName, value); return qLen; } /// /// /// public long QueueRemove(RedisKey queueName, RedisValue value) { // count = 0 → rimuove TUTTE le occorrenze del valore return _db.ListRemove(queueName, value, 0); } /// /// /// public async Task QueueRemoveAsync(RedisKey queueName, RedisValue value) { // count = 0 → rimuove TUTTE le occorrenze del valore return await _db.ListRemoveAsync(queueName, value, 0); } /// /// /// public bool QueueReset(RedisKey queueName) { bool answ = _db.KeyDelete(queueName); return answ; } /// /// /// public async Task QueueResetAsync(RedisKey queueName) { bool answ = await _db.KeyDeleteAsync(queueName); return answ; } /// /// /// public bool Set(string key, string value, TimeSpan? tsExpiry = null) { Log.Trace($"Set request for {key}"); return _db.StringSet(key, value, tsExpiry, When.Always); } /// /// /// public async Task SetAsync(string key, string value, TimeSpan? tsExpiry = null) { Log.Trace($"SetAsync request for {key}"); return await _db.StringSetAsync(key, value, tsExpiry, When.Always); } /// /// /// public void Subscribe(string channel, Action handler) { Log.Trace($"Subscribed to channel {channel}"); RedisChannel rChannel = new RedisChannel(channel, RedisChannel.PatternMode.Literal); _subscriber.Subscribe(rChannel, handler); } #endregion Public Methods #region Private Fields private static Logger Log = LogManager.GetCurrentClassLogger(); private readonly IConnectionMultiplexer _connection; private readonly IDatabase _db; private readonly ISubscriber _subscriber; /// /// conf speciale serializzatore JSON /// private JsonSerializerSettings? JSSettings; #endregion Private Fields } }