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

Information
Class: AsyncResponse.Transports.GooglePubSub.GooglePubSubMessageDispatcher
Assembly: AsyncResponse.Transports.GooglePubSub
File(s): /_/src/Transports/AsyncResponse.Transports.GooglePubSub/GooglePubSubMessageDispatcher.cs
Line coverage
100%
Covered lines: 105
Uncovered lines: 0
Coverable lines: 105
Total lines: 463
Line coverage: 100%
Branch coverage
100%
Covered branches: 42
Total branches: 42
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_Logger()100%11100%
Create(...)100%22100%
ValidateOptions(...)100%2222100%
DisposeAsync()100%11100%
ExecuteHandlerAsync()100%1616100%
NotifyBackgroundFailureAsync()100%22100%

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>
 43623    protected GooglePubSubMessageDispatcher(
 43624        Func<PubsubMessage, CancellationToken, Task> handler,
 43625        GooglePubSubAsyncResponseOptions transportOptions,
 43626        GooglePubSubSubscriberOptions subscriberOptions,
 43627        ILogger logger,
 43628        string subscriptionId,
 43629        GooglePubSubSubscriberRole role)
 30    {
 43631        _handler = handler;
 43632        _transportOptions = transportOptions;
 43633        _subscriberOptions = subscriberOptions;
 43634        Logger = logger;
 43635        _subscriptionId = subscriptionId;
 43636        _role = role;
 43637    }
 38
 20039    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    {
 43650        ValidateOptions(transportOptions, subscriberOptions, role);
 51
 43652        return subscriberOptions.AckMode == GooglePubSubAckMode.AckAfterHandlerCompletes
 43653            ? new AwaitingGooglePubSubMessageDispatcher(
 43654                handler,
 43655                transportOptions,
 43656                subscriberOptions,
 43657                logger,
 43658                subscriptionId,
 43659                role)
 43660            : new QueuedGooglePubSubMessageDispatcher(
 43661                handler,
 43662                transportOptions,
 43663                subscriberOptions,
 43664                logger,
 43665                subscriptionId,
 43666                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    {
 92275        var optionPath = role is GooglePubSubSubscriberRole.Worker
 92276            ? $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.WorkerSubscriber)}"
 92277            : $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.ResponseSubscriber)}
 78
 92279        GooglePubSubOptionsValidator.ValidateTimeouts(transportOptions);
 91880        GooglePubSubOptionsValidator.ValidateStreamingPull(subscriberOptions, optionPath);
 81
 88482        if (!string.IsNullOrWhiteSpace(transportOptions.WorkerSubscriptionId)
 88483            && !string.IsNullOrWhiteSpace(transportOptions.ResponseSubscriptionId)
 88484            && StringComparer.Ordinal.Equals(transportOptions.WorkerSubscriptionId, transportOptions.ResponseSubscriptio
 85        {
 286            throw new InvalidOperationException(
 287                $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.WorkerSubscription
 288                $"{nameof(GooglePubSubAsyncResponseOptions.ResponseSubscriptionId)} must be distinct so worker and respo
 89        }
 90
 88291        if (!string.IsNullOrWhiteSpace(transportOptions.WorkerTopicId)
 88292            && !string.IsNullOrWhiteSpace(transportOptions.ResponseTopicId)
 88293            && StringComparer.Ordinal.Equals(transportOptions.WorkerTopicId, transportOptions.ResponseTopicId))
 94        {
 295            throw new InvalidOperationException(
 296                $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.WorkerTopicId)} an
 297                $"{nameof(GooglePubSubAsyncResponseOptions.ResponseTopicId)} must be distinct so worker jobs and respons
 98        }
 99
 880100        switch (subscriberOptions.AckMode)
 101        {
 102            case GooglePubSubAckMode.AckAfterHandlerCompletes:
 818103                return;
 104
 105            case GooglePubSubAckMode.AckAfterEnqueue:
 60106                if (subscriberOptions.BackgroundWorkerCount <= 0)
 107                {
 4108                    throw new InvalidOperationException(
 4109                        $"{optionPath}.{nameof(GooglePubSubSubscriberOptions.BackgroundWorkerCount)} must be explicitly 
 4110                        $"when {nameof(GooglePubSubSubscriberOptions.AckMode)} is {nameof(GooglePubSubAckMode.AckAfterEn
 111                }
 112
 56113                if (subscriberOptions.BackgroundQueueCapacity <= 0)
 114                {
 2115                    throw new InvalidOperationException(
 2116                        $"{optionPath}.{nameof(GooglePubSubSubscriberOptions.BackgroundQueueCapacity)} must be explicitl
 2117                        $"when {nameof(GooglePubSubSubscriberOptions.AckMode)} is {nameof(GooglePubSubAckMode.AckAfterEn
 118                }
 119
 54120                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.
 52124                ShutdownBudgetValidator.Validate(
 52125                    "Pub/Sub",
 52126                    $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.HostShutdownTi
 52127                    transportOptions.HostShutdownTimeout,
 52128                    ($"{optionPath}.{nameof(GooglePubSubSubscriberOptions.BackgroundDrainTimeout)}", subscriberOptions.B
 52129                    ($"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.ShutdownTimeo
 130
 48131                return;
 132
 133            default:
 2134                throw new InvalidOperationException(
 2135                    $"{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>
 404145    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    {
 458153        using var activity = AsyncResponseDiagnostics.StartActivity(
 458154            "asyncresponse.pubsub.receive",
 458155            ActivityKind.Consumer);
 458156        activity?.SetTag("asyncresponse.transport", "google_pubsub");
 458157        activity?.SetTag("asyncresponse.pubsub.role", _role.ToString());
 458158        activity?.SetTag("asyncresponse.pubsub.ack_mode", _subscriberOptions.AckMode.ToString());
 458159        activity?.SetTag("messaging.system", "gcp_pubsub");
 458160        activity?.SetTag("messaging.destination.name", _subscriptionId);
 458161        activity?.SetTag("messaging.message.id", message.MessageId);
 162
 458163        if (message.Attributes.TryGetValue(_transportOptions.CorrelationIdAttribute, out var correlationId))
 101164            AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId);
 165
 166        try
 167        {
 458168            await _handler(message, cancellationToken).ConfigureAwait(false);
 440169        }
 18170        catch (Exception ex)
 171        {
 18172            if (logFailures)
 8173                Logger.LogError(ex, "Pub/Sub message handling failed for message {MessageId}.", message.MessageId);
 18174            AsyncResponseDiagnostics.SetError(activity, ex);
 18175            throw;
 176        }
 440177    }
 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    {
 10186        var callback = _subscriberOptions.OnBackgroundFailure;
 10187        if (callback is null)
 2188            return;
 189
 190        try
 191        {
 8192            await callback(new GooglePubSubBackgroundFailureContext(
 8193                subscriptionId,
 8194                role.ToString(),
 8195                message,
 8196                exception)).ConfigureAwait(false);
 6197        }
 2198        catch (Exception callbackException)
 199        {
 2200            Logger.LogError(
 2201                callbackException,
 2202                "Pub/Sub background failure callback failed for already-ACKed message {MessageId} on {SubscriptionId}.",
 2203                message.MessageId,
 2204                subscriptionId);
 2205        }
 10206    }
 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;
 248    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)
 264        : base(handler, transportOptions, subscriberOptions, logger, subscriptionId, role)
 265    {
 266        _drainTimeout = subscriberOptions.BackgroundDrainTimeout;
 267        _subscriptionId = subscriptionId;
 268        _role = role;
 269        _queue = Channel.CreateBounded<PubsubMessage>(new BoundedChannelOptions(subscriberOptions.BackgroundQueueCapacit
 270        {
 271            AllowSynchronousContinuations = false,
 272            // Wait powers the queue-full backpressure path in HandleAsync: WriteAsync parks the
 273            // subscriber callback until a worker frees a slot instead of dropping or NACKing.
 274            FullMode = BoundedChannelFullMode.Wait,
 275            SingleReader = subscriberOptions.BackgroundWorkerCount == 1,
 276            SingleWriter = false
 277        });
 278
 279        _workers = Enumerable.Range(0, subscriberOptions.BackgroundWorkerCount)
 280            .Select(workerIndex => Task.Run(() => RunWorkerAsync(workerIndex)))
 281            .ToArray();
 282
 283        Logger.LogInformation(
 284            "Created Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId} with {WorkerCount} worker(s), queue capac
 285            _subscriptionId,
 286            subscriberOptions.BackgroundWorkerCount,
 287            subscriberOptions.BackgroundQueueCapacity,
 288            _drainTimeout);
 289    }
 290
 291    internal int PendingCount => Volatile.Read(ref _pendingCount);
 292    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        {
 301            Interlocked.Increment(ref _pendingCount);
 302            if (_queue.Writer.TryWrite(message))
 303            {
 304                Logger.LogDebug(
 305                    "Enqueued Pub/Sub message {MessageId} for background handling on {SubscriptionId}. Pending={PendingC
 306                    message.MessageId,
 307                    _subscriptionId,
 308                    PendingCount,
 309                    RunningCount);
 310                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.
 318            Logger.LogDebug(
 319                "Pub/Sub background queue is full for {SubscriptionId}; waiting for capacity before ACKing message {Mess
 320                _subscriptionId,
 321                message.MessageId,
 322                PendingCount,
 323                RunningCount);
 324            await _queue.Writer.WriteAsync(message, subscriberCancellationToken).ConfigureAwait(false);
 325            return SubscriberClient.Reply.Ack;
 326        }
 327        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.
 331            Interlocked.Decrement(ref _pendingCount);
 332            if (ex is OperationCanceledException or ChannelClosedException)
 333            {
 334                Logger.LogDebug(
 335                    "Pub/Sub message {MessageId} for {SubscriptionId} could not be enqueued during shutdown; returning N
 336                    message.MessageId,
 337                    _subscriptionId);
 338            }
 339            else
 340            {
 341                Logger.LogError(ex, "Failed to enqueue Pub/Sub message {MessageId} for {SubscriptionId}; returning NACK.
 342            }
 343
 344            return SubscriberClient.Reply.Nack;
 345        }
 346    }
 347
 348    /// <summary>Releases resources held by this instance.</summary>
 349    public override async ValueTask DisposeAsync()
 350    {
 351        if (Interlocked.Exchange(ref _disposeStarted, 1) != 0)
 352            return;
 353
 354        Logger.LogInformation(
 355            "Draining Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId}. Pending={PendingCount}, Running={Runnin
 356            _subscriptionId,
 357            PendingCount,
 358            RunningCount);
 359        _queue.Writer.TryComplete();
 360
 361        try
 362        {
 363            await Task.WhenAll(_workers).WaitAsync(_drainTimeout).ConfigureAwait(false);
 364            _drainCancellation.Dispose();
 365            Logger.LogInformation(
 366                "Drained Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId}. Pending={PendingCount}, Running={Run
 367                _subscriptionId,
 368                PendingCount,
 369                RunningCount);
 370        }
 371        catch (TimeoutException ex)
 372        {
 373            _drainCancellation.Cancel();
 374            Logger.LogWarning(
 375                ex,
 376                "Timed out while draining Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId}. Pending={PendingCou
 377                _subscriptionId,
 378                PendingCount,
 379                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.
 384            _ = Task.WhenAll(_workers).ContinueWith(
 385                _ => _drainCancellation.Dispose(),
 386                CancellationToken.None,
 387                TaskContinuationOptions.ExecuteSynchronously,
 388                TaskScheduler.Default);
 389        }
 390        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.
 395            Logger.LogDebug(ex, "Pub/Sub ACK-after-enqueue dispatcher drain for {SubscriptionId} ended with an error.", 
 396            _drainCancellation.Dispose();
 397        }
 398    }
 399
 400    private async Task RunWorkerAsync(int workerIndex)
 401    {
 402        await foreach (var message in _queue.Reader.ReadAllAsync().ConfigureAwait(false))
 403        {
 404            Interlocked.Decrement(ref _pendingCount);
 405            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.
 412            if (_drainCancellation.IsCancellationRequested)
 413            {
 414                Logger.LogWarning(
 415                    "Pub/Sub background handler for already-ACKed message {MessageId} on {SubscriptionId} was not starte
 416                    message.MessageId,
 417                    _subscriptionId);
 418
 419                await NotifyBackgroundFailureAsync(
 420                    message,
 421                    new OperationCanceledException(
 422                        "The ACK-after-enqueue drain budget lapsed before this already-ACKed message was handled."),
 423                    _subscriptionId,
 424                    _role).ConfigureAwait(false);
 425
 426                Interlocked.Decrement(ref _runningCount);
 427                continue;
 428            }
 429
 430            try
 431            {
 432                Logger.LogDebug(
 433                    "Pub/Sub background worker {WorkerIndex} handling message {MessageId} for {SubscriptionId}. Pending=
 434                    workerIndex,
 435                    message.MessageId,
 436                    _subscriptionId,
 437                    PendingCount,
 438                    RunningCount);
 439                await ExecuteHandlerAsync(
 440                    message,
 441                    _drainCancellation.Token,
 442                    logFailures: false).ConfigureAwait(false);
 443            }
 444            catch (Exception ex)
 445            {
 446                Logger.LogError(
 447                    ex,
 448                    "Pub/Sub background handler failed for already-ACKed message {MessageId} on {SubscriptionId}.",
 449                    message.MessageId,
 450                    _subscriptionId);
 451                await NotifyBackgroundFailureAsync(
 452                    message,
 453                    ex,
 454                    _subscriptionId,
 455                    _role).ConfigureAwait(false);
 456            }
 457            finally
 458            {
 459                Interlocked.Decrement(ref _runningCount);
 460            }
 461        }
 462    }
 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_Logger()
Create(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)
ValidateOptions(AsyncResponse.Transports.GooglePubSub.GooglePubSubAsyncResponseOptions,AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberOptions,AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberRole)
DisposeAsync()
ExecuteHandlerAsync()
NotifyBackgroundFailureAsync()