< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.Transports.Redis.RedisSubscriberService
Assembly: AsyncResponse.Transports.Redis
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.Redis/RedisSubscriberServices.cs
Line coverage
100%
Covered lines: 147
Uncovered lines: 0
Coverable lines: 147
Total lines: 355
Line coverage: 100%
Branch coverage
96%
Covered branches: 31
Total branches: 32
Branch coverage: 96.8%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.cctor()100%11100%
.ctor(...)100%11100%
.ctor(...)100%11100%
ExecuteAsync()50%22100%
RunSubscriberAsync()100%1212100%
DispatchEntryAsync()100%11100%
EnsureConsumerGroupAsync()100%11100%
ClaimPendingAsync()100%1010100%
CreateDelivery(...)100%44100%
ToPositiveMilliseconds(...)100%11100%
ResolveConsumerName(...)100%22100%
CreateGeneratedConsumerName()100%11100%
TrimConsumerName(...)100%22100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.Redis/RedisSubscriberServices.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using StackExchange.Redis;
 5
 6namespace AsyncResponse.Transports.Redis;
 7
 8internal abstract class RedisSubscriberService : BackgroundService
 9{
 310    private static readonly string GeneratedConsumerName = CreateGeneratedConsumerName();
 11
 12    private readonly IRedisStreamDatabase _database;
 13
 14    /// <summary>Runs the RedisSubscriberService operation.</summary>
 15    protected RedisSubscriberService(
 16        IOptions<RedisAsyncResponseTransportOptions> options,
 17        IConnectionMultiplexer multiplexer,
 18        ILogger logger)
 319        : this(
 320            options,
 321            new RedisStreamDatabaseAdapter(multiplexer.GetDatabase(), options.Value.OperationTimeout),
 322            logger)
 23    {
 324    }
 25
 26    /// <summary>Runs the RedisSubscriberService operation.</summary>
 327    protected RedisSubscriberService(
 328        IOptions<RedisAsyncResponseTransportOptions> options,
 329        IRedisStreamDatabase database,
 330        ILogger logger)
 31    {
 332        Options = options.Value;
 333        RedisTransportOptionsValidator.ValidateCommon(Options);
 334        _database = database;
 335        Logger = logger;
 336    }
 37
 38    protected RedisAsyncResponseTransportOptions Options { get; }
 39    protected ILogger Logger { get; }
 40
 41    protected abstract RedisKey Stream { get; }
 42    protected abstract RedisValue ConsumerGroup { get; }
 43    protected abstract RedisSubscriberOptions SubscriberOptions { get; }
 44    protected abstract RedisSubscriberRole SubscriberRole { get; }
 45    /// <summary>Handles the delivered message.</summary>
 46    protected abstract Task HandleMessageAsync(RedisStreamDelivery delivery, CancellationToken cancellationToken);
 47
 48    /// <summary>Runs this background operation until cancellation is requested.</summary>
 49    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 50    {
 351        RedisMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 52
 353        var failures = 0;
 354        while (!stoppingToken.IsCancellationRequested)
 55        {
 56            try
 57            {
 358                await RunSubscriberAsync(stoppingToken).ConfigureAwait(false);
 359                return;
 60            }
 361            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 62            {
 363                return;
 64            }
 265            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 66            {
 267                failures++;
 268                var retryDelay = AsyncResponseRetry.Backoff(
 269                    failures,
 270                    Options.SubscriberRetryBaseDelay,
 271                    Options.SubscriberRetryMaxDelay);
 272                Logger.LogWarning(
 273                    ex,
 274                    "Redis subscriber failed for stream {Stream} ({Role}); retrying in {RetryDelay}.",
 275                    Stream.ToString(),
 276                    SubscriberRole,
 277                    retryDelay);
 278                await Task.Delay(retryDelay, stoppingToken).ConfigureAwait(false);
 79            }
 80        }
 381    }
 82
 83    private async Task RunSubscriberAsync(CancellationToken stoppingToken)
 84    {
 385        if (Options.CreateConsumerGroups)
 386            await EnsureConsumerGroupAsync(stoppingToken).ConfigureAwait(false);
 87
 388        var consumerName = ResolveConsumerName(Options, SubscriberRole);
 389        await using var dispatcher = RedisMessageDispatcher.Create(
 390            HandleMessageAsync,
 391            _database,
 392            Options,
 393            SubscriberOptions,
 394            Logger,
 395            Stream,
 396            ConsumerGroup,
 397            SubscriberRole);
 98
 399        Logger.LogInformation(
 3100            "Redis subscriber started. Stream: {Stream}. Group: {ConsumerGroup}. Consumer: {ConsumerName}. Role: {Role}.
 3101            Stream.ToString(),
 3102            ConsumerGroup.ToString(),
 3103            consumerName.ToString(),
 3104            SubscriberRole,
 3105            SubscriberOptions.AckMode);
 106
 3107        var nextPendingClaimAt = DateTimeOffset.UtcNow;
 3108        while (!stoppingToken.IsCancellationRequested)
 109        {
 3110            var processed = 0;
 111
 112            // When the dispatcher is saturated (ACK-after-enqueue queue full) stop pulling new entries:
 113            // reading them would only move the backlog into the pending-entry list and spin the loop.
 114            // The unread entries stay as new messages in the stream until capacity frees.
 3115            if (dispatcher.CanAcceptMore)
 116            {
 3117                var utcNow = DateTimeOffset.UtcNow;
 3118                if (utcNow >= nextPendingClaimAt)
 119                {
 3120                    processed += await ClaimPendingAsync(dispatcher, consumerName, stoppingToken).ConfigureAwait(false);
 3121                    nextPendingClaimAt = utcNow + SubscriberOptions.PendingClaimInterval;
 122                }
 123
 3124                var entries = await _database.StreamReadGroupAsync(
 3125                    Stream,
 3126                    ConsumerGroup,
 3127                    consumerName,
 3128                    SubscriberOptions.BatchSize,
 3129                    stoppingToken).ConfigureAwait(false);
 130
 3131                foreach (var entry in entries)
 132                {
 3133                    if (await DispatchEntryAsync(dispatcher, entry, attempt: 1, stoppingToken).ConfigureAwait(false)
 3134                        == RedisDispatchOutcome.Processed)
 135                    {
 3136                        processed++;
 137                    }
 138                }
 139            }
 140
 141            // Throttle when nothing advanced — an empty stream, or every entry deferred under backpressure.
 3142            if (processed == 0)
 3143                await Task.Delay(SubscriberOptions.EmptyPollDelay, stoppingToken).ConfigureAwait(false);
 144        }
 3145    }
 146
 147    private async Task<RedisDispatchOutcome> DispatchEntryAsync(
 148        RedisMessageDispatcher dispatcher,
 149        StreamEntry entry,
 150        int attempt,
 151        CancellationToken cancellationToken)
 152    {
 153        RedisStreamDelivery delivery;
 154        try
 155        {
 3156            delivery = CreateDelivery(entry, attempt);
 3157        }
 158        catch (InvalidDataException ex)
 159        {
 160            // A foreign/malformed entry (or a trimmed tombstone) can never be handled; dead-letter and
 161            // ACK it so it drains instead of poisoning the pending-claim loop forever.
 2162            await dispatcher.DiscardUnprocessableAsync(Stream, ConsumerGroup, entry, ex, cancellationToken).ConfigureAwa
 3163            return RedisDispatchOutcome.Processed;
 164        }
 165
 3166        return await dispatcher.HandleAsync(delivery, cancellationToken).ConfigureAwait(false);
 3167    }
 168
 169    private async Task EnsureConsumerGroupAsync(CancellationToken cancellationToken)
 170    {
 171        try
 172        {
 3173            await _database.StreamCreateConsumerGroupAsync(
 3174                Stream,
 3175                ConsumerGroup,
 3176                StreamPosition.Beginning,
 3177                createStream: true,
 3178                cancellationToken).ConfigureAwait(false);
 3179        }
 2180        catch (RedisServerException ex) when (ex.Message.Contains("BUSYGROUP", StringComparison.OrdinalIgnoreCase))
 181        {
 182            // The group already exists; this is the expected path after the first app instance.
 3183        }
 3184    }
 185
 186    private async Task<int> ClaimPendingAsync(
 187        RedisMessageDispatcher dispatcher,
 188        RedisValue consumerName,
 189        CancellationToken cancellationToken)
 190    {
 3191        var minIdleMs = ToPositiveMilliseconds(SubscriberOptions.PendingMessageMinIdleTime);
 3192        var pending = await _database.StreamPendingMessagesAsync(
 3193            Stream,
 3194            ConsumerGroup,
 3195            SubscriberOptions.PendingClaimBatchSize,
 3196            RedisValue.Null,
 3197            minId: null,
 3198            maxId: null,
 3199            minIdleMs,
 3200            cancellationToken).ConfigureAwait(false);
 201
 3202        if (pending.Length == 0)
 3203            return 0;
 204
 2205        var pendingById = pending.ToDictionary(
 2206            item => item.MessageId.ToString(),
 2207            StringComparer.Ordinal);
 2208        var claimed = await _database.StreamClaimAsync(
 2209            Stream,
 2210            ConsumerGroup,
 2211            consumerName,
 2212            minIdleMs,
 3213            pending.Select(item => item.MessageId).ToArray(),
 2214            cancellationToken).ConfigureAwait(false);
 215
 2216        var processed = 0;
 2217        foreach (var entry in claimed)
 218        {
 2219            var priorDeliveries = pendingById.TryGetValue(entry.Id.ToString(), out var info)
 2220                ? info.DeliveryCount
 2221                : 1;
 2222            if (await DispatchEntryAsync(dispatcher, entry, Math.Max(1, priorDeliveries + 1), cancellationToken).Configu
 2223                == RedisDispatchOutcome.Processed)
 224            {
 2225                processed++;
 226            }
 227        }
 228
 2229        return processed;
 3230    }
 231
 232    private RedisStreamDelivery CreateDelivery(StreamEntry entry, int attempt)
 233    {
 3234        var payload = RedisCorrelationIdExtractor.TryReadField(entry, Options.PayloadField);
 3235        if (string.IsNullOrWhiteSpace(payload))
 236        {
 3237            throw new InvalidDataException(
 3238                $"Redis stream entry {entry.Id} on {Stream.ToString()} does not contain payload field '{Options.PayloadF
 239        }
 240
 3241        var correlationId = SubscriberRole is RedisSubscriberRole.ResponseIngress
 3242            ? RedisCorrelationIdExtractor.Extract(entry, payload, Options)
 3243            : RedisCorrelationIdExtractor.TryReadField(entry, Options.CorrelationIdField);
 244
 3245        return new RedisStreamDelivery(
 3246            Stream,
 3247            ConsumerGroup,
 3248            entry.Id,
 3249            payload,
 3250            correlationId,
 3251            attempt,
 3252            entry);
 253    }
 254
 255    private static long ToPositiveMilliseconds(TimeSpan value)
 3256        => Math.Max(1, (long)Math.Ceiling(value.TotalMilliseconds));
 257
 258    private static RedisValue ResolveConsumerName(
 259        RedisAsyncResponseTransportOptions options,
 260        RedisSubscriberRole role)
 261    {
 262        // Append the role even to an explicitly configured name so the worker and response subscribers
 263        // never share a consumer identity (and therefore a pending-entry list) within their groups.
 3264        var baseName = !string.IsNullOrWhiteSpace(options.ConsumerName)
 3265            ? options.ConsumerName
 3266            : GeneratedConsumerName;
 267
 3268        return $"{baseName}-{role.ToString().ToLowerInvariant()}";
 269    }
 270
 271    private static string CreateGeneratedConsumerName()
 272    {
 3273        var name = $"{Environment.MachineName}-{Environment.ProcessId}-{Guid.NewGuid():N}";
 3274        return TrimConsumerName(name);
 275    }
 276
 277    internal static string TrimConsumerName(string name)
 3278        => name.Length <= 64 ? name : name[..64];
 279}
 280
 281internal sealed class RedisWorkerSubscriber : RedisSubscriberService
 282{
 283    private readonly IAsyncResponseIngress _ingress;
 284    private readonly RedisTransportKeySchema _keys;
 285
 286    /// <summary>Runs the RedisWorkerSubscriber operation.</summary>
 287    public RedisWorkerSubscriber(
 288        IOptions<RedisAsyncResponseTransportOptions> options,
 289        IConnectionMultiplexer multiplexer,
 290        IAsyncResponseIngress ingress,
 291        ILogger<RedisWorkerSubscriber> logger)
 292        : base(options, multiplexer, logger)
 293    {
 294        _ingress = ingress;
 295        _keys = new RedisTransportKeySchema(options.Value);
 296    }
 297
 298    internal RedisWorkerSubscriber(
 299        IOptions<RedisAsyncResponseTransportOptions> options,
 300        IRedisStreamDatabase database,
 301        IAsyncResponseIngress ingress,
 302        ILogger<RedisWorkerSubscriber> logger)
 303        : base(options, database, logger)
 304    {
 305        _ingress = ingress;
 306        _keys = new RedisTransportKeySchema(options.Value);
 307    }
 308
 309    protected override RedisKey Stream => _keys.WorkerStream;
 310    protected override RedisValue ConsumerGroup => Options.WorkerConsumerGroup;
 311    protected override RedisSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 312    protected override RedisSubscriberRole SubscriberRole => RedisSubscriberRole.Worker;
 313
 314    /// <summary>Handles the delivered message.</summary>
 315    protected override Task HandleMessageAsync(RedisStreamDelivery delivery, CancellationToken cancellationToken)
 316        => _ingress.HandleWorkerMessageAsync(delivery.Payload);
 317}
 318
 319internal sealed class RedisResponseIngressSubscriber : RedisSubscriberService
 320{
 321    private readonly IAsyncResponseIngress _ingress;
 322    private readonly RedisTransportKeySchema _keys;
 323
 324    /// <summary>Runs the RedisResponseIngressSubscriber operation.</summary>
 325    public RedisResponseIngressSubscriber(
 326        IOptions<RedisAsyncResponseTransportOptions> options,
 327        IConnectionMultiplexer multiplexer,
 328        IAsyncResponseIngress ingress,
 329        ILogger<RedisResponseIngressSubscriber> logger)
 330        : base(options, multiplexer, logger)
 331    {
 332        _ingress = ingress;
 333        _keys = new RedisTransportKeySchema(options.Value);
 334    }
 335
 336    internal RedisResponseIngressSubscriber(
 337        IOptions<RedisAsyncResponseTransportOptions> options,
 338        IRedisStreamDatabase database,
 339        IAsyncResponseIngress ingress,
 340        ILogger<RedisResponseIngressSubscriber> logger)
 341        : base(options, database, logger)
 342    {
 343        _ingress = ingress;
 344        _keys = new RedisTransportKeySchema(options.Value);
 345    }
 346
 347    protected override RedisKey Stream => _keys.ResponseStream;
 348    protected override RedisValue ConsumerGroup => Options.ResponseConsumerGroup;
 349    protected override RedisSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 350    protected override RedisSubscriberRole SubscriberRole => RedisSubscriberRole.ResponseIngress;
 351
 352    /// <summary>Handles the delivered message.</summary>
 353    protected override Task HandleMessageAsync(RedisStreamDelivery delivery, CancellationToken cancellationToken)
 354        => _ingress.HandleResponseMessageAsync(delivery.Payload, delivery.CorrelationId);
 355}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.cctor()
.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions>, StackExchange.Redis.IConnectionMultiplexer, Microsoft.Extensions.Logging.ILogger)
.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions>, AsyncResponse.Transports.Redis.IRedisStreamDatabase, Microsoft.Extensions.Logging.ILogger)
ExecuteAsync()
RunSubscriberAsync()
DispatchEntryAsync()
EnsureConsumerGroupAsync()
ClaimPendingAsync()
CreateDelivery(StackExchange.Redis.StreamEntry, int)
ToPositiveMilliseconds(System.TimeSpan)
ResolveConsumerName(AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions, AsyncResponse.Transports.Redis.RedisSubscriberRole)
CreateGeneratedConsumerName()
TrimConsumerName(string)