hanzo/jube-fork
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #130

Browse Source 
129 Fix partial implementation in Polly Wrapper for Redis
This commit is contained in:
Richard Churchman
2026-03-22 11:03:20 +01:00
committed by GitHub
parent 71963dea84 4481615382
commit 7c9919bb51
12 changed files with 218 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Jube.Cache/CacheService.cs
+3 -2
View File
@@ -17,6 +17,7 @@ namespace Jube.Cache
using log4net;
using Redis;
using Redis.Callback;
using ResilientRedisConnection;
using StackExchange.Redis;
using TaskCancellation;
@@ -52,12 +53,12 @@ namespace Jube.Cache
ConnectionMultiplexer =
ConnectionMultiplexer.Connect(redisConnectionString);
RedisDatabase = ConnectionMultiplexer.GetDatabase();
RedisDatabase = new ResilientRedisConnection(ConnectionMultiplexer, log).GetDatabase();
}
public Task InstantiateRepositoriesTask { get; set; }
public ConnectionMultiplexer ConnectionMultiplexer { get; set; }
public IDatabase RedisDatabase { get; set; }
public ResilientRedisDatabase RedisDatabase { get; set; }
public CacheAbstractionRepository CacheAbstractionRepository { get; set; }
public CachePayloadLatestRepository CachePayloadLatestRepository { get; set; }
public CachePayloadRepository CachePayloadRepository { get; set; }
Jube.Cache/Redis/CacheAbstractionRepository.cs
+2 -1
View File
@@ -16,10 +16,11 @@ namespace Jube.Cache.Redis
using Interfaces;
using log4net;
using Models;
using ResilientRedisConnection;
using StackExchange.Redis;
public class CacheAbstractionRepository(
IDatabaseAsync redisDatabase,
ResilientRedisDatabase redisDatabase,
ILog log,
CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheAbstractionRepository
{
Jube.Cache/Redis/CacheCallbackPublishSubscribe.cs
+3 -2
View File
@@ -16,6 +16,7 @@ namespace Jube.Cache.Redis
using System.Collections.Concurrent;
using System.Net;
using log4net;
using ResilientRedisConnection;
using StackExchange.Redis;
using TaskCancellation;
@@ -30,10 +31,10 @@ namespace Jube.Cache.Redis
private readonly ConnectionMultiplexer connectionMultiplexer;
private readonly string localCacheInstanceGuidString;
private readonly ILog log;
private readonly IDatabaseAsync redisDatabase;
private readonly ResilientRedisDatabase redisDatabase;
public CacheCallbackPublishSubscribe(ConnectionMultiplexer connectionMultiplexer,
IDatabaseAsync redisDatabase,
ResilientRedisDatabase redisDatabase,
ConcurrentDictionary<Guid, TaskCompletionSource<Callback.Callback>> callbacks,
int callbackTimeout,
ILog log,
Jube.Cache/Redis/CachePayloadLatestRepository.cs
+5 -4
View File
@@ -25,13 +25,14 @@ namespace Jube.Cache.Redis
using log4net;
using MessagePack;
using Models;
using ResilientRedisConnection;
using Serialization;
using StackExchange.Redis;
using TaskCancellation.TaskHelper;
public class CachePayloadLatestRepository(
string postgresConnectionString,
IDatabaseAsync redisDatabase,
ResilientRedisDatabase redisDatabase,
ILog log,
CommandFlags commandFlag = CommandFlags.FireAndForget) : ICachePayloadLatestRepository
{
@@ -191,7 +192,7 @@ namespace Jube.Cache.Redis
var updateLatestTask = redisDatabase.SortedSetUpdateAsync(redisKeyReferenceDateLatest, redisHSetKey, referenceDateTimestamp);
await redisDatabase.HashIncrementAsync(redisKeyPayloadLatestCount, entryKey);
await redisDatabase.HashIncrementAsync(redisKeyPayloadLatestCount, entryKey, 1);
await redisDatabase.HashSetAsync(redisKeyPayloadLatest, redisHSetKey, bytes);
await updateLatestTask;
@@ -286,7 +287,7 @@ namespace Jube.Cache.Redis
}
}
private static async Task BatchSortedSetRemoveAsync(IDatabaseAsync db, RedisKey key, IEnumerable<RedisValue> values, CommandFlags flags)
private static async Task BatchSortedSetRemoveAsync(ResilientRedisDatabase db, RedisKey key, IEnumerable<RedisValue> values, CommandFlags flags)
{
const int batchSize = 1000;
var valuesArray = values.ToArray();
@@ -300,7 +301,7 @@ namespace Jube.Cache.Redis
}
}
private static async Task BatchHashDeleteAsync(IDatabaseAsync db, RedisKey key, IEnumerable<RedisValue> fields, CommandFlags flags)
private static async Task BatchHashDeleteAsync(ResilientRedisDatabase db, RedisKey key, IEnumerable<RedisValue> fields, CommandFlags flags)
{
const int batchSize = 1000;
var fieldsArray = fields.ToArray();
Jube.Cache/Redis/CachePayloadRepository.cs
+7 -6
View File
@@ -24,6 +24,7 @@ namespace Jube.Cache.Redis
using Interfaces;
using log4net;
using MessagePack;
using ResilientRedisConnection;
using Serialization;
using Serialization.DictionaryNoBoxing.MessagePack;
using StackExchange.Redis;
@@ -43,7 +44,7 @@ namespace Jube.Cache.Redis
private readonly MessagePackSerializerOptions messagePackSerializerOptions;
private readonly string postgresConnectionString;
private readonly bool publishSubscribe;
private readonly IDatabaseAsync redisDatabase;
private readonly ResilientRedisDatabase redisDatabase;
private readonly bool storePayloadCountsAndBytes;
private readonly SemaphoreSlim timerSemaphore = new SemaphoreSlim(1, 1);
private LocalCacheInstance localCacheInstance;
@@ -52,7 +53,7 @@ namespace Jube.Cache.Redis
private LruCacheConcurrentSizedDictionary<string, byte[]> lruCacheConcurrentSizedDictionary;
private Timer timer;
private CachePayloadRepository(ConnectionMultiplexer connectionMultiplexer, IDatabaseAsync redisDatabase,
private CachePayloadRepository(ConnectionMultiplexer connectionMultiplexer, ResilientRedisDatabase redisDatabase,
string postgresConnectionString, ILog log,
CommandFlags commandFlag, bool fill, bool localCache, long localCacheBytes, bool messagePackCompression, bool storePayloadCountsAndBytes,
bool publishSubscribe, CancellationToken token = default)
@@ -120,7 +121,7 @@ namespace Jube.Cache.Redis
{
var redisKeyPayloadCount = $"PayloadCount:{tenantRegistryId}";
var redisKeyPayloadBytes = $"PayloadBytes:{tenantRegistryId}";
tasks.Add(redisDatabase.HashIncrementAsync(redisKeyPayloadCount, entityAnalysisModelGuid.ToString("N")));
tasks.Add(redisDatabase.HashIncrementAsync(redisKeyPayloadCount, entityAnalysisModelGuid.ToString("N"), 1));
tasks.Add(redisDatabase.HashIncrementAsync(redisKeyPayloadBytes, entityAnalysisModelGuid.ToString("N"), bytes.Length));
}
@@ -299,7 +300,7 @@ namespace Jube.Cache.Redis
public static async Task<CachePayloadRepository> CreateAsync(
ConnectionMultiplexer connectionMultiplexer,
IDatabaseAsync redisDatabase,
ResilientRedisDatabase redisDatabase,
string postgresConnectionString,
ILog log,
CommandFlags commandFlag,
@@ -826,7 +827,7 @@ namespace Jube.Cache.Redis
)));
tasks.Add(TaskHelper.MeasureTaskTimeAndMemoryAllocatedAsync(TaskType.HashDecrementCount, async () => await redisDatabase.HashDecrementAsync(
redisKeyCount, redisHashKeyForRedisKey
redisKeyCount, redisHashKeyForRedisKey, 1
)));
tasks.Add(TaskHelper.MeasureTaskTimeAndMemoryAllocatedAsync(TaskType.HashDeletePayload, async () => await redisDatabase.HashDeleteAsync(
@@ -1073,7 +1074,7 @@ namespace Jube.Cache.Redis
}
private async Task BatchSortedSetRemoveAsync(IDatabaseAsync db, RedisKey key, IEnumerable<RedisValue> values)
private async Task BatchSortedSetRemoveAsync(ResilientRedisDatabase db, RedisKey key, IEnumerable<RedisValue> values)
{
const int batchSize = 1000;
var valuesArray = values.ToArray();
Jube.Cache/Redis/CacheReferenceDate.cs
+2 -1
View File
@@ -16,10 +16,11 @@ namespace Jube.Cache.Redis
using Extensions;
using Interfaces;
using log4net;
using ResilientRedisConnection;
using StackExchange.Redis;
public class CacheReferenceDate(
IDatabaseAsync redisDatabase,
ResilientRedisDatabase redisDatabase,
ILog log,
CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheReferenceDate
{
Jube.Cache/Redis/CacheSanctionRepository.cs
+2 -1
View File
@@ -17,11 +17,12 @@ namespace Jube.Cache.Redis
using log4net;
using MessagePack;
using Models;
using ResilientRedisConnection;
using Serialization;
using StackExchange.Redis;
public class CacheSanctionRepository(
IDatabaseAsync redisDatabase,
ResilientRedisDatabase redisDatabase,
ILog log,
CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheSanctionRepository
{
Jube.Cache/Redis/CacheTtlCounterEntryRepository.cs
+3 -2
View File
@@ -17,10 +17,11 @@ namespace Jube.Cache.Redis
using Interfaces;
using log4net;
using Models;
using ResilientRedisConnection;
using StackExchange.Redis;
public class CacheTtlCounterEntryRepository(
IDatabaseAsync redisDatabase,
ResilientRedisDatabase redisDatabase,
ILog log,
CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheTtlCounterEntryRepository
{
@@ -47,7 +48,7 @@ namespace Jube.Cache.Redis
{
continue;
}
if (keyTtlCounterEntry.Value.HasValue)
{
expired.Add(new ExpiredTtlCounterEntry
Jube.Cache/Redis/CacheTtlCounterRepository.cs
+2 -1
View File
@@ -15,10 +15,11 @@ namespace Jube.Cache.Redis
{
using Interfaces;
using log4net;
using ResilientRedisConnection;
using StackExchange.Redis;
public class CacheTtlCounterRepository(
IDatabaseAsync redisDatabase,
ResilientRedisDatabase redisDatabase,
ILog log,
CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheTtlCounterRepository
{
Jube.Cache/Redis/CacheWalRepository.cs
+2 -1
View File
@@ -15,10 +15,11 @@ namespace Jube.Cache.Redis
{
using Interfaces;
using log4net;
using ResilientRedisConnection;
using StackExchange.Redis;
public class CacheWalRepository(
IDatabaseAsync redisDatabase,
ResilientRedisDatabase redisDatabase,
ILog log,
CommandFlags commandFlag = CommandFlags.FireAndForget) : ICacheWalRepository
{
Jube.ResilientRedisConnection/ResilientRedisConnection.cs
+48 -18
View File
@@ -21,36 +21,66 @@ namespace Jube.ResilientRedisConnection
{
private readonly IConnectionMultiplexer multiplexer;
public ResilientRedisConnection(IConnectionMultiplexer multiplexer, ILog log, int maxRetries = 10)
public ResilientRedisConnection(IConnectionMultiplexer multiplexer, ILog log, int maxRetries = 5)
{
this.multiplexer = multiplexer ?? throw new ArgumentNullException(nameof(multiplexer));
FailoverPolicy = Policy
var circuitBreaker = Policy
.Handle<RedisConnectionException>()
.Or<RedisTimeoutException>()
.Or<RedisServerException>(ex =>
ex.Message.StartsWith("LOADING", StringComparison.OrdinalIgnoreCase) ||
ex.Message.StartsWith("MASTERDOWN", StringComparison.OrdinalIgnoreCase) ||
ex.Message.StartsWith("CLUSTERDOWN", StringComparison.OrdinalIgnoreCase) ||
ex.Message.StartsWith("TRYAGAIN", StringComparison.OrdinalIgnoreCase) ||
ex.Message.StartsWith("MOVED", StringComparison.OrdinalIgnoreCase) ||
ex.Message.StartsWith("ASK", StringComparison.OrdinalIgnoreCase))
ex.Message.Contains("LOADING", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("MASTERDOWN", StringComparison.OrdinalIgnoreCase))
.CircuitBreakerAsync(5, TimeSpan.FromSeconds(30));
var fullRetry = Policy
.Handle<RedisConnectionException>()
.Or<RedisTimeoutException>()
.Or<RedisServerException>(ex =>
ex.Message.Contains("LOADING", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("MASTERDOWN", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("CLUSTERDOWN", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("TRYAGAIN", StringComparison.OrdinalIgnoreCase))
.Or<RedisException>(ex =>
ex.Message.Contains("No connection is available", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("multiplexer is not connected", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("The message was already in a completed state", StringComparison.OrdinalIgnoreCase))
ex.Message.Contains("multiplexer is not connected", StringComparison.OrdinalIgnoreCase))
.WaitAndRetryAsync(
maxRetries,
attempt =>
TimeSpan.FromSeconds(Math.Min(Math.Pow(2, attempt), 10)) +
TimeSpan.FromMilliseconds(Random.Shared.Next(0, 500)),
(ex, delay, retryCount, _) =>
attempt => TimeSpan.FromSeconds(Math.Min(Math.Pow(2, attempt), 10)) +
TimeSpan.FromMilliseconds(Random.Shared.Next(0, 500)),
(ex, _, count, _) =>
{
log.Warn($"Redis failover: attempt {retryCount} of {maxRetries}, " +
$"waiting {delay.TotalSeconds:F1}s. {ex.Message}");
if (count == maxRetries)
{
log.Warn($"Redis Retry threshold hit: {count}/{maxRetries}. {ex.Message}");
}
});
var connectionRetry = Policy
.Handle<RedisConnectionException>()
.Or<RedisServerException>(ex =>
ex.Message.Contains("READONLY", StringComparison.OrdinalIgnoreCase) ||
ex.Message.Contains("TRYAGAIN", StringComparison.OrdinalIgnoreCase))
.Or<RedisException>(ex =>
ex.Message.Contains("No connection is available", StringComparison.OrdinalIgnoreCase))
.WaitAndRetryAsync(
maxRetries,
attempt => TimeSpan.FromSeconds(Math.Min(Math.Pow(2, attempt), 10)) +
TimeSpan.FromMilliseconds(Random.Shared.Next(0, 500)),
(ex, _, count, _) =>
{
if (count == maxRetries)
{
log.Warn($"Redis Retry threshold hit: {count}/{maxRetries}. {ex.Message}");
}
}
);
IdempotentPolicy = Policy.WrapAsync(circuitBreaker, fullRetry);
NonIdempotentPolicy = Policy.WrapAsync(circuitBreaker, connectionRetry);
}
private IAsyncPolicy FailoverPolicy { get; }
private IAsyncPolicy IdempotentPolicy { get; }
private IAsyncPolicy NonIdempotentPolicy { get; }
public void Dispose()
{
@@ -59,7 +89,7 @@ namespace Jube.ResilientRedisConnection
public ResilientRedisDatabase GetDatabase(int db = -1)
{
return new ResilientRedisDatabase(multiplexer.GetDatabase(db), FailoverPolicy);
return new ResilientRedisDatabase(multiplexer.GetDatabase(db), IdempotentPolicy, NonIdempotentPolicy);
}
}
}
Jube.ResilientRedisConnection/ResilientRedisDatabase.cs
+139 -80
View File
@@ -16,163 +16,222 @@ namespace Jube.ResilientRedisConnection
using Polly;
using StackExchange.Redis;
public class ResilientRedisDatabase(IDatabase inner, IAsyncPolicy policy)
public class ResilientRedisDatabase(IDatabase inner, IAsyncPolicy idempotentPolicy, IAsyncPolicy nonIdempotentPolicy)
{
private readonly IAsyncPolicy policy = policy ?? throw new ArgumentNullException(nameof(policy));
private readonly IAsyncPolicy idempotentPolicy = idempotentPolicy ?? throw new ArgumentNullException(nameof(idempotentPolicy));
public IDatabase UnderlyingDatabase
private IDatabase UnderlyingDatabase
{
get;
} = inner ?? throw new ArgumentNullException(nameof(inner));
public Task<bool> StringSetAsync(RedisKey key, RedisValue value,
TimeSpan? expiry = null, When when = When.Always, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.StringSetAsync(key, value, expiry, when, flags),
new Context());
}
public Task<RedisValue> StringGetAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.StringGetAsync(key, flags),
new Context());
}
public Task<bool> StringSetAsync(KeyValuePair<RedisKey, RedisValue>[] values,
When when = When.Always, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.StringSetAsync(values, when, flags),
new Context());
}
public Task<RedisValue[]> StringGetAsync(RedisKey[] keys, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.StringGetAsync(keys, flags),
new Context());
}
public Task<bool> KeyDeleteAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.KeyDeleteAsync(key, flags),
new Context());
}
public Task<bool> KeyExistsAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.KeyExistsAsync(key, flags),
new Context());
}
public Task<bool> KeyExpireAsync(RedisKey key, TimeSpan? expiry, CommandFlags flags = CommandFlags.None)
private Task<bool> KeyRenameAsync(RedisKey key, RedisKey newKey, When when = When.Always,
CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.KeyExpireAsync(key, expiry, flags),
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.KeyRenameAsync(key, newKey, when, flags),
new Context());
}
public bool KeyRename(RedisKey key, RedisKey newKey, When when = When.Always,
CommandFlags flags = CommandFlags.None)
{
return KeyRenameAsync(key, newKey, when, flags).GetAwaiter().GetResult();
}
public Task<bool> HashSetAsync(RedisKey key, RedisValue field, RedisValue value,
When when = When.Always, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashSetAsync(key, field, value, when, flags),
new Context());
}
public Task HashSetAsync(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None)
private Task HashSetAsync(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashSetAsync(key, hashFields, flags),
new Context());
}
public void HashSet(RedisKey key, RedisValue field, RedisValue value,
When when = When.Always, CommandFlags flags = CommandFlags.None)
{
HashSetAsync(key, field, value, when, flags).GetAwaiter().GetResult();
}
public void HashSet(RedisKey key, HashEntry[] hashFields, CommandFlags flags = CommandFlags.None)
{
HashSetAsync(key, hashFields, flags).GetAwaiter().GetResult();
}
public Task<RedisValue> HashGetAsync(RedisKey key, RedisValue field, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashGetAsync(key, field, flags),
new Context());
}
public Task<HashEntry[]> HashGetAllAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
public Task<RedisValue[]> HashGetAsync(RedisKey key, RedisValue[] fields, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.HashGetAllAsync(key, flags),
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashGetAsync(key, fields, flags),
new Context());
}
public Task<bool> HashDeleteAsync(RedisKey key, RedisValue field, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashDeleteAsync(key, field, flags),
new Context());
}
public Task<long> ListRightPushAsync(RedisKey key, RedisValue value, When when = When.Always,
public Task<long> HashDeleteAsync(RedisKey key, RedisValue[] fields, CommandFlags flags = CommandFlags.None)
{
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashDeleteAsync(key, fields, flags),
new Context());
}
public bool HashDelete(RedisKey key, RedisValue field, CommandFlags flags = CommandFlags.None)
{
return HashDeleteAsync(key, field, flags).GetAwaiter().GetResult();
}
public Task<long> HashIncrementAsync(RedisKey key, RedisValue field, long value,
CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.ListRightPushAsync(key, value, when, flags),
return nonIdempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashIncrementAsync(key, field, value, flags),
new Context());
}
public Task<RedisValue> ListLeftPopAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
public Task<double> HashIncrementAsync(RedisKey key, RedisValue field, double value,
CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.ListLeftPopAsync(key, flags),
return nonIdempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashIncrementAsync(key, field, value, flags),
new Context());
}
public Task<long> ListLengthAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
public Task<long> HashDecrementAsync(RedisKey key, RedisValue field, long value,
CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.ListLengthAsync(key, flags),
return nonIdempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashDecrementAsync(key, field, value, flags),
new Context());
}
public Task<double> HashDecrementAsync(RedisKey key, RedisValue field, double value,
CommandFlags flags = CommandFlags.None)
{
return nonIdempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashDecrementAsync(key, field, value, flags),
new Context());
}
public Task<long> HashStringLengthAsync(RedisKey key, RedisValue field, CommandFlags flags = CommandFlags.None)
{
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.HashStringLengthAsync(key, field, flags),
new Context());
}
public IAsyncEnumerable<HashEntry> HashScanAsync(RedisKey key, RedisValue pattern = default,
int pageSize = 250, long cursor = 0, int pageOffset = 0, CommandFlags flags = CommandFlags.None)
{
return UnderlyingDatabase.HashScanAsync(key, pattern, pageSize, cursor, pageOffset, flags);
}
public IEnumerable<HashEntry> HashScan(RedisKey key, RedisValue pattern = default,
int pageSize = 250, long cursor = 0, int pageOffset = 0, CommandFlags flags = CommandFlags.None)
{
return UnderlyingDatabase.HashScan(key, pattern, pageSize, cursor, pageOffset, flags);
}
public Task<bool> SetAddAsync(RedisKey key, RedisValue value, CommandFlags flags = CommandFlags.None)
{
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.SetAddAsync(key, value, flags),
new Context());
}
public Task<bool> SetRemoveAsync(RedisKey key, RedisValue value, CommandFlags flags = CommandFlags.None)
{
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.SetRemoveAsync(key, value, flags),
new Context());
}
public Task<RedisValue[]> SetMembersAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
{
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.SetMembersAsync(key, flags),
new Context());
}
public Task<bool> SortedSetAddAsync(RedisKey key, RedisValue member, double score,
CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.SortedSetAddAsync(key, member, score, flags),
new Context());
}
public Task<bool> SortedSetUpdateAsync(RedisKey key, RedisValue member, double score,
SortedSetWhen when = SortedSetWhen.Always, CommandFlags flags = CommandFlags.None)
{
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.SortedSetUpdateAsync(key, member, score, when, flags),
new Context());
}
public Task<bool> SortedSetRemoveAsync(RedisKey key, RedisValue member, CommandFlags flags = CommandFlags.None)
{
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.SortedSetRemoveAsync(key, member, flags),
new Context());
}
public Task<long> SortedSetRemoveAsync(RedisKey key, RedisValue[] members, CommandFlags flags = CommandFlags.None)
{
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.SortedSetRemoveAsync(key, members, flags),
new Context());
}
public Task<SortedSetEntry[]> SortedSetRangeByRankWithScoresAsync(RedisKey key,
long start = 0, long stop = -1, Order order = Order.Ascending,
CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.SortedSetRangeByRankWithScoresAsync(key, start, stop, order, flags),
new Context());
}
public ITransaction CreateTransaction(object? asyncState = null)
public Task<SortedSetEntry[]> SortedSetRangeByScoreWithScoresAsync(RedisKey key,
double start = Double.NegativeInfinity, double stop = Double.PositiveInfinity,
Exclude exclude = Exclude.None, Order order = Order.Ascending,
long skip = 0, long take = -1, CommandFlags flags = CommandFlags.None)
{
return UnderlyingDatabase.CreateTransaction(asyncState);
}
public Task<RedisResult> ScriptEvaluateAsync(string script, RedisKey[]? keys = null,
RedisValue[]? values = null, CommandFlags flags = CommandFlags.None)
{
return policy.ExecuteAsync(_ =>
UnderlyingDatabase.ScriptEvaluateAsync(script, keys, values, flags),
return idempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.SortedSetRangeByScoreWithScoresAsync(key, start, stop, exclude, order, skip, take, flags),
new Context());
}
public bool StringSet(RedisKey key, RedisValue value, TimeSpan? expiry = null,
When when = When.Always, CommandFlags flags = CommandFlags.None)
public Task<long> PublishAsync(RedisChannel channel, RedisValue message,
CommandFlags flags = CommandFlags.None)
{
return StringSetAsync(key, value, expiry, when, flags).GetAwaiter().GetResult();
}
public RedisValue StringGet(RedisKey key, CommandFlags flags = CommandFlags.None)
{
return StringGetAsync(key, flags).GetAwaiter().GetResult();
return nonIdempotentPolicy.ExecuteAsync(_ =>
UnderlyingDatabase.PublishAsync(channel, message, flags),
new Context());
}
}
}