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

Information
Class: AsyncResponse.Transports.Redis.RedisSubscriberService
Assembly: AsyncResponse.Transports.Redis
File(s): /_/src/Transports/AsyncResponse.Transports.Redis/RedisSubscriberServices.cs
Line coverage
92%
Covered lines: 206
Uncovered lines: 17
Coverable lines: 223
Total lines: 549
Line coverage: 92.3%
Branch coverage
92%
Covered branches: 39
Total branches: 42
Branch coverage: 92.8%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
IsWithinInboundBudget(...)100%210%
.cctor()100%11100%
.ctor(...)100%11100%
.ctor(...)100%11100%
get_Options()100%11100%
get_Logger()100%11100%
StartAsync(...)100%11100%
ExecuteAsync(...)100%11100%
RunSubscriberAsync()100%1212100%
DispatchEntryAsync()100%11100%
EnsureConsumerGroupAsync()100%11100%
ClaimPendingAsync()91.66%131284%
DispatchBatchAsync()100%66100%
RenewClaimLoopAsync()75%4471.42%
get_SettledCount()100%11100%
MarkSettled()100%11100%
CreateDelivery(...)83.33%66100%
ToPositiveMilliseconds(...)100%11100%
ResolveConsumerName(...)100%22100%
CreateGeneratedConsumerName()100%11100%
ComposeGeneratedConsumerName(...)100%11100%

File(s)

/_/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{
 10    /// <summary>
 11    /// Whether the payload is within the engine's inbound size budget. Overridden by the response
 12    /// ingress subscriber, which is the only role that parses the BODY to find a correlation id —
 13    /// the worker role reads a header and never touches payload size. Default true so a role
 14    /// without a budget behaves exactly as before.
 15    /// </summary>
 016    protected virtual bool IsWithinInboundBudget(string payload) => true;
 17
 518    private static readonly string GeneratedConsumerName = CreateGeneratedConsumerName();
 19
 20    private readonly IRedisStreamDatabase _database;
 21
 22    /// <summary>Runs the RedisSubscriberService operation.</summary>
 23    protected RedisSubscriberService(
 24        IOptions<RedisAsyncResponseTransportOptions> options,
 25        IConnectionMultiplexer multiplexer,
 26        ILogger logger)
 39027        : this(
 39028            options,
 39029            new RedisStreamDatabaseAdapter(multiplexer.GetDatabase(), options.Value.OperationTimeout),
 39030            logger)
 31    {
 39032    }
 33
 34    /// <summary>Runs the RedisSubscriberService operation.</summary>
 42435    protected RedisSubscriberService(
 42436        IOptions<RedisAsyncResponseTransportOptions> options,
 42437        IRedisStreamDatabase database,
 42438        ILogger logger)
 39    {
 42440        Options = options.Value;
 42441        RedisTransportOptionsValidator.ValidateCommon(Options);
 42442        _database = database;
 42443        Logger = logger;
 42444    }
 45
 2031346    protected RedisAsyncResponseTransportOptions Options { get; }
 83647    protected ILogger Logger { get; }
 48
 49    protected abstract RedisKey Stream { get; }
 50    protected abstract RedisValue ConsumerGroup { get; }
 51    protected abstract RedisSubscriberOptions SubscriberOptions { get; }
 52    protected abstract RedisSubscriberRole SubscriberRole { get; }
 53    /// <summary>Handles the delivered message.</summary>
 54    protected abstract Task HandleMessageAsync(RedisStreamDelivery delivery, CancellationToken cancellationToken);
 55
 56    /// <summary>Runs this background operation until cancellation is requested.</summary>
 57    /// <summary>
 58    /// Validates subscriber options here rather than at the top of <c>ExecuteAsync</c>: since
 59    /// Microsoft.Extensions.Hosting.Abstractions 10.0.10, <c>BackgroundService.StartAsync</c> no
 60    /// longer runs <c>ExecuteAsync</c> inline, so a throw there surfaces only through the host's
 61    /// background-exception handling — or never, when a fast stop discards the queued work —
 62    /// instead of failing host startup synchronously.
 63    /// </summary>
 64    public override Task StartAsync(CancellationToken cancellationToken)
 65    {
 41466        RedisMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 41267        return base.StartAsync(cancellationToken);
 68    }
 69
 70    protected override Task ExecuteAsync(CancellationToken stoppingToken)
 41471        => SubscriberSupervisor.RunAsync(
 41472            RunSubscriberAsync,
 41473            stoppingToken,
 274            failures => AsyncResponseRetry.Backoff(
 275                failures,
 276                Options.SubscriberRetryBaseDelay,
 277                Options.SubscriberRetryMaxDelay),
 41678            (ex, retryDelay) => Logger.LogWarning(
 41679                ex,
 41680                "Redis subscriber failed for stream {Stream} ({Role}); retrying in {RetryDelay}.",
 41681                Stream.ToString(),
 41682                SubscriberRole,
 41683                retryDelay));
 84
 85    private async Task RunSubscriberAsync(CancellationToken stoppingToken)
 86    {
 41687        if (Options.CreateConsumerGroups)
 41488            await EnsureConsumerGroupAsync(stoppingToken).ConfigureAwait(false);
 89
 41690        var consumerName = ResolveConsumerName(Options, SubscriberRole);
 41691        await using var dispatcher = RedisMessageDispatcher.Create(
 41692            HandleMessageAsync,
 41693            _database,
 41694            Options,
 41695            SubscriberOptions,
 41696            Logger,
 41697            Stream,
 41698            ConsumerGroup,
 41699            SubscriberRole);
 100
 416101        Logger.LogInformation(
 416102            "Redis subscriber started. Stream: {Stream}. Group: {ConsumerGroup}. Consumer: {ConsumerName}. Role: {Role}.
 416103            Stream.ToString(),
 416104            ConsumerGroup.ToString(),
 416105            consumerName.ToString(),
 416106            SubscriberRole,
 416107            SubscriberOptions.AckMode);
 108
 416109        var nextPendingClaimAt = DateTimeOffset.UtcNow;
 4247110        while (!stoppingToken.IsCancellationRequested)
 111        {
 4228112            var processed = 0;
 113
 114            // When the dispatcher is saturated (ACK-after-enqueue queue full) stop pulling new entries:
 115            // reading them would only move the backlog into the pending-entry list and spin the loop.
 116            // The unread entries stay as new messages in the stream until capacity frees.
 4228117            if (dispatcher.CanAcceptMore)
 118            {
 4225119                var utcNow = DateTimeOffset.UtcNow;
 4225120                if (utcNow >= nextPendingClaimAt)
 121                {
 426122                    processed += await ClaimPendingAsync(dispatcher, consumerName, stoppingToken).ConfigureAwait(false);
 426123                    nextPendingClaimAt = utcNow + SubscriberOptions.PendingClaimInterval;
 124                }
 125
 126                // Clamp the read to the dispatcher's FREE slots (ASB/SQS parity), not merely to
 127                // "is there one": a full BatchSize read into a nearly-full early-ACK queue deferred
 128                // the surplus into the PEL un-ACKed, every reclaim bumped its delivery count, and
 129                // the pre-execution cap eventually dead-lettered healthy jobs whose handler never
 130                // ran. The claim above may have taken the last slot.
 4225131                var readCount = Math.Min(SubscriberOptions.BatchSize, dispatcher.FreeCapacity);
 4225132                if (readCount > 0)
 133                {
 4225134                    var entries = await _database.StreamReadGroupAsync(
 4225135                        Stream,
 4225136                        ConsumerGroup,
 4225137                        consumerName,
 4225138                        readCount,
 4225139                        stoppingToken).ConfigureAwait(false);
 140
 4207141                    processed += await DispatchBatchAsync(
 4207142                        dispatcher,
 4207143                        entries,
 4207144                        consumerName,
 441145                        static _ => 1,
 4207146                        stoppingToken).ConfigureAwait(false);
 147                }
 148            }
 149
 150            // Throttle when nothing advanced — an empty stream, or every entry deferred under backpressure.
 4210151            if (processed == 0)
 3784152                await Task.Delay(SubscriberOptions.EmptyPollDelay, stoppingToken).ConfigureAwait(false);
 153        }
 19154    }
 155
 156    private async Task<RedisDispatchOutcome> DispatchEntryAsync(
 157        RedisMessageDispatcher dispatcher,
 158        StreamEntry entry,
 159        int attempt,
 160        CancellationToken cancellationToken)
 161    {
 162        RedisStreamDelivery delivery;
 163        try
 164        {
 452165            delivery = CreateDelivery(entry, attempt);
 448166        }
 4167        catch (Exception ex) when (ex is not OperationCanceledException)
 168        {
 169            // A foreign/malformed entry (or a trimmed tombstone) can never be handled; dead-letter and
 170            // ACK it so it drains instead of poisoning the pending-claim loop forever.
 171            //
 172            // Deliberately every non-cancellation exception, not just InvalidDataException:
 173            // CreateDelivery also runs correlation-id extraction, and anything that escapes here
 174            // leaves the entry in the PEL to be reclaimed and re-thrown every PendingClaimInterval,
 175            // abandoning the rest of the claimed batch behind it — MaxDeliveryAttempts cannot help,
 176            // because it is keyed on a delivery this path never constructed.
 4177            await dispatcher.DiscardUnprocessableAsync(Stream, ConsumerGroup, entry, ex, cancellationToken).ConfigureAwa
 4178            return RedisDispatchOutcome.Processed;
 179        }
 180
 448181        return await dispatcher.HandleAsync(delivery, cancellationToken).ConfigureAwait(false);
 452182    }
 183
 184    private async Task EnsureConsumerGroupAsync(CancellationToken cancellationToken)
 185    {
 186        try
 187        {
 414188            await _database.StreamCreateConsumerGroupAsync(
 414189                Stream,
 414190                ConsumerGroup,
 414191                StreamPosition.Beginning,
 414192                createStream: true,
 414193                cancellationToken).ConfigureAwait(false);
 412194        }
 2195        catch (RedisServerException ex) when (ex.Message.Contains("BUSYGROUP", StringComparison.OrdinalIgnoreCase))
 196        {
 197            // The group already exists; this is the expected path after the first app instance.
 2198        }
 414199    }
 200
 201    private async Task<int> ClaimPendingAsync(
 202        RedisMessageDispatcher dispatcher,
 203        RedisValue consumerName,
 204        CancellationToken cancellationToken)
 205    {
 206        // Same clamp as the read: reclaiming more than the dispatcher can take defers the rest
 207        // straight back into the PEL with a bumped delivery count.
 426208        var claimCount = Math.Min(SubscriberOptions.PendingClaimBatchSize, dispatcher.FreeCapacity);
 426209        if (claimCount <= 0)
 0210            return 0;
 211
 426212        var minIdleMs = ToPositiveMilliseconds(SubscriberOptions.PendingMessageMinIdleTime);
 426213        var pending = await _database.StreamPendingMessagesAsync(
 426214            Stream,
 426215            ConsumerGroup,
 426216            claimCount,
 426217            RedisValue.Null,
 426218            minId: null,
 426219            maxId: null,
 426220            minIdleMs,
 426221            cancellationToken).ConfigureAwait(false);
 222
 426223        if (pending.Length == 0)
 413224            return 0;
 225
 13226        var pendingById = pending.ToDictionary(
 17227            item => item.MessageId.ToString(),
 13228            StringComparer.Ordinal);
 13229        var claimed = await _database.StreamClaimAsync(
 13230            Stream,
 13231            ConsumerGroup,
 13232            consumerName,
 13233            minIdleMs,
 17234            pending.Select(item => item.MessageId).ToArray(),
 13235            cancellationToken).ConfigureAwait(false);
 236
 237        // Redis 5/6 answer XCLAIM with a nil entry for an id whose message was trimmed while still
 238        // pending (7.x drops it from the PEL instead). A nil entry has no id, so neither the
 239        // dispatch path nor the JUSTID heartbeat can name it, and it stayed in the PEL to be
 240        // re-claimed every cycle. When the reply is complete its order matches the request, so
 241        // the tombstones are ACKed by their pending ids here; a partial reply leaves them for the
 242        // next cycle. Either way only real entries are dispatched.
 26243        if (Array.Exists(claimed, static entry => entry.Id.IsNull))
 244        {
 4245            if (claimed.Length == pending.Length)
 246            {
 12247                for (var index = 0; index < claimed.Length; index++)
 248                {
 4249                    if (!claimed[index].Id.IsNull)
 250                        continue;
 251
 2252                    var tombstoneId = pending[index].MessageId;
 253                    try
 254                    {
 2255                        await _database.StreamAcknowledgeAsync(Stream, ConsumerGroup, tombstoneId, CancellationToken.Non
 2256                        Logger.LogWarning(
 2257                            "Redis pending entry {MessageId} on {Stream} was trimmed while still pending; ACKed the tomb
 2258                            tombstoneId.ToString(),
 2259                            Stream.ToString());
 2260                    }
 0261                    catch (Exception ex)
 262                    {
 0263                        Logger.LogWarning(
 0264                            ex,
 0265                            "Failed to ACK trimmed pending entry {MessageId} on {Stream}; it is retried on the next pend
 0266                            tombstoneId.ToString(),
 0267                            Stream.ToString());
 0268                    }
 2269                }
 270            }
 271
 10272            claimed = Array.FindAll(claimed, static entry => !entry.Id.IsNull);
 273        }
 274
 13275        return await DispatchBatchAsync(
 13276            dispatcher,
 13277            claimed,
 13278            consumerName,
 13279            entry =>
 13280            {
 11281                var priorDeliveries = pendingById.TryGetValue(entry.Id.ToString(), out var info)
 11282                    ? info.DeliveryCount
 11283                    : 1;
 11284                return Math.Max(1, priorDeliveries + 1);
 13285            },
 13286            cancellationToken).ConfigureAwait(false);
 426287    }
 288
 289    private async Task<int> DispatchBatchAsync(
 290        RedisMessageDispatcher dispatcher,
 291        StreamEntry[] entries,
 292        RedisValue consumerName,
 293        Func<StreamEntry, int> attemptFor,
 294        CancellationToken stoppingToken)
 295    {
 4220296        if (entries.Length == 0)
 3794297            return 0;
 298
 299        // The batch is dispatched serially, and XREADGROUP/XCLAIM stamped every entry's idle
 300        // clock at read time — so a slow handler lets the idle time of the later (still
 301        // unprocessed) entries cross PendingMessageMinIdleTime, where a sibling's pending-claim
 302        // scan steals and re-runs them concurrently, bumping their PEL delivery count toward the
 303        // dead-letter cap on work that never once failed. While the batch is in flight, a
 304        // heartbeat claims the unprocessed entries back to this consumer with XCLAIM JUSTID,
 305        // which resets idle WITHOUT bumping the delivery count.
 426306        var progress = new BatchProgress();
 426307        using var renewalCancellation = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
 426308        var renewalTask = RenewClaimLoopAsync(entries, consumerName, progress, renewalCancellation.Token);
 426309        var processed = 0;
 310        try
 311        {
 1756312            foreach (var entry in entries)
 313            {
 314                try
 315                {
 452316                    if (await DispatchEntryAsync(dispatcher, entry, attemptFor(entry), stoppingToken).ConfigureAwait(fal
 452317                        == RedisDispatchOutcome.Processed)
 318                    {
 452319                        processed++;
 320                    }
 452321                }
 322                finally
 323                {
 324                    // Also counts Deferred entries: they were left pending ON PURPOSE, so the
 325                    // heartbeat must stop touching them and let their idle accrue toward reclaim.
 452326                    progress.MarkSettled();
 327                }
 328            }
 329        }
 330        finally
 331        {
 426332            renewalCancellation.Cancel();
 426333            await renewalTask.ConfigureAwait(false);
 334        }
 335
 426336        return processed;
 4220337    }
 338
 339    private async Task RenewClaimLoopAsync(
 340        StreamEntry[] entries,
 341        RedisValue consumerName,
 342        BatchProgress progress,
 343        CancellationToken cancellationToken)
 344    {
 345        // ~PendingMessageMinIdleTime/3: two chances to land an idle reset inside every reclaim
 346        // window even when one sweep is delayed by a slow round trip.
 426347        var interval = TimeSpan.FromMilliseconds(Math.Max(1, SubscriberOptions.PendingMessageMinIdleTime.TotalMillisecon
 348        try
 349        {
 6350            while (true)
 351            {
 432352                await Task.Delay(interval, cancellationToken).ConfigureAwait(false);
 353
 354                // Claim from the first unsettled entry onward: that covers the entry currently in
 355                // the handler plus everything still waiting its turn. minIdle 0 resets the idle
 356                // clock unconditionally, and the races are harmless — an entry ACKed while this
 357                // sweep is in flight has left the PEL (the claim simply skips it), and a failed
 358                // entry is settled before MarkSettled runs, so its post-failure idle countdown is
 359                // never stretched.
 6360                var settled = progress.SettledCount;
 6361                if (settled >= entries.Length)
 0362                    return;
 363
 6364                var remaining = new RedisValue[entries.Length - settled];
 28365                for (var i = settled; i < entries.Length; i++)
 8366                    remaining[i - settled] = entries[i].Id;
 367
 368                try
 369                {
 6370                    await _database.StreamClaimIdsOnlyAsync(
 6371                        Stream,
 6372                        ConsumerGroup,
 6373                        consumerName,
 6374                        minIdleTimeInMilliseconds: 0,
 6375                        remaining,
 6376                        cancellationToken).ConfigureAwait(false);
 6377                }
 0378                catch (Exception ex) when (ex is not OperationCanceledException)
 379                {
 0380                    Logger.LogWarning(
 0381                        ex,
 0382                        "Failed to refresh the pending idle time of {EntryCount} Redis entries on {Stream}; a sibling ma
 0383                        remaining.Length,
 0384                        Stream.ToString());
 0385                }
 6386            }
 387        }
 426388        catch (OperationCanceledException)
 389        {
 390            // The batch finished or the subscriber is stopping.
 426391        }
 426392    }
 393
 394    private sealed class BatchProgress
 395    {
 396        // Settled only ever increments, so a monotonic volatile read is enough — no lock, and a
 397        // stale read only claims an already-settled entry once more (an ACKed id is simply
 398        // absent from the PEL, and a failed one gets its reclaim delayed by one sweep).
 399        private int _settledCount;
 400
 6401        public int SettledCount => Volatile.Read(ref _settledCount);
 402
 452403        public void MarkSettled() => Interlocked.Increment(ref _settledCount);
 404    }
 405
 406    private RedisStreamDelivery CreateDelivery(StreamEntry entry, int attempt)
 407    {
 452408        var payload = RedisCorrelationIdExtractor.TryReadField(entry, Options.PayloadField);
 452409        if (string.IsNullOrWhiteSpace(payload))
 410        {
 4411            throw new InvalidDataException(
 4412                $"Redis stream entry {entry.Id} on {Stream.ToString()} does not contain payload field '{Options.PayloadF
 413        }
 414
 415        // Body-path extraction parses the whole payload, so it is gated on the inbound budget;
 416        // the field/header path reads metadata only and is unaffected by payload size.
 448417        var correlationId = SubscriberRole is RedisSubscriberRole.ResponseIngress
 448418            ? IsWithinInboundBudget(payload)
 448419                ? RedisCorrelationIdExtractor.Extract(entry, payload, Options)
 448420                : null
 448421            : RedisCorrelationIdExtractor.TryReadField(entry, Options.CorrelationIdField);
 422
 448423        return new RedisStreamDelivery(
 448424            Stream,
 448425            ConsumerGroup,
 448426            entry.Id,
 448427            payload,
 448428            correlationId,
 448429            attempt,
 448430            entry);
 431    }
 432
 433    private static long ToPositiveMilliseconds(TimeSpan value)
 426434        => Math.Max(1, (long)Math.Ceiling(value.TotalMilliseconds));
 435
 436    private static RedisValue ResolveConsumerName(
 437        RedisAsyncResponseTransportOptions options,
 438        RedisSubscriberRole role)
 439    {
 440        // Append the role even to an explicitly configured name so the worker and response subscribers
 441        // never share a consumer identity (and therefore a pending-entry list) within their groups.
 416442        var baseName = !string.IsNullOrWhiteSpace(options.ConsumerName)
 416443            ? options.ConsumerName
 416444            : GeneratedConsumerName;
 445
 416446        return $"{baseName}-{role.ToString().ToLowerInvariant()}";
 447    }
 448
 449    private static string CreateGeneratedConsumerName()
 5450        => ComposeGeneratedConsumerName(Environment.MachineName, Environment.ProcessId, Guid.NewGuid());
 451
 452    /// <summary>Length budget of the generated consumer name, before the role suffix.</summary>
 453    internal const int MaxGeneratedConsumerNameLength = 64;
 454
 455    /// <summary>
 456    /// <c>{machine}-{pid}-{guid}</c> within <see cref="MaxGeneratedConsumerNameLength"/>, with the
 457    /// MACHINE NAME giving up the characters. The process id and the GUID are the only parts that
 458    /// make the name unique, and they sit at the end: the old head-keeping cut
 459    /// (<c>name[..64]</c>) removed them first, so a 63-character host name — a Kubernetes pod
 460    /// name, or any host at Linux's HOST_NAME_MAX — left every process on that host with the SAME
 461    /// consumer identity. Consumers that share a name share one pending-entry list inside the
 462    /// group: each could claim and acknowledge entries the other was still handling. The suffix is
 463    /// at most 44 characters, so the machine name always keeps at least 20.
 464    /// </summary>
 465    internal static string ComposeGeneratedConsumerName(string machineName, int processId, Guid instance)
 466    {
 27467        var suffix = $"-{processId.ToString(System.Globalization.CultureInfo.InvariantCulture)}-{instance:N}";
 27468        return PortableText.TruncateWellFormed(machineName, MaxGeneratedConsumerNameLength - suffix.Length) + suffix;
 469    }
 470}
 471
 472internal sealed class RedisWorkerSubscriber : RedisSubscriberService
 473{
 474    private readonly IAsyncResponseIngress _ingress;
 475    private readonly RedisTransportKeySchema _keys;
 476
 477    /// <summary>Runs the RedisWorkerSubscriber operation.</summary>
 478    public RedisWorkerSubscriber(
 479        IOptions<RedisAsyncResponseTransportOptions> options,
 480        IConnectionMultiplexer multiplexer,
 481        IAsyncResponseIngress ingress,
 482        ILogger<RedisWorkerSubscriber> logger)
 483        : base(options, multiplexer, logger)
 484    {
 485        _ingress = ingress;
 486        _keys = new RedisTransportKeySchema(options.Value);
 487    }
 488
 489    internal RedisWorkerSubscriber(
 490        IOptions<RedisAsyncResponseTransportOptions> options,
 491        IRedisStreamDatabase database,
 492        IAsyncResponseIngress ingress,
 493        ILogger<RedisWorkerSubscriber> logger)
 494        : base(options, database, logger)
 495    {
 496        _ingress = ingress;
 497        _keys = new RedisTransportKeySchema(options.Value);
 498    }
 499
 500    protected override RedisKey Stream => _keys.WorkerStream;
 501    protected override RedisValue ConsumerGroup => Options.WorkerConsumerGroup;
 502    protected override RedisSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 503    protected override RedisSubscriberRole SubscriberRole => RedisSubscriberRole.Worker;
 504
 505    /// <summary>Handles the delivered message.</summary>
 506    protected override Task HandleMessageAsync(RedisStreamDelivery delivery, CancellationToken cancellationToken)
 507        => _ingress.HandleWorkerMessageAsync(delivery.Payload);
 508}
 509
 510internal sealed class RedisResponseIngressSubscriber : RedisSubscriberService
 511{
 512    private readonly IAsyncResponseIngress _ingress;
 513    private readonly RedisTransportKeySchema _keys;
 514
 515    /// <summary>Runs the RedisResponseIngressSubscriber operation.</summary>
 516    public RedisResponseIngressSubscriber(
 517        IOptions<RedisAsyncResponseTransportOptions> options,
 518        IConnectionMultiplexer multiplexer,
 519        IAsyncResponseIngress ingress,
 520        ILogger<RedisResponseIngressSubscriber> logger)
 521        : base(options, multiplexer, logger)
 522    {
 523        _ingress = ingress;
 524        _keys = new RedisTransportKeySchema(options.Value);
 525    }
 526
 527    internal RedisResponseIngressSubscriber(
 528        IOptions<RedisAsyncResponseTransportOptions> options,
 529        IRedisStreamDatabase database,
 530        IAsyncResponseIngress ingress,
 531        ILogger<RedisResponseIngressSubscriber> logger)
 532        : base(options, database, logger)
 533    {
 534        _ingress = ingress;
 535        _keys = new RedisTransportKeySchema(options.Value);
 536    }
 537
 538    protected override RedisKey Stream => _keys.ResponseStream;
 539    protected override RedisValue ConsumerGroup => Options.ResponseConsumerGroup;
 540    protected override RedisSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 541    protected override RedisSubscriberRole SubscriberRole => RedisSubscriberRole.ResponseIngress;
 542
 543    /// <inheritdoc />
 544    protected override bool IsWithinInboundBudget(string payload) => !_ingress.IsOverInboundBudget(payload);
 545
 546    /// <summary>Handles the delivered message.</summary>
 547    protected override Task HandleMessageAsync(RedisStreamDelivery delivery, CancellationToken cancellationToken)
 548        => _ingress.HandleResponseMessageAsync(delivery.Payload, delivery.CorrelationId);
 549}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

IsWithinInboundBudget(System.String)
.cctor()
.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions>,StackExchange.Redis.IConnectionMultiplexer,Microsoft.Extensions.Logging.ILogger)
.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions>,AsyncResponse.Transports.Redis.IRedisStreamDatabase,Microsoft.Extensions.Logging.ILogger)
get_Options()
get_Logger()
StartAsync(System.Threading.CancellationToken)
ExecuteAsync(System.Threading.CancellationToken)
RunSubscriberAsync()
DispatchEntryAsync()
EnsureConsumerGroupAsync()
ClaimPendingAsync()
DispatchBatchAsync()
RenewClaimLoopAsync()
get_SettledCount()
MarkSettled()
CreateDelivery(StackExchange.Redis.StreamEntry,System.Int32)
ToPositiveMilliseconds(System.TimeSpan)
ResolveConsumerName(AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions,AsyncResponse.Transports.Redis.RedisSubscriberRole)
CreateGeneratedConsumerName()
ComposeGeneratedConsumerName(System.String,System.Int32,System.Guid)