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

Information
Class: AsyncResponse.Transports.GooglePubSub.QueuedGooglePubSubMessageDispatcher
Assembly: AsyncResponse.Transports.GooglePubSub
File(s): /_/src/Transports/AsyncResponse.Transports.GooglePubSub/GooglePubSubMessageDispatcher.cs
Line coverage
100%
Covered lines: 130
Uncovered lines: 0
Coverable lines: 130
Total lines: 463
Line coverage: 100%
Branch coverage
100%
Covered branches: 14
Total branches: 14
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_PendingCount()100%11100%
get_RunningCount()100%11100%
HandleAsync()100%88100%
DisposeAsync()100%22100%
RunWorkerAsync()100%44100%

File(s)

/_/src/Transports/AsyncResponse.Transports.GooglePubSub/GooglePubSubMessageDispatcher.cs

#LineLine coverage
 1using Google.Cloud.PubSub.V1;
 2using Microsoft.Extensions.Logging;
 3using System.Diagnostics;
 4using System.Threading.Channels;
 5
 6namespace AsyncResponse.Transports.GooglePubSub;
 7
 8internal enum GooglePubSubSubscriberRole
 9{
 10    Worker,
 11    ResponseIngress
 12}
 13
 14internal abstract class GooglePubSubMessageDispatcher : IAsyncDisposable
 15{
 16    private readonly Func<PubsubMessage, CancellationToken, Task> _handler;
 17    private readonly GooglePubSubAsyncResponseOptions _transportOptions;
 18    private readonly GooglePubSubSubscriberOptions _subscriberOptions;
 19    private readonly string _subscriptionId;
 20    private readonly GooglePubSubSubscriberRole _role;
 21
 22    /// <summary>Runs the GooglePubSubMessageDispatcher operation.</summary>
 23    protected GooglePubSubMessageDispatcher(
 24        Func<PubsubMessage, CancellationToken, Task> handler,
 25        GooglePubSubAsyncResponseOptions transportOptions,
 26        GooglePubSubSubscriberOptions subscriberOptions,
 27        ILogger logger,
 28        string subscriptionId,
 29        GooglePubSubSubscriberRole role)
 30    {
 31        _handler = handler;
 32        _transportOptions = transportOptions;
 33        _subscriberOptions = subscriberOptions;
 34        Logger = logger;
 35        _subscriptionId = subscriptionId;
 36        _role = role;
 37    }
 38
 39    protected ILogger Logger { get; }
 40
 41    /// <summary>Creates the configured dispatcher.</summary>
 42    public static GooglePubSubMessageDispatcher Create(
 43        Func<PubsubMessage, CancellationToken, Task> handler,
 44        GooglePubSubAsyncResponseOptions transportOptions,
 45        GooglePubSubSubscriberOptions subscriberOptions,
 46        ILogger logger,
 47        string subscriptionId,
 48        GooglePubSubSubscriberRole role)
 49    {
 50        ValidateOptions(transportOptions, subscriberOptions, role);
 51
 52        return subscriberOptions.AckMode == GooglePubSubAckMode.AckAfterHandlerCompletes
 53            ? new AwaitingGooglePubSubMessageDispatcher(
 54                handler,
 55                transportOptions,
 56                subscriberOptions,
 57                logger,
 58                subscriptionId,
 59                role)
 60            : new QueuedGooglePubSubMessageDispatcher(
 61                handler,
 62                transportOptions,
 63                subscriberOptions,
 64                logger,
 65                subscriptionId,
 66                role);
 67    }
 68
 69    /// <summary>Validates the supplied options.</summary>
 70    public static void ValidateOptions(
 71        GooglePubSubAsyncResponseOptions transportOptions,
 72        GooglePubSubSubscriberOptions subscriberOptions,
 73        GooglePubSubSubscriberRole role)
 74    {
 75        var optionPath = role is GooglePubSubSubscriberRole.Worker
 76            ? $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.WorkerSubscriber)}"
 77            : $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.ResponseSubscriber)}
 78
 79        GooglePubSubOptionsValidator.ValidateTimeouts(transportOptions);
 80        GooglePubSubOptionsValidator.ValidateStreamingPull(subscriberOptions, optionPath);
 81
 82        if (!string.IsNullOrWhiteSpace(transportOptions.WorkerSubscriptionId)
 83            && !string.IsNullOrWhiteSpace(transportOptions.ResponseSubscriptionId)
 84            && StringComparer.Ordinal.Equals(transportOptions.WorkerSubscriptionId, transportOptions.ResponseSubscriptio
 85        {
 86            throw new InvalidOperationException(
 87                $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.WorkerSubscription
 88                $"{nameof(GooglePubSubAsyncResponseOptions.ResponseSubscriptionId)} must be distinct so worker and respo
 89        }
 90
 91        if (!string.IsNullOrWhiteSpace(transportOptions.WorkerTopicId)
 92            && !string.IsNullOrWhiteSpace(transportOptions.ResponseTopicId)
 93            && StringComparer.Ordinal.Equals(transportOptions.WorkerTopicId, transportOptions.ResponseTopicId))
 94        {
 95            throw new InvalidOperationException(
 96                $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.WorkerTopicId)} an
 97                $"{nameof(GooglePubSubAsyncResponseOptions.ResponseTopicId)} must be distinct so worker jobs and respons
 98        }
 99
 100        switch (subscriberOptions.AckMode)
 101        {
 102            case GooglePubSubAckMode.AckAfterHandlerCompletes:
 103                return;
 104
 105            case GooglePubSubAckMode.AckAfterEnqueue:
 106                if (subscriberOptions.BackgroundWorkerCount <= 0)
 107                {
 108                    throw new InvalidOperationException(
 109                        $"{optionPath}.{nameof(GooglePubSubSubscriberOptions.BackgroundWorkerCount)} must be explicitly 
 110                        $"when {nameof(GooglePubSubSubscriberOptions.AckMode)} is {nameof(GooglePubSubAckMode.AckAfterEn
 111                }
 112
 113                if (subscriberOptions.BackgroundQueueCapacity <= 0)
 114                {
 115                    throw new InvalidOperationException(
 116                        $"{optionPath}.{nameof(GooglePubSubSubscriberOptions.BackgroundQueueCapacity)} must be explicitl
 117                        $"when {nameof(GooglePubSubSubscriberOptions.AckMode)} is {nameof(GooglePubSubAckMode.AckAfterEn
 118                }
 119
 120                AsyncResponseChannelOptions.EnsureTimerBacked(subscriberOptions.BackgroundDrainTimeout, optionPath, name
 121
 122                // Pub/Sub spends the background drain plus the bounded subscriber-client stop
 123                // (ShutdownTimeout) at shutdown; both must fit inside the host budget.
 124                ShutdownBudgetValidator.Validate(
 125                    "Pub/Sub",
 126                    $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.HostShutdownTi
 127                    transportOptions.HostShutdownTimeout,
 128                    ($"{optionPath}.{nameof(GooglePubSubSubscriberOptions.BackgroundDrainTimeout)}", subscriberOptions.B
 129                    ($"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.ShutdownTimeo
 130
 131                return;
 132
 133            default:
 134                throw new InvalidOperationException(
 135                    $"{optionPath}.{nameof(GooglePubSubSubscriberOptions.AckMode)} has unsupported value '{subscriberOpt
 136        }
 137    }
 138
 139    /// <summary>Handles the delivered message.</summary>
 140    public abstract Task<SubscriberClient.Reply> HandleAsync(
 141        PubsubMessage message,
 142        CancellationToken subscriberCancellationToken);
 143
 144    /// <summary>Releases resources held by this instance.</summary>
 145    public virtual ValueTask DisposeAsync() => ValueTask.CompletedTask;
 146
 147    /// <summary>Runs the ExecuteHandlerAsync operation.</summary>
 148    protected async Task ExecuteHandlerAsync(
 149        PubsubMessage message,
 150        CancellationToken cancellationToken,
 151        bool logFailures = true)
 152    {
 153        using var activity = AsyncResponseDiagnostics.StartActivity(
 154            "asyncresponse.pubsub.receive",
 155            ActivityKind.Consumer);
 156        activity?.SetTag("asyncresponse.transport", "google_pubsub");
 157        activity?.SetTag("asyncresponse.pubsub.role", _role.ToString());
 158        activity?.SetTag("asyncresponse.pubsub.ack_mode", _subscriberOptions.AckMode.ToString());
 159        activity?.SetTag("messaging.system", "gcp_pubsub");
 160        activity?.SetTag("messaging.destination.name", _subscriptionId);
 161        activity?.SetTag("messaging.message.id", message.MessageId);
 162
 163        if (message.Attributes.TryGetValue(_transportOptions.CorrelationIdAttribute, out var correlationId))
 164            AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId);
 165
 166        try
 167        {
 168            await _handler(message, cancellationToken).ConfigureAwait(false);
 169        }
 170        catch (Exception ex)
 171        {
 172            if (logFailures)
 173                Logger.LogError(ex, "Pub/Sub message handling failed for message {MessageId}.", message.MessageId);
 174            AsyncResponseDiagnostics.SetError(activity, ex);
 175            throw;
 176        }
 177    }
 178
 179    /// <summary>Runs the NotifyBackgroundFailureAsync operation.</summary>
 180    protected async ValueTask NotifyBackgroundFailureAsync(
 181        PubsubMessage message,
 182        Exception exception,
 183        string subscriptionId,
 184        GooglePubSubSubscriberRole role)
 185    {
 186        var callback = _subscriberOptions.OnBackgroundFailure;
 187        if (callback is null)
 188            return;
 189
 190        try
 191        {
 192            await callback(new GooglePubSubBackgroundFailureContext(
 193                subscriptionId,
 194                role.ToString(),
 195                message,
 196                exception)).ConfigureAwait(false);
 197        }
 198        catch (Exception callbackException)
 199        {
 200            Logger.LogError(
 201                callbackException,
 202                "Pub/Sub background failure callback failed for already-ACKed message {MessageId} on {SubscriptionId}.",
 203                message.MessageId,
 204                subscriptionId);
 205        }
 206    }
 207}
 208
 209internal sealed class AwaitingGooglePubSubMessageDispatcher(
 210    Func<PubsubMessage, CancellationToken, Task> handler,
 211    GooglePubSubAsyncResponseOptions transportOptions,
 212    GooglePubSubSubscriberOptions subscriberOptions,
 213    ILogger logger,
 214    string subscriptionId,
 215    GooglePubSubSubscriberRole role)
 216    : GooglePubSubMessageDispatcher(handler, transportOptions, subscriberOptions, logger, subscriptionId, role)
 217{
 218    /// <summary>Handles the delivered message.</summary>
 219    public override async Task<SubscriberClient.Reply> HandleAsync(
 220        PubsubMessage message,
 221        CancellationToken subscriberCancellationToken)
 222    {
 223        try
 224        {
 225            await ExecuteHandlerAsync(message, subscriberCancellationToken).ConfigureAwait(false);
 226            return SubscriberClient.Reply.Ack;
 227        }
 228        catch (OperationCanceledException) when (subscriberCancellationToken.IsCancellationRequested)
 229        {
 230            // Host shutdown, not a handler failure. Pub/Sub's handler contract offers no
 231            // "leave unsettled": the only redelivery primitive is Nack (an expired ack
 232            // deadline counts a delivery attempt exactly the same), so Nack is returned here
 233            // too — but through this explicit branch so shutdown cancellation is never
 234            // mistaken for (or later routed through) a failure policy.
 235            return SubscriberClient.Reply.Nack;
 236        }
 237        catch
 238        {
 239            return SubscriberClient.Reply.Nack;
 240        }
 241    }
 242}
 243
 244internal sealed class QueuedGooglePubSubMessageDispatcher : GooglePubSubMessageDispatcher
 245{
 246    private readonly Channel<PubsubMessage> _queue;
 247    private readonly Task[] _workers;
 30248    private readonly CancellationTokenSource _drainCancellation = new();
 249    private readonly TimeSpan _drainTimeout;
 250    private readonly string _subscriptionId;
 251    private readonly GooglePubSubSubscriberRole _role;
 252    private int _pendingCount;
 253    private int _runningCount;
 254    private int _disposeStarted;
 255
 256    /// <summary>Runs the QueuedGooglePubSubMessageDispatcher operation.</summary>
 257    public QueuedGooglePubSubMessageDispatcher(
 258        Func<PubsubMessage, CancellationToken, Task> handler,
 259        GooglePubSubAsyncResponseOptions transportOptions,
 260        GooglePubSubSubscriberOptions subscriberOptions,
 261        ILogger logger,
 262        string subscriptionId,
 263        GooglePubSubSubscriberRole role)
 30264        : base(handler, transportOptions, subscriberOptions, logger, subscriptionId, role)
 265    {
 30266        _drainTimeout = subscriberOptions.BackgroundDrainTimeout;
 30267        _subscriptionId = subscriptionId;
 30268        _role = role;
 30269        _queue = Channel.CreateBounded<PubsubMessage>(new BoundedChannelOptions(subscriberOptions.BackgroundQueueCapacit
 30270        {
 30271            AllowSynchronousContinuations = false,
 30272            // Wait powers the queue-full backpressure path in HandleAsync: WriteAsync parks the
 30273            // subscriber callback until a worker frees a slot instead of dropping or NACKing.
 30274            FullMode = BoundedChannelFullMode.Wait,
 30275            SingleReader = subscriberOptions.BackgroundWorkerCount == 1,
 30276            SingleWriter = false
 30277        });
 278
 30279        _workers = Enumerable.Range(0, subscriberOptions.BackgroundWorkerCount)
 72280            .Select(workerIndex => Task.Run(() => RunWorkerAsync(workerIndex)))
 30281            .ToArray();
 282
 30283        Logger.LogInformation(
 30284            "Created Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId} with {WorkerCount} worker(s), queue capac
 30285            _subscriptionId,
 30286            subscriberOptions.BackgroundWorkerCount,
 30287            subscriberOptions.BackgroundQueueCapacity,
 30288            _drainTimeout);
 30289    }
 290
 144291    internal int PendingCount => Volatile.Read(ref _pendingCount);
 142292    internal int RunningCount => Volatile.Read(ref _runningCount);
 293
 294    /// <summary>Handles the delivered message.</summary>
 295    public override async Task<SubscriberClient.Reply> HandleAsync(
 296        PubsubMessage message,
 297        CancellationToken subscriberCancellationToken)
 298    {
 299        try
 300        {
 46301            Interlocked.Increment(ref _pendingCount);
 46302            if (_queue.Writer.TryWrite(message))
 303            {
 40304                Logger.LogDebug(
 40305                    "Enqueued Pub/Sub message {MessageId} for background handling on {SubscriptionId}. Pending={PendingC
 40306                    message.MessageId,
 40307                    _subscriptionId,
 40308                    PendingCount,
 40309                    RunningCount);
 40310                return SubscriberClient.Reply.Ack;
 311            }
 312
 313            // Queue full: apply backpressure instead of NACKing. A NACK burns one delivery attempt of
 314            // a DeadLetterPolicy configured on the subscription, so a saturated worker pool would
 315            // dead-letter healthy, never-executed messages. The streaming pull is flow-control-bounded
 316            // to the queue capacity, so at most capacity callbacks wait here; the await completes as
 317            // soon as a background worker frees a slot.
 4318            Logger.LogDebug(
 4319                "Pub/Sub background queue is full for {SubscriptionId}; waiting for capacity before ACKing message {Mess
 4320                _subscriptionId,
 4321                message.MessageId,
 4322                PendingCount,
 4323                RunningCount);
 4324            await _queue.Writer.WriteAsync(message, subscriberCancellationToken).ConfigureAwait(false);
 2325            return SubscriberClient.Reply.Ack;
 326        }
 4327        catch (Exception ex)
 328        {
 329            // Cancellation (subscriber stopping) or a completed channel (dispatcher disposing):
 330            // NACK so Pub/Sub redelivers the message to the next subscriber instance.
 4331            Interlocked.Decrement(ref _pendingCount);
 4332            if (ex is OperationCanceledException or ChannelClosedException)
 333            {
 2334                Logger.LogDebug(
 2335                    "Pub/Sub message {MessageId} for {SubscriptionId} could not be enqueued during shutdown; returning N
 2336                    message.MessageId,
 2337                    _subscriptionId);
 338            }
 339            else
 340            {
 2341                Logger.LogError(ex, "Failed to enqueue Pub/Sub message {MessageId} for {SubscriptionId}; returning NACK.
 342            }
 343
 4344            return SubscriberClient.Reply.Nack;
 345        }
 46346    }
 347
 348    /// <summary>Releases resources held by this instance.</summary>
 349    public override async ValueTask DisposeAsync()
 350    {
 32351        if (Interlocked.Exchange(ref _disposeStarted, 1) != 0)
 2352            return;
 353
 30354        Logger.LogInformation(
 30355            "Draining Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId}. Pending={PendingCount}, Running={Runnin
 30356            _subscriptionId,
 30357            PendingCount,
 30358            RunningCount);
 30359        _queue.Writer.TryComplete();
 360
 361        try
 362        {
 30363            await Task.WhenAll(_workers).WaitAsync(_drainTimeout).ConfigureAwait(false);
 24364            _drainCancellation.Dispose();
 24365            Logger.LogInformation(
 24366                "Drained Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId}. Pending={PendingCount}, Running={Run
 24367                _subscriptionId,
 24368                PendingCount,
 24369                RunningCount);
 24370        }
 4371        catch (TimeoutException ex)
 372        {
 4373            _drainCancellation.Cancel();
 4374            Logger.LogWarning(
 4375                ex,
 4376                "Timed out while draining Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId}. Pending={PendingCou
 4377                _subscriptionId,
 4378                PendingCount,
 4379                RunningCount);
 380
 381            // The workers are still running and read _drainCancellation.Token each loop, so disposing
 382            // it now would throw ObjectDisposedException inside them. Dispose once they actually finish,
 383            // off the shutdown path, so the source is not leaked either.
 4384            _ = Task.WhenAll(_workers).ContinueWith(
 4385                _ => _drainCancellation.Dispose(),
 4386                CancellationToken.None,
 4387                TaskContinuationOptions.ExecuteSynchronously,
 4388                TaskScheduler.Default);
 4389        }
 2390        catch (Exception ex)
 391        {
 392            // A worker faulted outside its own handler guard (DB/NATS dispatcher parity). WhenAll
 393            // only completes once every worker has finished, so the source is safe to dispose here
 394            // — and the fault must not escape DisposeAsync and mask the real shutdown path.
 2395            Logger.LogDebug(ex, "Pub/Sub ACK-after-enqueue dispatcher drain for {SubscriptionId} ended with an error.", 
 2396            _drainCancellation.Dispose();
 2397        }
 32398    }
 399
 400    private async Task RunWorkerAsync(int workerIndex)
 401    {
 154402        await foreach (var message in _queue.Reader.ReadAllAsync().ConfigureAwait(false))
 403        {
 42404            Interlocked.Decrement(ref _pendingCount);
 42405            Interlocked.Increment(ref _runningCount);
 406
 407            // Once the drain budget has lapsed, STOP executing. The token below cannot stop the
 408            // real handler — it is the ingress, whose target takes no CancellationToken — so the
 409            // loop kept starting fresh work past the budget and every message still queued at
 410            // process exit vanished with no record (they were ACKed at enqueue, so Pub/Sub will not
 411            // redeliver them). Route them through OnBackgroundFailure instead of losing them.
 42412            if (_drainCancellation.IsCancellationRequested)
 413            {
 2414                Logger.LogWarning(
 2415                    "Pub/Sub background handler for already-ACKed message {MessageId} on {SubscriptionId} was not starte
 2416                    message.MessageId,
 2417                    _subscriptionId);
 418
 2419                await NotifyBackgroundFailureAsync(
 2420                    message,
 2421                    new OperationCanceledException(
 2422                        "The ACK-after-enqueue drain budget lapsed before this already-ACKed message was handled."),
 2423                    _subscriptionId,
 2424                    _role).ConfigureAwait(false);
 425
 2426                Interlocked.Decrement(ref _runningCount);
 2427                continue;
 428            }
 429
 430            try
 431            {
 40432                Logger.LogDebug(
 40433                    "Pub/Sub background worker {WorkerIndex} handling message {MessageId} for {SubscriptionId}. Pending=
 40434                    workerIndex,
 40435                    message.MessageId,
 40436                    _subscriptionId,
 40437                    PendingCount,
 40438                    RunningCount);
 40439                await ExecuteHandlerAsync(
 40440                    message,
 40441                    _drainCancellation.Token,
 40442                    logFailures: false).ConfigureAwait(false);
 30443            }
 10444            catch (Exception ex)
 445            {
 10446                Logger.LogError(
 10447                    ex,
 10448                    "Pub/Sub background handler failed for already-ACKed message {MessageId} on {SubscriptionId}.",
 10449                    message.MessageId,
 10450                    _subscriptionId);
 8451                await NotifyBackgroundFailureAsync(
 8452                    message,
 8453                    ex,
 8454                    _subscriptionId,
 8455                    _role).ConfigureAwait(false);
 456            }
 457            finally
 458            {
 40459                Interlocked.Decrement(ref _runningCount);
 460            }
 38461        }
 34462    }
 463}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Func`3<Google.Cloud.PubSub.V1.PubsubMessage,System.Threading.CancellationToken,System.Threading.Tasks.Task>,AsyncResponse.Transports.GooglePubSub.GooglePubSubAsyncResponseOptions,AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberOptions,Microsoft.Extensions.Logging.ILogger,System.String,AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberRole)
get_PendingCount()
get_RunningCount()
HandleAsync()
DisposeAsync()
RunWorkerAsync()