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

Information
Class: AsyncResponse.Transports.GooglePubSub.GooglePubSubMessageDispatcher
Assembly: AsyncResponse.Transports.GooglePubSub
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.GooglePubSub/GooglePubSubMessageDispatcher.cs
Line coverage
100%
Covered lines: 107
Uncovered lines: 0
Coverable lines: 107
Total lines: 430
Line coverage: 100%
Branch coverage
100%
Covered branches: 46
Total branches: 46
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%
Create(...)100%22100%
ValidateOptions(...)100%2626100%
DisposeAsync()100%11100%
ExecuteHandlerAsync()100%1616100%
NotifyBackgroundFailureAsync()100%22100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/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>
 323    protected GooglePubSubMessageDispatcher(
 324        Func<PubsubMessage, CancellationToken, Task> handler,
 325        GooglePubSubAsyncResponseOptions transportOptions,
 326        GooglePubSubSubscriberOptions subscriberOptions,
 327        ILogger logger,
 328        string subscriptionId,
 329        GooglePubSubSubscriberRole role)
 30    {
 331        _handler = handler;
 332        _transportOptions = transportOptions;
 333        _subscriberOptions = subscriberOptions;
 334        Logger = logger;
 335        _subscriptionId = subscriptionId;
 336        _role = role;
 337    }
 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    {
 350        ValidateOptions(transportOptions, subscriberOptions, role);
 51
 352        return subscriberOptions.AckMode == GooglePubSubAckMode.AckAfterHandlerCompletes
 353            ? new AwaitingGooglePubSubMessageDispatcher(
 354                handler,
 355                transportOptions,
 356                subscriberOptions,
 357                logger,
 358                subscriptionId,
 359                role)
 360            : new QueuedGooglePubSubMessageDispatcher(
 361                handler,
 362                transportOptions,
 363                subscriberOptions,
 364                logger,
 365                subscriptionId,
 366                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    {
 375        var optionPath = role is GooglePubSubSubscriberRole.Worker
 376            ? $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.WorkerSubscriber)}"
 377            : $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.ResponseSubscriber)}
 78
 379        if (!string.IsNullOrWhiteSpace(transportOptions.WorkerSubscriptionId)
 380            && !string.IsNullOrWhiteSpace(transportOptions.ResponseSubscriptionId)
 381            && StringComparer.Ordinal.Equals(transportOptions.WorkerSubscriptionId, transportOptions.ResponseSubscriptio
 82        {
 383            throw new InvalidOperationException(
 384                $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.WorkerSubscription
 385                $"{nameof(GooglePubSubAsyncResponseOptions.ResponseSubscriptionId)} must be distinct so worker and respo
 86        }
 87
 388        if (!string.IsNullOrWhiteSpace(transportOptions.WorkerTopicId)
 389            && !string.IsNullOrWhiteSpace(transportOptions.ResponseTopicId)
 390            && StringComparer.Ordinal.Equals(transportOptions.WorkerTopicId, transportOptions.ResponseTopicId))
 91        {
 392            throw new InvalidOperationException(
 393                $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.WorkerTopicId)} an
 394                $"{nameof(GooglePubSubAsyncResponseOptions.ResponseTopicId)} must be distinct so worker jobs and respons
 95        }
 96
 397        switch (subscriberOptions.AckMode)
 98        {
 99            case GooglePubSubAckMode.AckAfterHandlerCompletes:
 3100                return;
 101
 102            case GooglePubSubAckMode.AckAfterEnqueue:
 3103                if (subscriberOptions.BackgroundWorkerCount <= 0)
 104                {
 3105                    throw new InvalidOperationException(
 3106                        $"{optionPath}.{nameof(GooglePubSubSubscriberOptions.BackgroundWorkerCount)} must be explicitly 
 3107                        $"when {nameof(GooglePubSubSubscriberOptions.AckMode)} is {nameof(GooglePubSubAckMode.AckAfterEn
 108                }
 109
 3110                if (subscriberOptions.BackgroundQueueCapacity <= 0)
 111                {
 3112                    throw new InvalidOperationException(
 3113                        $"{optionPath}.{nameof(GooglePubSubSubscriberOptions.BackgroundQueueCapacity)} must be explicitl
 3114                        $"when {nameof(GooglePubSubSubscriberOptions.AckMode)} is {nameof(GooglePubSubAckMode.AckAfterEn
 115                }
 116
 3117                if (subscriberOptions.BackgroundDrainTimeout <= TimeSpan.Zero)
 118                {
 3119                    throw new InvalidOperationException(
 3120                        $"{optionPath}.{nameof(GooglePubSubSubscriberOptions.BackgroundDrainTimeout)} must be positive."
 121                }
 122
 3123                if (transportOptions.ShutdownTimeout <= TimeSpan.Zero)
 124                {
 3125                    throw new InvalidOperationException(
 3126                        $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.ShutdownTi
 127                }
 128
 129                // Pub/Sub spends the background drain plus the bounded subscriber-client stop
 130                // (ShutdownTimeout) at shutdown; both must fit inside the host budget.
 3131                ShutdownBudgetValidator.Validate(
 3132                    "Pub/Sub",
 3133                    $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.HostShutdownTi
 3134                    transportOptions.HostShutdownTimeout,
 3135                    ($"{optionPath}.{nameof(GooglePubSubSubscriberOptions.BackgroundDrainTimeout)}", subscriberOptions.B
 3136                    ($"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.ShutdownTimeo
 137
 3138                return;
 139
 140            default:
 3141                throw new InvalidOperationException(
 3142                    $"{optionPath}.{nameof(GooglePubSubSubscriberOptions.AckMode)} has unsupported value '{subscriberOpt
 143        }
 144    }
 145
 146    /// <summary>Handles the delivered message.</summary>
 147    public abstract Task<SubscriberClient.Reply> HandleAsync(
 148        PubsubMessage message,
 149        CancellationToken subscriberCancellationToken);
 150
 151    /// <summary>Releases resources held by this instance.</summary>
 3152    public virtual ValueTask DisposeAsync() => ValueTask.CompletedTask;
 153
 154    /// <summary>Runs the ExecuteHandlerAsync operation.</summary>
 155    protected async Task ExecuteHandlerAsync(
 156        PubsubMessage message,
 157        CancellationToken cancellationToken,
 158        bool logFailures = true)
 159    {
 3160        using var activity = AsyncResponseDiagnostics.StartActivity(
 3161            "asyncresponse.pubsub.receive",
 3162            ActivityKind.Consumer);
 3163        activity?.SetTag("asyncresponse.transport", "google_pubsub");
 3164        activity?.SetTag("asyncresponse.pubsub.role", _role.ToString());
 3165        activity?.SetTag("asyncresponse.pubsub.ack_mode", _subscriberOptions.AckMode.ToString());
 3166        activity?.SetTag("messaging.system", "gcp_pubsub");
 3167        activity?.SetTag("messaging.destination.name", _subscriptionId);
 3168        activity?.SetTag("messaging.message.id", message.MessageId);
 169
 3170        if (message.Attributes.TryGetValue(_transportOptions.CorrelationIdAttribute, out var correlationId))
 3171            AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId);
 172
 173        try
 174        {
 3175            await _handler(message, cancellationToken).ConfigureAwait(false);
 3176        }
 2177        catch (Exception ex)
 178        {
 2179            if (logFailures)
 2180                Logger.LogError(ex, "Pub/Sub message handling failed for message {MessageId}.", message.MessageId);
 2181            AsyncResponseDiagnostics.SetError(activity, ex);
 3182            throw;
 183        }
 3184    }
 185
 186    /// <summary>Runs the NotifyBackgroundFailureAsync operation.</summary>
 187    protected async ValueTask NotifyBackgroundFailureAsync(
 188        PubsubMessage message,
 189        Exception exception,
 190        string subscriptionId,
 191        GooglePubSubSubscriberRole role)
 192    {
 2193        var callback = _subscriberOptions.OnBackgroundFailure;
 2194        if (callback is null)
 2195            return;
 196
 197        try
 198        {
 2199            await callback(new GooglePubSubBackgroundFailureContext(
 2200                subscriptionId,
 2201                role.ToString(),
 2202                message,
 2203                exception)).ConfigureAwait(false);
 2204        }
 2205        catch (Exception callbackException)
 206        {
 2207            Logger.LogError(
 2208                callbackException,
 2209                "Pub/Sub background failure callback failed for already-ACKed message {MessageId} on {SubscriptionId}.",
 2210                message.MessageId,
 2211                subscriptionId);
 2212        }
 2213    }
 214}
 215
 216internal sealed class AwaitingGooglePubSubMessageDispatcher(
 217    Func<PubsubMessage, CancellationToken, Task> handler,
 218    GooglePubSubAsyncResponseOptions transportOptions,
 219    GooglePubSubSubscriberOptions subscriberOptions,
 220    ILogger logger,
 221    string subscriptionId,
 222    GooglePubSubSubscriberRole role)
 223    : GooglePubSubMessageDispatcher(handler, transportOptions, subscriberOptions, logger, subscriptionId, role)
 224{
 225    /// <summary>Handles the delivered message.</summary>
 226    public override async Task<SubscriberClient.Reply> HandleAsync(
 227        PubsubMessage message,
 228        CancellationToken subscriberCancellationToken)
 229    {
 230        try
 231        {
 232            await ExecuteHandlerAsync(message, subscriberCancellationToken).ConfigureAwait(false);
 233            return SubscriberClient.Reply.Ack;
 234        }
 235        catch
 236        {
 237            return SubscriberClient.Reply.Nack;
 238        }
 239    }
 240}
 241
 242internal sealed class QueuedGooglePubSubMessageDispatcher : GooglePubSubMessageDispatcher
 243{
 244    private readonly Channel<PubsubMessage> _queue;
 245    private readonly Task[] _workers;
 246    private readonly CancellationTokenSource _drainCancellation = new();
 247    private readonly TimeSpan _drainTimeout;
 248    private readonly string _subscriptionId;
 249    private readonly GooglePubSubSubscriberRole _role;
 250    private int _pendingCount;
 251    private int _runningCount;
 252    private int _disposeStarted;
 253
 254    /// <summary>Runs the QueuedGooglePubSubMessageDispatcher operation.</summary>
 255    public QueuedGooglePubSubMessageDispatcher(
 256        Func<PubsubMessage, CancellationToken, Task> handler,
 257        GooglePubSubAsyncResponseOptions transportOptions,
 258        GooglePubSubSubscriberOptions subscriberOptions,
 259        ILogger logger,
 260        string subscriptionId,
 261        GooglePubSubSubscriberRole role)
 262        : base(handler, transportOptions, subscriberOptions, logger, subscriptionId, role)
 263    {
 264        _drainTimeout = subscriberOptions.BackgroundDrainTimeout;
 265        _subscriptionId = subscriptionId;
 266        _role = role;
 267        _queue = Channel.CreateBounded<PubsubMessage>(new BoundedChannelOptions(subscriberOptions.BackgroundQueueCapacit
 268        {
 269            AllowSynchronousContinuations = false,
 270            // Wait powers the queue-full backpressure path in HandleAsync: WriteAsync parks the
 271            // subscriber callback until a worker frees a slot instead of dropping or NACKing.
 272            FullMode = BoundedChannelFullMode.Wait,
 273            SingleReader = subscriberOptions.BackgroundWorkerCount == 1,
 274            SingleWriter = false
 275        });
 276
 277        _workers = Enumerable.Range(0, subscriberOptions.BackgroundWorkerCount)
 278            .Select(workerIndex => Task.Run(() => RunWorkerAsync(workerIndex)))
 279            .ToArray();
 280
 281        Logger.LogInformation(
 282            "Created Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId} with {WorkerCount} worker(s), queue capac
 283            _subscriptionId,
 284            subscriberOptions.BackgroundWorkerCount,
 285            subscriberOptions.BackgroundQueueCapacity,
 286            _drainTimeout);
 287    }
 288
 289    internal int PendingCount => Volatile.Read(ref _pendingCount);
 290    internal int RunningCount => Volatile.Read(ref _runningCount);
 291
 292    /// <summary>Handles the delivered message.</summary>
 293    public override async Task<SubscriberClient.Reply> HandleAsync(
 294        PubsubMessage message,
 295        CancellationToken subscriberCancellationToken)
 296    {
 297        try
 298        {
 299            Interlocked.Increment(ref _pendingCount);
 300            if (_queue.Writer.TryWrite(message))
 301            {
 302                Logger.LogDebug(
 303                    "Enqueued Pub/Sub message {MessageId} for background handling on {SubscriptionId}. Pending={PendingC
 304                    message.MessageId,
 305                    _subscriptionId,
 306                    PendingCount,
 307                    RunningCount);
 308                return SubscriberClient.Reply.Ack;
 309            }
 310
 311            // Queue full: apply backpressure instead of NACKing. A NACK burns one delivery attempt of
 312            // a DeadLetterPolicy configured on the subscription, so a saturated worker pool would
 313            // dead-letter healthy, never-executed messages. The streaming pull is flow-control-bounded
 314            // to the queue capacity, so at most capacity callbacks wait here; the await completes as
 315            // soon as a background worker frees a slot.
 316            Logger.LogDebug(
 317                "Pub/Sub background queue is full for {SubscriptionId}; waiting for capacity before ACKing message {Mess
 318                _subscriptionId,
 319                message.MessageId,
 320                PendingCount,
 321                RunningCount);
 322            await _queue.Writer.WriteAsync(message, subscriberCancellationToken).ConfigureAwait(false);
 323            return SubscriberClient.Reply.Ack;
 324        }
 325        catch (Exception ex)
 326        {
 327            // Cancellation (subscriber stopping) or a completed channel (dispatcher disposing):
 328            // NACK so Pub/Sub redelivers the message to the next subscriber instance.
 329            Interlocked.Decrement(ref _pendingCount);
 330            if (ex is OperationCanceledException or ChannelClosedException)
 331            {
 332                Logger.LogDebug(
 333                    "Pub/Sub message {MessageId} for {SubscriptionId} could not be enqueued during shutdown; returning N
 334                    message.MessageId,
 335                    _subscriptionId);
 336            }
 337            else
 338            {
 339                Logger.LogError(ex, "Failed to enqueue Pub/Sub message {MessageId} for {SubscriptionId}; returning NACK.
 340            }
 341
 342            return SubscriberClient.Reply.Nack;
 343        }
 344    }
 345
 346    /// <summary>Releases resources held by this instance.</summary>
 347    public override async ValueTask DisposeAsync()
 348    {
 349        if (Interlocked.Exchange(ref _disposeStarted, 1) != 0)
 350            return;
 351
 352        Logger.LogInformation(
 353            "Draining Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId}. Pending={PendingCount}, Running={Runnin
 354            _subscriptionId,
 355            PendingCount,
 356            RunningCount);
 357        _queue.Writer.TryComplete();
 358
 359        try
 360        {
 361            await Task.WhenAll(_workers).WaitAsync(_drainTimeout).ConfigureAwait(false);
 362            _drainCancellation.Dispose();
 363            Logger.LogInformation(
 364                "Drained Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId}. Pending={PendingCount}, Running={Run
 365                _subscriptionId,
 366                PendingCount,
 367                RunningCount);
 368        }
 369        catch (TimeoutException ex)
 370        {
 371            _drainCancellation.Cancel();
 372            Logger.LogWarning(
 373                ex,
 374                "Timed out while draining Pub/Sub ACK-after-enqueue dispatcher for {SubscriptionId}. Pending={PendingCou
 375                _subscriptionId,
 376                PendingCount,
 377                RunningCount);
 378
 379            // The workers are still running and read _drainCancellation.Token each loop, so disposing
 380            // it now would throw ObjectDisposedException inside them. Dispose once they actually finish,
 381            // off the shutdown path, so the source is not leaked either.
 382            _ = Task.WhenAll(_workers).ContinueWith(
 383                _ => _drainCancellation.Dispose(),
 384                CancellationToken.None,
 385                TaskContinuationOptions.ExecuteSynchronously,
 386                TaskScheduler.Default);
 387        }
 388    }
 389
 390    private async Task RunWorkerAsync(int workerIndex)
 391    {
 392        await foreach (var message in _queue.Reader.ReadAllAsync().ConfigureAwait(false))
 393        {
 394            Interlocked.Decrement(ref _pendingCount);
 395            Interlocked.Increment(ref _runningCount);
 396
 397            try
 398            {
 399                Logger.LogDebug(
 400                    "Pub/Sub background worker {WorkerIndex} handling message {MessageId} for {SubscriptionId}. Pending=
 401                    workerIndex,
 402                    message.MessageId,
 403                    _subscriptionId,
 404                    PendingCount,
 405                    RunningCount);
 406                await ExecuteHandlerAsync(
 407                    message,
 408                    _drainCancellation.Token,
 409                    logFailures: false).ConfigureAwait(false);
 410            }
 411            catch (Exception ex)
 412            {
 413                Logger.LogError(
 414                    ex,
 415                    "Pub/Sub background handler failed for already-ACKed message {MessageId} on {SubscriptionId}.",
 416                    message.MessageId,
 417                    _subscriptionId);
 418                await NotifyBackgroundFailureAsync(
 419                    message,
 420                    ex,
 421                    _subscriptionId,
 422                    _role).ConfigureAwait(false);
 423            }
 424            finally
 425            {
 426                Interlocked.Decrement(ref _runningCount);
 427            }
 428        }
 429    }
 430}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Func<Google.Cloud.PubSub.V1.PubsubMessage, System.Threading.CancellationToken, System.Threading.Tasks.Task>, AsyncResponse.Transports.GooglePubSub.GooglePubSubAsyncResponseOptions, AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberOptions, Microsoft.Extensions.Logging.ILogger, string, AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberRole)
Create(System.Func<Google.Cloud.PubSub.V1.PubsubMessage, System.Threading.CancellationToken, System.Threading.Tasks.Task>, AsyncResponse.Transports.GooglePubSub.GooglePubSubAsyncResponseOptions, AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberOptions, Microsoft.Extensions.Logging.ILogger, string, AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberRole)
ValidateOptions(AsyncResponse.Transports.GooglePubSub.GooglePubSubAsyncResponseOptions, AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberOptions, AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberRole)
DisposeAsync()
ExecuteHandlerAsync()
NotifyBackgroundFailureAsync()