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

Information
Class: AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberService
Assembly: AsyncResponse.Transports.GooglePubSub
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.GooglePubSub/GooglePubSubSubscriberServices.cs
Line coverage
98%
Covered lines: 64
Uncovered lines: 1
Coverable lines: 65
Total lines: 232
Line coverage: 98.4%
Branch coverage
75%
Covered branches: 3
Total branches: 4
Branch coverage: 75%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%22100%
.ctor(...)100%11100%
ExecuteAsync()50%2296.55%
RunSubscriberAsync()100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.GooglePubSub/GooglePubSubSubscriberServices.cs

#LineLine coverage
 1using Google.Api.Gax;
 2using Google.Cloud.PubSub.V1;
 3using Microsoft.Extensions.Hosting;
 4using Microsoft.Extensions.Logging;
 5using Microsoft.Extensions.Options;
 6
 7namespace AsyncResponse.Transports.GooglePubSub;
 8
 9internal abstract class GooglePubSubSubscriberService : BackgroundService
 10{
 11    private readonly Func<SubscriptionName, GooglePubSubSubscriberOptions, Task<IGooglePubSubSubscriberClient>> _subscri
 12
 13    /// <summary>Runs the GooglePubSubSubscriberService operation.</summary>
 14    protected GooglePubSubSubscriberService(
 15        IOptions<GooglePubSubAsyncResponseOptions> options,
 16        ILogger logger)
 317        : this(options, logger, CreateSubscriberAsync)
 18    {
 319    }
 20
 21    /// <summary>Runs the GooglePubSubSubscriberService operation.</summary>
 322    protected GooglePubSubSubscriberService(
 323        IOptions<GooglePubSubAsyncResponseOptions> options,
 324        ILogger logger,
 325        Func<SubscriptionName, GooglePubSubSubscriberOptions, Task<IGooglePubSubSubscriberClient>> subscriberFactory)
 26    {
 327        Options = options.Value;
 328        Logger = logger;
 329        _subscriberFactory = subscriberFactory;
 330    }
 31
 32    protected GooglePubSubAsyncResponseOptions Options { get; }
 33    protected ILogger Logger { get; }
 34
 35    protected abstract string SubscriptionId { get; }
 36    protected abstract GooglePubSubSubscriberOptions SubscriberOptions { get; }
 37    protected abstract GooglePubSubSubscriberRole SubscriberRole { get; }
 38    /// <summary>Handles the delivered message.</summary>
 39    protected abstract Task HandleMessageAsync(PubsubMessage message, CancellationToken cancellationToken);
 40
 41    [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
 42    private static async Task<IGooglePubSubSubscriberClient> CreateSubscriberAsync(
 43        SubscriptionName subscriptionName,
 44        GooglePubSubSubscriberOptions subscriberOptions)
 45    {
 46        // EmulatorOrProduction honors PUBSUB_EMULATOR_HOST when present (local dev / tests) and uses
 47        // real Google Cloud otherwise — no behavior change in production.
 48        var builder = new SubscriberClientBuilder
 49        {
 50            SubscriptionName = subscriptionName,
 51            EmulatorDetection = EmulatorDetection.EmulatorOrProduction
 52        };
 53
 54        // In early-ACK mode, bound the streaming pull to the background queue capacity so the client
 55        // never holds more un-ACKed messages than the dispatcher can accept. Combined with the
 56        // dispatcher's write-side backpressure this keeps queue-full NACKs (which burn a configured
 57        // DeadLetterPolicy's delivery attempts) out of steady-state operation.
 58        if (subscriberOptions.AckMode is GooglePubSubAckMode.AckAfterEnqueue)
 59        {
 60            builder.Settings = new SubscriberClient.Settings
 61            {
 62                FlowControlSettings = new Google.Api.Gax.FlowControlSettings(
 63                    maxOutstandingElementCount: subscriberOptions.BackgroundQueueCapacity,
 64                    maxOutstandingByteCount: null)
 65            };
 66        }
 67
 68        var subscriber = await builder.BuildAsync().ConfigureAwait(false);
 69        return new GooglePubSubSubscriberClientAdapter(subscriber);
 70    }
 71
 72    /// <summary>Runs this background operation until cancellation is requested.</summary>
 73    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 74    {
 375        var projectId = GooglePubSubOptionsValidator.Required(Options.ProjectId, nameof(Options.ProjectId));
 376        var subscriptionId = SubscriptionId;
 377        GooglePubSubMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 378        var subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);
 79
 80        // The transport intentionally has no MaxDeliveryAttempts and no library-managed dead-letter
 81        // queue for Pub/Sub: capping redelivery is delegated to the subscription's native
 82        // DeadLetterPolicy. The client cannot cheaply probe whether one is configured, so tell the
 83        // operator unconditionally instead of failing silently forever on a poison message.
 384        Logger.LogWarning(
 385            "Pub/Sub redelivery is unbounded for subscription {Subscription} ({Role}): the transport enforces no deliver
 386            + "Configure a DeadLetterPolicy on the subscription to cap redeliveries of failing messages.",
 387            subscriptionName.ToString(),
 388            SubscriberRole);
 89
 390        var failures = 0;
 391        while (!stoppingToken.IsCancellationRequested)
 92        {
 93            try
 94            {
 395                await RunSubscriberAsync(subscriptionName, subscriptionId, stoppingToken).ConfigureAwait(false);
 396                return;
 97            }
 298            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 99            {
 0100                return;
 101            }
 2102            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 103            {
 2104                failures++;
 2105                var retryDelay = AsyncResponseRetry.Backoff(
 2106                    failures,
 2107                    Options.SubscriberRetryBaseDelay,
 2108                    Options.SubscriberRetryMaxDelay);
 2109                Logger.LogWarning(
 2110                    ex,
 2111                    "Pub/Sub subscriber failed for subscription {Subscription} ({Role}); retrying in {RetryDelay}.",
 2112                    subscriptionName.ToString(),
 2113                    SubscriberRole,
 2114                    retryDelay);
 2115                await Task.Delay(retryDelay, stoppingToken).ConfigureAwait(false);
 116            }
 117        }
 3118    }
 119
 120    private async Task RunSubscriberAsync(
 121        SubscriptionName subscriptionName,
 122        string subscriptionId,
 123        CancellationToken stoppingToken)
 124    {
 3125        var subscriber = await _subscriberFactory(subscriptionName, SubscriberOptions).ConfigureAwait(false);
 3126        await using var dispatcher = GooglePubSubMessageDispatcher.Create(
 3127            HandleMessageAsync,
 3128            Options,
 3129            SubscriberOptions,
 3130            Logger,
 3131            subscriptionId,
 3132            SubscriberRole);
 133
 3134        Logger.LogInformation(
 3135            "Pub/Sub subscriber started. Subscription: {Subscription}. Role: {Role}. AckMode: {AckMode}.",
 3136            subscriptionName.ToString(),
 3137            SubscriberRole,
 3138            SubscriberOptions.AckMode);
 139
 3140        var runTask = subscriber.StartAsync(dispatcher.HandleAsync);
 141
 142        try
 143        {
 3144            await runTask.WaitAsync(stoppingToken).ConfigureAwait(false);
 3145        }
 3146        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 147        {
 3148            await subscriber.StopAsync(
 3149                new SubscriberClient.ShutdownOptions
 3150                {
 3151                    Timeout = Options.ShutdownTimeout
 3152                },
 3153                CancellationToken.None).ConfigureAwait(false);
 3154            await runTask.ConfigureAwait(false);
 3155        }
 3156    }
 157
 158}
 159
 160internal sealed class GooglePubSubWorkerSubscriber : GooglePubSubSubscriberService
 161{
 162    private readonly IAsyncResponseIngress _ingress;
 163
 164    /// <summary>Runs the GooglePubSubWorkerSubscriber operation.</summary>
 165    public GooglePubSubWorkerSubscriber(
 166        IOptions<GooglePubSubAsyncResponseOptions> options,
 167        IAsyncResponseIngress ingress,
 168        ILogger<GooglePubSubWorkerSubscriber> logger)
 169        : base(options, logger)
 170    {
 171        _ingress = ingress;
 172    }
 173
 174    internal GooglePubSubWorkerSubscriber(
 175        IOptions<GooglePubSubAsyncResponseOptions> options,
 176        IAsyncResponseIngress ingress,
 177        ILogger<GooglePubSubWorkerSubscriber> logger,
 178        Func<SubscriptionName, GooglePubSubSubscriberOptions, Task<IGooglePubSubSubscriberClient>> subscriberFactory)
 179        : base(options, logger, subscriberFactory)
 180    {
 181        _ingress = ingress;
 182    }
 183
 184    protected override string SubscriptionId
 185        => GooglePubSubOptionsValidator.Required(Options.WorkerSubscriptionId, nameof(Options.WorkerSubscriptionId));
 186
 187    protected override GooglePubSubSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 188    protected override GooglePubSubSubscriberRole SubscriberRole => GooglePubSubSubscriberRole.Worker;
 189
 190    /// <summary>Handles the delivered message.</summary>
 191    protected override Task HandleMessageAsync(PubsubMessage message, CancellationToken cancellationToken)
 192        => _ingress.HandleWorkerMessageAsync(message.Data.ToStringUtf8());
 193}
 194
 195internal sealed class GooglePubSubResponseIngressSubscriber : GooglePubSubSubscriberService
 196{
 197    private readonly IAsyncResponseIngress _ingress;
 198
 199    /// <summary>Runs the GooglePubSubResponseIngressSubscriber operation.</summary>
 200    public GooglePubSubResponseIngressSubscriber(
 201        IOptions<GooglePubSubAsyncResponseOptions> options,
 202        IAsyncResponseIngress ingress,
 203        ILogger<GooglePubSubResponseIngressSubscriber> logger)
 204        : base(options, logger)
 205    {
 206        _ingress = ingress;
 207    }
 208
 209    internal GooglePubSubResponseIngressSubscriber(
 210        IOptions<GooglePubSubAsyncResponseOptions> options,
 211        IAsyncResponseIngress ingress,
 212        ILogger<GooglePubSubResponseIngressSubscriber> logger,
 213        Func<SubscriptionName, GooglePubSubSubscriberOptions, Task<IGooglePubSubSubscriberClient>> subscriberFactory)
 214        : base(options, logger, subscriberFactory)
 215    {
 216        _ingress = ingress;
 217    }
 218
 219    protected override string SubscriptionId
 220        => GooglePubSubOptionsValidator.Required(Options.ResponseSubscriptionId, nameof(Options.ResponseSubscriptionId))
 221
 222    protected override GooglePubSubSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 223    protected override GooglePubSubSubscriberRole SubscriberRole => GooglePubSubSubscriberRole.ResponseIngress;
 224
 225    /// <summary>Handles the delivered message.</summary>
 226    protected override Task HandleMessageAsync(PubsubMessage message, CancellationToken cancellationToken)
 227    {
 228        var messageJson = message.Data.ToStringUtf8();
 229        var correlationId = GooglePubSubCorrelationIdExtractor.Extract(message, messageJson, Options);
 230        return _ingress.HandleResponseMessageAsync(messageJson, correlationId);
 231    }
 232}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.GooglePubSub.GooglePubSubAsyncResponseOptions>, Microsoft.Extensions.Logging.ILogger)
.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.GooglePubSub.GooglePubSubAsyncResponseOptions>, Microsoft.Extensions.Logging.ILogger, System.Func<Google.Cloud.PubSub.V1.SubscriptionName, AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberOptions, System.Threading.Tasks.Task<AsyncResponse.Transports.GooglePubSub.IGooglePubSubSubscriberClient>>)
ExecuteAsync()
RunSubscriberAsync()